模拟RabbitMQ实现消息队列

老婆出轨 · 7 天前

代码地址： https://gitee.com/lv-shixiong/linux_test/tree/master/mq
   项目介绍

   壅闭队列实现生产者消费者模型：
   可以或许解耦合，将数据的生产和消费分开
   支持并发
   支持忙闲不均
   消息队列就是把壅闭队列封装成独立的服务器程序，让它运行到某一台主机上，发消息到该主机，从该主机拉取消息(跨主机的生产者消费者模型)
   实现消息队列服务器，发布客户端，订阅客户端

   开发环境

   ubuntu-22.04、VSCode、g++、Makefile
   技能选型

   主开发语言：C++
   序列化框架：Protobuf二进制序列化
   网络通讯：自界说应用层协议+muduo库：对tcp长连接的封装，并且使用epoll的事件驱动模式，实现高并发服务器与客户端
   源数据信息数据库：SQLite3
   单元测试框架：Gtest
   环境搭建

   1、安装wget
   2、安装protobuf
   在 /usr/local/google里找protobuf
   3、安装muduo
   安装好的程序在../build
   Protobuf

   是序列化反序列化的框架
   特点：语言无关、平台无关；高效；扩展性、兼容性好。
   使用方法：

编写.proto文件(界说结构对象)。
使用protoc编译器编译文件天生一系列接口代码、.h中界说了我们所描述的数据结构对象类，.cc界说实现告终构化对象数据的访问&操作&序列化&反序列化。
将天生的代码包含进我们的代码中，引入头文件。

语法规范：

proto文件使用全小写，单词间_隔开。
书写代码时，使用2个空格缩进。
每行语句分号末了
//注释

   语法：
   指定proto3语法，syntax = "proto3"。
   声明命名空间，package 声明符，示.proto文件的命名空间。
   界说结构对象描述message 消息体名称{}，名称用驼峰
   界说消息字段字段范例字段名 = 字段唯一编号

   编译：protoc --cpp_out=路径 proto文件名称
   protoc --cpp_out=. contacts.proto
                                 登录后复制

syntax= "proto3";//声明语法版本
package contacts;//声明代码的命名空间
message contact{
//各个字段字段类型字段名 = 字段唯一编号
uint64 sn = 1;
string name = 2;
float score = 3;
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

登录后复制

main:main.cpp contacts.pb.cc
g++ -o $@ $^ -std=c++11 -lprotobuf

复制代码

1.
2.

登录后复制

#include "contacts.pb.h"
#include <iostream>
int main()
{
//创建通讯录对象
contacts::contact con;
//添加字段
con.set_name("Nesta");
con.set_sn(13);
con.set_score(99);
//序列化
std::string str=con.SerializeAsString();
//反序列化
contacts::contact conR;
conR.ParseFromString(str);
//打印信息
std::cout<<"name="<<conR.name()<<std::endl;
std::cout<<"sn="<<conR.sn()<<std::endl;
std::cout<<"socre="<<conR.score()<<std::endl;
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.

                     Muduo

   muduo库是一个基于reactor模型的高性能服务器框架。
   什么是reactor？基于时间触发的模型(基于epoll进行IO事件监控)
   主从reactor模型：将IO事件监控进行进一步的层次分别。
   主reactor：只对新建连接事件进行监控(保证不受IO壅闭影响实现高效的新建连接获取)
   从reactor：针对新建连接进行IO事件监控(进行IO操作和业务处理)
   主从reactor必然是一个多执行流的并发模式---one thread one loop，一个事件监控，占据一个线程，进行事件监控。

   muduo::net::TcpServer类
   muduo::net::EventLoop类
   muduo::net::TcpConnection类
   muduo::net::Buffer类
   Server：
                                 登录后复制

#include "./include/muduo/net/TcpServer.h"
#include "./include/muduo/net/EventLoop.h"
#include "./include/muduo/net/TcpConnection.h"
#include <iostream>
#include <string>
#include <functional>
#include <unordered_map>
class TranslateServer
{
public:
TranslateServer(uint16_t port)
:_server(&_loop
,muduo::net::InetAddress("0.0.0.0",8888)
,"translateServer"
,muduo::net::TcpServer::kNoReusePort)
{
//注册回调函数
auto ccb=std::bind(&TranslateServer::OnConnection,this,std::placeholders::_1);
auto mcb=std::bind(&TranslateServer::OnMessage,this,std::placeholders::_1
,std::placeholders::_2,std::placeholders::_3);
_server.setConnectionCallback(ccb);
_server.setMessageCallback(mcb);
_server.setConnectionCallback(ccb);
_server.setMessageCallback(mcb);
}
void Start()//启动服务器
{
_server.start();
_loop.loop();//这是一个死循环
}
private:
void OnConnection(const muduo::net::TcpConnectionPtr& conn)
{
//对连接的回调
//建立连接和断开连接时都会触发
if(conn->connected())//确定是否连接成功
{
std::cout<<"连接成功"<<std::endl;
}
else
{
std::cout<<"连接失败"<<std::endl;
}
}
std::string Translate(std::string key)
{
std::unordered_map<std::string,std::string> dict={
{"你好","hello"}
,{"hello","你好"}
,{"world","世界"}
,{"世界","world"}
};//字典
auto it=dict.find(key);
if(it!=dict.end())//找到了
{
return it->second;
}
else//没找到
{
return "不认识这个词语";
}
}
void OnMessage(const muduo::net::TcpConnectionPtr& ptr,muduo::net::Buffer* buffer,muduo::Timestamp)
{
//对消息的回调
//1.从缓冲区里拿到数据
std::string key=buffer->retrieveAllAsString();
//2.翻译
std::string val=Translate(key);
//3.将翻译结果发回客户端
muduo::net::Buffer message;
message.append(val);
ptr->send(&message);
}
//loop是对epool的事件监听，触发事件后进行IO操作
muduo::net::EventLoop _loop;
//_server主要用于设置回调函数，告诉服务器收到什么消息进行怎样的处理
muduo::net::TcpServer _server;
};
int main()
{
TranslateServer svr(8888);
svr.Start();
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.

Client
登录后复制

#include "include/muduo/net/TcpClient.h"
#include "include/muduo/net/Buffer.h"
#include "include/muduo/net/EventLoopThread.h"
#include "include/muduo/net/InetAddress.h"
#include "include/muduo/base/CountDownLatch.h"
#include <string>
#include <iostream>
#include <functional>
class TranslateClient
{
public:
TranslateClient(std::string serverIp,uint16_t serverPort)
:_client(_loopThread.startLoop()
,muduo::net::InetAddress(serverIp,serverPort)
,"translateClient")
,_latch(1)//一定要设为1
{
//注册回调函数
auto ccb=std::bind(&TranslateClient::OnConnection,this,std::placeholders::_1);
auto mcb=std::bind(&TranslateClient::OnMessage,this
,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
_client.setConnectionCallback(ccb);
_client.setMessageCallback(mcb);
std::cout<<"初始化完成"<<std::endl;
}
void Connect()//连接服务器
{
_client.connect();//进行连接
//之所以要阻塞就是为了方式没有连接成功就进行了其他操作，导致错误
std::cout<<"开始连接"<<std::endl;
_latch.wait();//阻塞等待连接成功
}
void Send(std::string& message)//发送消息
{
if(_conn->connected())//确认连接成功再进行IO
{
_conn->send(message);
}
}
private:
void OnConnection(const muduo::net::TcpConnectionPtr& conn)
{
if(conn->connected())//连接上了
{
_latch.countDown();//解除阻塞
_conn=conn;//保存连接
}
else
{
//......
}
}
void OnMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer,muduo::Timestamp ts)
{
//来消息了就打印出来
std::string message=buffer->retrieveAllAsString();
std::cout<<"Server Say> "<<message<<std::endl;
}
private:
muduo::CountDownLatch _latch;//用于同步
muduo::net::EventLoopThread _loopThread;//创建线程进行IO
muduo::net::TcpClient _client;
muduo::net::TcpConnectionPtr _conn;//客户端与服务器的连接
};
int main()
{
TranslateClient cli("127.0.0.1",8888);
cli.Connect();//建立连接
while(1)
{
std::string message;
std::cin>>message;
cli.Send(message);
}
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.

                     上面的服务器与客户端没有协议，直接通讯。
   下面是基于protobuf来进行通讯。

   muduo-master/examples/protobuf/codec/codec.cc和codec.h以及dispatcher.h
   examples是一些例子，要学会参考别人的代码。

   SQLite3

   是什么？进程内的轻量级数据库，是本地的，不必要服务器，它给我们提供了一系列接口。
   官方文档： List Of SQLite Functions
                                 登录后复制

sqlite3操作流程：0. 查看当前数据库在编译阶段是否启动了线程安全
int sqlite3_threadsafe()； 0-未启⽤； 1-启⽤
需要注意的是sqlite3是有三种安全等级的：
1. ⾮线程安全模式--不使用任何锁，效率最高
2. 线程安全模式(不同的连接在不同的线程/进程间是安全的，即⼀个句柄不能⽤于多线程
间)
3. 串⾏化模式(可以在不同的线程/进程间使⽤同⼀个句柄)
1. 创建/打开数据库⽂件，并返回操作句柄
int sqlite3_open(const char *filename, sqlite3 **ppDb) 成功返回SQLITE_OK
//若在编译阶段启动了线程安全，则在程序运⾏阶段可以通过参数选择线程安全等级
int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const
char *zVfs );
flag:
SQLITE_OPEN_READWRITE -- 以可读可写⽅式打开数据库⽂件
SQLITE_OPEN_CREATE -- 不存在数据库⽂件则创建
SQLITE_OPEN_NOMUTEX--多线程模式，只要不同的线程使⽤不同的连接即可保证线程
安全
SQLITE_OPEN_FULLMUTEX--串⾏化模式
返回：SQLITE_OK表⽰成功，否则有问题
2. 执⾏语句
int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),
void* arg, char **err)
err是错误信息
int (*callback)(void*,int,char**,char**)
void* : 是设置的在回调时传⼊的arg参数
int：⼀⾏中数据的列数
char**：存储⼀⾏数据的字符指针数组
char**：每⼀列的字段名称
这个回调函数有个int返回值，成功处理的情况下必须返回0，返回⾮0会触发ABORT退出程序
返回：SQLITE_OK表⽰成功
3. 销毁句柄
int sqlite3_close(sqlite3* db); 成功返回SQLITE_OK
int sqlite3_close_v2(sqlite3*); 推荐使⽤--⽆论如何都会返回SQLITE_OK
获取错误信息
const char *sqlite3_errmsg(sqlite3* db);

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.

                     sqlite3 test.db//打开数据库文件
   .table//查看表
                                 登录后复制

//封装实现一个SqliteHelper类，提供简单的sqlite数据库操作接口，完成对数据增删改查操作
//1.创建/打开数据库文件
//2.针对打开的数据库进行操作
//2.1表的操作
//2.2数据的操作
//3.关闭数据库
#include <sqlite3.h>
#include <string>
#include <iostream>
class SqliteHelper
{
public:
typedef int(*SqliteCallback)(void*,int,char**,char**);
SqliteHelper(std::string dbfile)
:_dbfile(dbfile)
{}
bool open(int safeLevel=SQLITE_OPEN_FULLMUTEX)//打开数据库文件
{
//默认安全级别为串行化模式
//int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs )
int ret=sqlite3_open_v2(_dbfile.c_str(),&_handler,SQLITE_OPEN_CREATE|SQLITE_OPEN_READWRITE|safeLevel,nullptr);
if(ret!=SQLITE_OK)
{
std::cout<<"创建/打开数据库文件失败："<<sqlite3_errmsg(_handler)<<std::endl;
return false;
}
return true;
}
bool exec(const std::string& sql,SqliteCallback cb,void* arg)//进行操作
{
//int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),void* arg, char **err)
int ret=sqlite3_exec(_handler,sql.c_str(),cb,arg,nullptr);
if(ret!=SQLITE_OK)
{
std::cout<<sql<<"执行失败："<<sqlite3_errmsg(_handler)<<std::endl;
return false;
}
return true;
}
void close()//关闭数据库文件
{
//int sqlite3_close_v2(sqlite3*)
sqlite3_close_v2(_handler);
}
private:
sqlite3* _handler;//句柄
std::string _dbfile;//数据库文件名
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.

登录后复制

#include "sqlite.hpp"
#include <vector>
#include <cassert>
int select_stu_callback(void* arg,int col_count,char** result,char** fields_name) {
std::vector<std::string> *arry = (std::vector<std::string>*)arg;
arry->push_back(result[0]);
return 0; //必须有！！！
}
int main()
{
SqliteHelper helper("./test.db");
//1. 创建/打开库文件
assert(helper.open());
//2. 创建表(不存在则创建)，学生信息：学号，姓名，年龄
const char *ct = "create table if not exists student(sn int primary key, name varchar(32), age int);";
assert(helper.exec(ct, nullptr, nullptr));
//3. 新增数据，修改，删除，查询
// const char *insert_sql = "insert into student values(1, '小明', 18), (2, '小黑', 19), (3, '小红', 18);";
// assert(helper.exec(insert_sql, nullptr, nullptr));
// const char *update_sql = "update student set name='张小明' where sn=1";
// assert(helper.exec(update_sql, nullptr, nullptr));
// const char *delete_sql = "delete from student where sn=3";
// assert(helper.exec(delete_sql, nullptr, nullptr));
const char *select_sql = "select name from student;";
std::vector<std::string> arry;
assert(helper.exec(select_sql, select_stu_callback, &arry));
for (auto &name : arry) {
std::cout << name << std::endl;
}
//4. 关闭数据库
helper.close();
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.

                     GTest

   使用：

宏断言
事件机制(全局，单独)

GTest中的断⾔的宏可以分为两类：

ASSERT_系列：如果当前点检测失败则退出当前函数
EXPECT_系列：如果当前点检测失败则继承往下执⾏

常用断言：
登录后复制

// bool值检查
ASSERT_TRUE(参数)，期待结果是true
ASSERT_FALSE(参数)，期待结果是false
//数值型数据检查
ASSERT_EQ(参数1，参数2)，传⼊的是需要⽐较的两个数 equal
ASSERT_NE(参数1，参数2)，not equal，不等于才返回true
ASSERT_LT(参数1，参数2)，less than，⼩于才返回true
ASSERT_GT(参数1，参数2)，greater than，⼤于才返回true
ASSERT_LE(参数1，参数2)，less equal，⼩于等于才返回true
ASSERT_GE(参数1，参数2)，greater equal，⼤于等于才返回true

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.

                     TEST(参数1,参数2){//......}
   参数1是测试名称，第二个参数是详细名称
   RUN_ALL_TESTS();//运行全部测试
                                 登录后复制

#include<iostream>
#include<gtest/gtest.h>
int abs(int x)
{
return x > 0 ? x : -x;
}
TEST(abs_test, test1)
{
ASSERT_TRUE(abs(1) == 1) << "abs(1)=1";
ASSERT_TRUE(abs(-1) == 1);
ASSERT_FALSE(abs(-2) == -2);
ASSERT_EQ(abs(1),abs(-1));
ASSERT_NE(abs(-1),0);
ASSERT_LT(abs(-1),2);
ASSERT_GT(abs(-1),0);
ASSERT_LE(abs(-1),2);
ASSERT_GE(abs(-1),0);
}
int main(int argc,char *argv[])
{
//将命令⾏参数传递给gtest
testing::InitGoogleTest(&argc, argv);
// 运⾏所有测试案例
return RUN_ALL_TESTS();
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.

   全局事件：针对整个测试程序。实现全局的事件机制，必要创建⼀个⾃⼰的类，然后继承 testing::Environment类，然后分别实现成员函数 SetUp 和 TearDown ，同时在main函数内进⾏调⽤ testing::AddGlobalTestEnvironment(new MyEnvironment); 函数添加全局的事件机制
   全局测试套件，其实就是用户本身界说一个全局的测试环境类
                                 登录后复制

#include <gtest/gtest.h>
#include <iostream>
#include <unordered_map>
//全局事件:针对整个测试程序,提供全局事件机制,能够在测试之前配置测试环境数据,测试完毕后清理数据
//先定义环境类,通过继承testing::Environment的派⽣类来完成
//重写的虚函数接⼝SetUp会在测试之前被调⽤; TearDown会在测试完毕后调⽤.
std::unordered_map<std::string,std::string> dict;
class MyMapTest:public testing::Environment
{
public:
virtual void SetUp() override
{
std::cout<<"所有单元开始测试"<<std::endl;
dict.insert(std::make_pair("hello","你好"));
dict.insert(std::make_pair("世界","world"));
}
virtual void TearDown() override
{
std::cout<<"所有单元测试结束"<<std::endl;
dict.clear();
}
};
TEST(MyMapTest,test1)
{
ASSERT_EQ(dict.size(),2);
dict.erase("hello");
}
TEST(MyMapTest,test2)
{
ASSERT_EQ(dict.size(),2);
}
int main(int argc,char* argv[])
{
testing::InitGoogleTest(&argc,argv);
testing::AddGlobalTestEnvironment(new MyMapTest());
RUN_ALL_TESTS();
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.

TestSuite事件：针对⼀个个测试套件。测试套件的事件机制我们同样必要去创建⼀个类，继承⾃testing::Test ，实现两个静态函数 SetUpTestCase 和 TearDownTestCase ，测试套件的事件机制不必要像全局事件机制⼀样在 main 注册，⽽是必要将我们平常使⽤的 TEST 宏改为 TEST_F 宏。

SetUpTestCase() 函数是在测试套件第⼀个测试⽤例开始前执⾏
TearDownTestCase() 函数是在测试套件最后⼀个测试⽤例竣事后执⾏
必要注意TEST_F的第⼀个参数是我们创建的类名，也就是当前测试套件的名称，如许在TEST_F宏的测试套件中就可以访问类中的成员了。

登录后复制

#include <iostream>
#include <gtest/gtest.h>
#include <unordered_map>
class MyTest:public testing::Test
{
public:
static void SetUpTestCase()//所有单元测试开始时初始化
{
//假设有一个全局的变量，公共的测试数据就可以在这里初始化
std::cout<<"所有单元测试开始"<<std::endl;
}
virtual void SetUp() override//每次单元测试时初始化
{
//假设有一个全局的变量，在这里插入每个单元测试所需的独立的测试数据
std::cout<<"单元测试开始"<<std::endl;
mymap.insert(std::make_pair("hello","world"));
}
virtual void TearDown() override//每次单元测试结束时清理
{
std::cout<<"单元测试结束"<<std::endl;
mymap.clear();
}
static void TearDownTestCase()//所有单元测试结束时清理
{
std::cout<<"所有单元测试结束"<<std::endl;
}
public:
std::unordered_map<std::string,std::string> mymap;
};
//必须和类同名
TEST_F(MyTest,test1)
{
mymap.insert(std::make_pair("aaa","aaa"));
ASSERT_EQ(mymap.size(),2);
}
TEST_F(MyTest,test2)
{
ASSERT_EQ(mymap.size(),1);
}
int main(int argc,char* argv[])
{
testing::InitGoogleTest(&argc,argv);
RUN_ALL_TESTS();
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.

                     C++11异步操作实现线程池

   怎样获取线程池里的线程完成使命后的结果？
   std::future：获取使命结果
   std::future是C++11标准库中的⼀个模板类，它表⽰q⼀个异步操作的结果。当我们在多线程编程中使⽤异步使命时，std::future可以帮助我们在必要的时候获取使命的执⾏结果。std::future的⼀个重要特性是可以或许壅闭当前线程，直到异步操作完成，从⽽确保我们在获取结果时不会遇到未完成的操作。
   asyn
                                 登录后复制

#include <iostream>
#include <future>
#include <thread>
#include <chrono>
int add(int num1,int num2)
{
std::cout<<"aaaaaaaaaaaaaaaaaaaaa"<<std::endl;
std::cout<<"加法任务"<<std::endl;
std::this_thread::sleep_for(std::chrono::seconds(5));
std::cout<<"bbbbbbbbbbbbbbbbbbbbb"<<std::endl;
return num1+num2;
}
int main()
{
//std::async(func, ...) std::async(policy, func, ...)
//std::launch::deferred 在执行get获取异步结果的时候，才会执行异步任务
//std::launch::async 内部会创建工作线程，异步的完成任务
std::cout<<"----------1----------"<<std::endl;
std::future<int> result=std::async(std::launch::deferred,add,2,3);//把任务交它，它会创建一个工作线程来执行任务
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout<<"----------2----------"<<std::endl;
int ret=result.get();//此时如果任务还未完成，会阻塞等待
std::cout<<"----------3----------"<<std::endl;
std::cout<<"result="<<ret<<std::endl;
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.

promise
登录后复制

#include <future>
#include <thread>
#include <chrono>
#include <iostream>
//通过在线程中对promise对象设置数据，其他线程中通过future获取设置数据的方式实现获取异步任务执行结果的功能
void add(int num1,int num2,std::promise<int>& prom)
{
prom.set_value(num1+num2);
std::this_thread::sleep_for(std::chrono::seconds(3));
}
int main()
{
std::promise<int> prom;
std::future<int> fut=prom.get_future();//二者从此绑定了关系
std::thread th(add,11,12,std::ref(prom));//创建线程执行任务
std::cout<<fut.get()<<std::endl;//get会进行同步，等任务完成了才得到结果
th.join();//等待线程
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.

packaged_task
登录后复制

#include <iostream>
#include <future>
#include <memory>
#include <thread>
//pakcaged_task的使用
//pakcaged_task 是一个模板类，实例化的对象可以对一个函数进行二次封装，
//pakcaged_task可以通过get_future获取一个future对象，来获取封装的这个函数的异步执行结果
int add(int num1,int num2)
{
return num1+num2;
}
int main()
{
//task(11, 22); task可以当作一个可调用对象来调用执行任务
//但是它又不能完全的当作一个函数来使用
//std::async(std::launch::async, task, 11, 22);
//std::thread thr(task, 11, 22);
//但是我们可以把task定义成为一个指针，传递到线程中，然后进行解引用执行
//但是如果单纯指针指向一个对象，存在生命周期的问题，很有可能出现风险
//思想就是在堆上new对象，用智能指针管理它的生命周期
auto pTask=std::make_shared<std::packaged_task<int(int,int)>>(add);//make_shared效率更高
std::future<int> fut=pTask->get_future();//建立同步关系
std::thread th([pTask](){
(*pTask)(1,2);
});
std::cout<<"结果为："<<fut.get()<<std::endl;
th.join();
return 0;
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

                     实现线程池
   packaged_task+futur
                                 登录后复制

#include <thread>
#include <future>
#include <vector>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <memory>
#include <iostream>
class ThreadPool
{
public:
using Func = std::function<void(void)>;
ThreadPool(int threadCount = 1) // 构造函数
: _stop(false)
{
for (int i = 0; i < threadCount; ++i) // 创建指定数量的线程
{
_threads.emplace_back(std::thread(&ThreadPool::entry, this));
}
}
~ThreadPool() // 析构函数
{
stop();
}
template <class F, class... Args>
// func是用户传过来的函数，接下来就是不定参数
// push函数会把func封装成一个异步任务(packaged_task)，抛入任务池中，由工作线程取出处理
// push函数会返回一个future对象，让用户能得到任务结果
auto push(F &&func, Args &&...args) -> std::future<decltype(func(args...))> // 用户添加任务
{
// 1.将用户发过来的函数封装成packaged_task
using retType = decltype(func(args...));
auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)...); // 将参数绑定，注意使用完美转发
auto task = std::make_shared<std::packaged_task<retType()>>(tmp_func);
std::future<retType> fut = task->get_future();
// 2.构造匿名lambda表达式，里面执行task
{
std::unique_lock<std::mutex> lock(_mutex); // 加锁
// 将lambda表达式抛入任务池中
_tasks.push_back([task]()
{ (*task)(); });
_cond.notify_one(); // 唤醒一个线程
}
return fut;
}
void stop() // 线程池停止工作
{
// 0.如果已经停止
if (_stop == true)
return;
// 1.将停止标志设为true
_stop = true;
// 2.唤醒所有线程
_cond.notify_all();
// 3.等待所有线程
for (auto &th : _threads)
{
th.join();
}
}
private:
// 线程入口函数，内部不断从任务池中取出任务执行
void entry()
{
while (!_stop)
{
std::vector<Func> tmpTasks;
{
std::unique_lock<std::mutex> lock(_mutex); // 加锁
_cond.wait(lock, [this]()
{ return _stop || !_tasks.empty(); }); // 等待唤醒
tmpTasks.swap(_tasks);//取出任务
}
std::cout<<"线程开始执行任务"<<std::endl;
for(auto& task:tmpTasks)//执行任务
{
task();
}
}
}
private:
std::atomic<bool> _stop; // 线程池是否停止运行
std::mutex _mutex; // 互斥
std::condition_variable _cond; // 同步
std::vector<Func> _tasks; // 任务池
std::vector<std::thread> _threads; // 线程池
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.

                     需求分析

   1、焦点概念

   其中， Broker Server是最核⼼的部门, 负责消息的存储和转发。
   ⽽在AMQP(Advanced Message Queuing Protocol-⾼级消息队列协议，⼀个提供统⼀消息服务的应⽤层标准⾼级消息队列协议，为⾯向消息的中间件设计，使得服从该规范的客⼾端应⽤和消息中间件服务器的全功能互操作成为可能)模型中，也就是消息中间件服务器Broker中，⼜存在以下概念：

假造机 (VirtualHost): 类似于 MySQL 的 "database", 是⼀个逻辑上的集合。⼀个 BrokerServer 上可以存在多个 VirtualHost
交换机 (Exchange): ⽣产者把消息先发送到 Broker 的 Exchange 上，再根据不同的规则, 把消息转发给不同的 Queue
队列 (Queue): 真正⽤来存储消息的部门，每个消费者决定⾃⼰从哪个 Queue 上读取消息
绑定 (Binding): Exchange 和 Queue 之间的关联关系，Exchange 和 Queue 可以理解成 "多对多"关系，使⽤⼀个关联表就可以把这两个概念接洽起来
消息 (Message): 传递的内容

   所谓的 Exchange 和 Queue 可以理解成 "多对多" 关系, 和数据库中的 "多对多" ⼀样. 意思是:
   ⼀个 Exchange 可以绑定多个 Queue (可以向多个 Queue 中转发消息)
   ⼀个 Queue 也可以被多个 Exchange 绑定 (⼀个 Queue 中的消息可以来⾃于多个 Exchange)

   2、焦点API

   对于 Broker 来说, 要实现以下核⼼ API，通过这些 API 来实现消息队列的基本功能
   1. 创建交换机 (exchangeDeclare)
   2. 销毁交换机 (exchangeDelete)
   3. 创建队列 (queueDeclare)
   4. 销毁队列 (queueDelete)
   5. 创建绑定 (queueBind)
   6. 解除绑定 (queueUnbind)
   7. 发布消息 (basicPublish)
   8. 订阅消息 (basicConsume)
   9. 确认消息 (basicAck)
   10. 取消订阅 (basicCancel)
   另⼀⽅⾯, Producer 和 Consumer 则通过⽹络的⽅式, 远程调⽤这些 API, 实现⽣产者消费者模型
   3、交换机范例

   对于 RabbitMQ 来说, 主要⽀持四种交换机范例：

Direct
Fanout
Topic
Header

其中 Header 这种⽅式⽐较复杂, ⽐较少⻅。常⽤的是前三种交换机范例，项⽬中也主要实现这三种

Direct: ⽣产者发送消息时, 直接指定被该交换机绑定的队列名
Fanout: ⽣产者发送的消息会被复制到该交换机的全队伍列中
Topic: 绑定队列到交换机上时, 指定⼀个字符串为 bindingKey。发送消息指定⼀个字符串为routingKey。当 routingKey 和 bindingKey 满⾜⼀定的匹配条件的时候, 则把消息投递到指定队列

   这三种操作就像给 qq 群发红包.
   Direct 是发⼀个专属红包, 只有指定的⼈能领.
   Fanout 是使⽤了魔法, 发⼀个 10 块钱红包, 群⾥的每个⼈都能领 10 块钱.
   Topic 是发⼀个画图红包, 发 10 块钱红包, 同时出个题, 得画的像的⼈, 才能领. 也是每个领到的⼈都能领 10 块钱.
   4、持久化

   Exchange, Queue, Binding, Message 等数据都有持久化需求
   当程序重启 / 主机重启, 保证上述内容不丢失
   5、网络通讯

   ⽣产者和消费者都是客⼾端程序, Broker 则是作为服务器，通过⽹络进⾏通讯。
   在⽹络通讯的过程中, 客⼾端部门要提供对应的 api, 来实现对服务器的操作。
   1. 创建 Connection
   2. 关闭 Connection
   3. 创建 Channel
   4. 关闭 Channel
   5. 创建队列 (queueDeclare)
   6. 销毁队列 (queueDelete)
   7. 创建交换机 (exchangeDeclare)
   8. 销毁交换机 (exchangeDelete)
   9. 创建绑定 (queueBind)
   10. 解除绑定 (queueUnbind)
   11. 发布消息 (basicPublish)
   12. 订阅消息 (basicConsume)
   13. 确认消息 (basicAck)
   14. 取消订阅(basicCancel)
   可以看到, 在 Broker 的基础上, 客⼾端还要增长 Connection 操作和 Channel 操作

Connection 对应⼀个 TCP 连接
Channel 则是 Connection 中的逻辑通道

   ⼀个 Connection 中可以包含多个 Channel。Channel 和 Channel 之间的数据是独⽴的，不会相互⼲扰。如许做主要是为了可以或许更好的复⽤ TCP 连接, 达到⻓连接的结果, 避免频仍的创建关闭 TCP 连接。
   Connection 可以理解成⼀根⽹线. Channel 则是⽹线⾥详细的线缆.

   6、消息应答

   被消费的消息, 必要进⾏应答。应答模式分成两种：

⾃动应答: 消费者只要消费了消息, 就算应答完毕了，Broker 直接删除这个消息
⼿动应答: 消费者⼿动调⽤应答接⼝, Broker 收到应答哀求之后, 才真正删除这个消息

   ⼿动应答的⽬的, 是为了保证消息确实被消费者处理成功了. 在⼀些对于数据可靠性要求⾼的场景, ⽐较常⻅.
   模块分别

   1、服务端模块

   1.1、数据管理模块

   1.1.1、交换机数据管理模块

   要管理的数据：描述了一个交换机应该有什么数据

交换机名称：唯一标识
交换机范例：决定了消息的转发方式，每个队列绑定中有个binding_key，每条消息中有个routing_key

         直接交换：binding_key与routing_key相同时，则将消息放入队列
   广播交换：将消息放入交换机绑定的全队伍列中
   主题交换：routing_key与多个绑定队列的binding_key有个匹配规则，匹配成功了则放入

持久化标志：决定了当前交换机信息是否必要持久化存储
自动删除标志：指的是关联了当前交换机的全部客户端都退出了，是否要自动删除交换机
交换机的其它参数：当前未使用

对交换机的管理操作：

创建交换机：本质上必要的是声明--------强断言头脑-有就OK，没有则创建的意思
删除交换机：注意每个交换机都会绑定一个或多个队列，意味着会有一个或多个绑定信息，因此删除交换机必要删除相干的绑定信息
获取指定名称交换机
获取当前交换机数目

   1.1.2、队列数据管理模块

   要管理的数据：

队列名称：唯一的表现
持久化标志：决定了是否将队列信息持久化存储起来，重启后，这个队列是否还存在
是否独占标志：独占指的就是只有当前客户端本身可以或许订阅队列消息
自动删除标志：当订阅了当前队列的全部客户端退出后，是否删除队列(暂未实现)
其它参数：暂未使用

提供的管理操作：

创建队列
删除队列
获取指定队列信息
获取队列数目
获取全队伍列名称：当体系重启后，必要重新加载数据，加载汗青消息(消息以队列为单元存储在文件中)，而加载消息必要知道队列名称，因为背面消息存储时，存储文件以队列名称命名

   一个队列如果持久化标志为false，则意味偏重启后，队列就没了，也就没有客户端可以或许订阅队列的消息，因此这个队列的消息如果持久化存储了，是没故意义的，因此通常一个队列的持久化标志是false，那么它的消息也就不必要持久化。
   1.1.3、绑定数据管理模块

   描述一下将哪个队列与哪个交换机绑定到了一起
   管理的数据：

交换机名称
队列名称
binding_key：绑定密钥，描述了交换机在主题交换&直接交换的消息发布匹配规则

         由数字、字符、_、#、。、*构成 binding_key: news.music.# 消息中的routing_key: news.music.pop
      管理的操作：

添加绑定
解除绑定
获取交换机相干的全部绑定信息：删除交换机的时候，要删除相干的绑定信息。当消息发布到交换机，交换机得通过这些信息来将消息发布到指定队列
获取队列相干的全部绑定信息：删除队列的时候，要删除相干的绑定信息
获取绑定信息数目

1.1.4、消息数据管理模块

消息信息：

消息属性

ID：消息的唯一标识
持久化标志
routing_key：决定了当前消息要发布的队列(消息发布到交换机后，根据绑定队列的binding_key决定是否发布到指定队列)

消息主体：消息内容

以下是服务端为了管理所添加的信息

存储偏移量：消息以队列为单元存储在文件中，这个偏移量，是当前相对于文件起始位置的偏移量
消息长度：从偏移量位置取出指定长度的消息(解决粘报标题)
是否有用标志：标识当前消息是否已经被删除(删除一条消息，并不必要从文件里把背面的数据拷贝到前面，而是设置标志，当一个文件中，有用消息占总消息比例不到50%，且数据量凌驾2000，则进行垃圾回收，重新整理文件数据存储)当体系重启也只必要重新加载有用消息即可(相称于进行了一次垃圾回收)

消息的管理：

管理方式：以队列为单元进行管理(因为消息的全部操作都是以队列为单元)
管理数据：

消息链表：保存全部的待推送消息
待确认消息hash：消息推送给客户端后，会等候客户端进行消息确认，收到确认后，才会真正删除消息
持久化消息hash：假设消息都会进行持久化存储，操作过程中会存在垃圾回收操作，但是垃圾回收会改变息的存储位置，内存中的消息也会存储消息的现实存储位置，垃圾回收后就不一致了。因此每次垃圾回收后都必要用新的位置，去更新持久化的消息。
持久化的有用消息数目
持久化的总消息数目：决定了什么时候进行垃圾回收

管理操作：

向队列新增消息
获取队首消息：获取消息后，就会将消息从待推送链表删除(不再是待发送，而是待确认)
对消息进行确认；从待确认消息中移除消息，并进行持久化数据的删除
恢复队列汗青消息：主要是在构造函数中进行(只有在重启时进行)
垃圾回收(消息持久化子模块完成)：持久化文件中有用消息比例小于50%，且总消息数目凌驾2000进行垃圾回收
删除队列相干消息文件：当一个队列被删除了，那它的消息也就没有存在意义了。

队列消息管理：

初始化队列消息结构
移除队列消息结构：在一个队列创建/删除时调用
向队列新增消息
对队列消息进行确认
恢复队列汗青消息

   以上四个模块分别实现数据的管理(增删查)，以及持久化的存储
   1.2、假造机数据管理模块

   假造机其实就是交换机+队列+绑定+消息的逻辑单元。
   因此假造机的数据管理其实就是将以上四个模块的归并管理
   要管理的数据：

交换机数据管理句柄
队列数据管理句柄
绑定信息数据管理句柄
消息数据管理句柄

要管理的操作：

声明/删除交换机：注意---在删除交换机的时候要删除相干的绑定信息
声明/删除队列：注意---在删除队列的时候，要删除相干的绑定信息
队列的绑定/解除绑定
获取指定队列的消息
对指定队列的指定消息进行确认
获取交换机相干的全部绑定信息：一个消息要发布给指定交换机的时候，交换机获取全部绑定信息啦，来确定消息要发布到哪个队列

   1.3、交换路由模块

   消息的发布，将一条消息发布到交换机上，由交换机决定放入哪些队列。而决定交给哪个队列，其中交换机范例起到了很大的作用(直接交换、广播交换、主题交换)
   直接交换和广播交换头脑都较为简朴，而主题交换涉及到了一个规则匹配的流程。交换路由模块就是专门做匹配过程的。
   在每个队列和交换机的绑定信息中，都有一个binding_key，这是队列发布的匹配规则。
   在每条要发布的消息中，都有一个routing_key，这是消息的发布规则。
   交换机有三种交换范例：直接、广播、主题。

广播：直接将消息发布给交换机的全部绑定队列
直接：routing_key与binding_key完全一致则匹配成功
主题：binding_key中是匹配规则，routing_key是消息规则，匹配成功才能发布

交换路由模块本质上来说，没有要管理的数据，只有向外提供的路由匹配操作：

提供一个判定routing_key与binding_key是否可以或许匹配成功的接口。
判定routing_key是否符合规定：只能由数字字母_. 构成
判定binding是否符合规定：只能由数字字母_.#* 构成

   1.4、消费者管理模块

   消费者指的是订阅了一个队列消息的客户端，一旦这个队列有了消息就会推送给这个客户端。在焦点API中有一个订阅消息的服务，这里的订阅不是订阅了某条消息，而是订阅了某个队列的消息。当前主要实现了消息推送功能，因此一旦有了消息就要可以或许找到消费者相干的信息(消费者对应的信道)。
   消费者信息：

消费者表现--tag
订阅队列名称：当前队列有消息就推送给这个客户端，以及当客户端收到消息，必要对指定队列的消息进行确认
自动确认标志：自动确认---推送消息后，直接删除消息不必要额外确认，手动确认---推送消息后，必要收到确认回复再去删除消息
消费处理回调函数指针：队列有一条消息后，通过哪个函数进行处理(函数内部其实逻辑固定---向指定客户端推送消息)

消费者管理：

管理头脑：一队列为单元进行管理

每个消费者订阅的都是指定队列的消息，消费者对消息进行确认也是以队列进行确认
最关键的是：当指定队列中有消息了，必然是获取订阅了这个队列的消费者进行消息推送

队列消费者管理结构：

数据信息：消费者链表---保存当前队列的全部消费者信息(RR轮转每次取出下一个消费者进行消息推送---一条消息只必要被一个客户端处理即可)
管理操作：

新增消费者
RR轮转获取消费者
删除消费者
队列消费者数目
是否为空

管理操作：

初始化队列消费者结构
删除队列消费者结构
向指定队列添加消费者
获取指定队列消费者
删除指定队列消费者

   1.5、信道管理模块

   信道是网络通讯中的一个概念，叫做通讯通道。
   网络通行的时候，必然都是通过网络通讯连接来完成的，为了可以或许更加充分地利用资源，因此对通讯连接又进行了进一步的细化，细化出了通讯信道。
   对于用户来说，一个通讯信道，就是进行网络通讯的载体，而一个真正的通讯连接，可以创建出多个通讯信道。
   每一个信道之间，在用户眼中是相互独立的，而在本质的底层它们使用同一个通讯连接进行网络服务。
   一个连接可能会对应有多个通讯信道。一旦某个客户端要关闭通讯，关闭的不是连接，而是本身对应通讯通道。，关闭信道就必要将客户端的订阅取消。
   信道提供的服务操作：

声明/删除交换机
声明/删除队列
绑定/解绑队列与交换机
发布消息/订阅队列消息/取消队列订阅/队列消息确认

信道要管理的数据：

信道ID
信道关联的假造机句柄
信道关联的消费者句柄：当信道关闭的时候，全部关联的消费者订阅都要取消，相称于删除全部的相干消费者
工作线程池句柄：信道进行了消息发布到指定队列操作之后，从指定队列获取一个消费者，对这条消息进行消费，也就是将这条消息推送给一个客户端的操作交给线程池执行，并非每个信道都有一个线程池，而是整个服务器有一个线程池，各人全部的信道都是通过同一个线程池进行异步操作而已

信道的管理：

创建一个信道
关闭一个信道
获取指定信道句柄

   1.6、连接管理模块

   就是一个网络通讯对应的连接。
   使用muduo库来实现底层通讯，muduo库中本身就有Connection连接的概念和对象类。但是我们在连接中，尚有一个上层通讯信道的概念，这个概念在muduo库中时没有的。因此，我们必要在用户层面，对这个muduo库的Connection连接进行二次封装。
   当一个连接要关闭时，就应该把连接关联的信道全部关闭，因此也要管理关联的信道。形成我们本身所必要的连接管理。
   管理数据：

muduo库的通讯连接
当前连接关联的信道管理句柄

连接提供的操作：

创建信道
关闭信道

管理的操作：

新增连接
关闭连接
获取指定连接信息

   1.7、服务端BrokerServer模块

   对以上全部模块的整合，整合成一个服务器

一个服务器有一个工作线程池，其它全部的信道操作的都是这同一个线程池
一个服务器有一个假造机，其它全部交换机，队列，绑定，消息的操作都是针对这个假造机进行的
一个服务器有一个消费者管理
通讯相干连接管理，协议处理模块句柄，也是一整个服务器有一套

管理信息：

假造机管理模块句柄
消费者管理模块句柄
连接管理模块句柄
工作线程池句柄
muduo库通讯所需元素......

   2、客户端模块

   2.1、消费者管理模块

   一个订阅客户端，当订阅一个队列消息时，就相称于创建了一个消费者

消费者标识
订阅的队列消息
自动确认标志
消息回调处理函数指针

   当当前消费者订阅了某一个队列的消息，这个队列有了消息之后，就会将消息推送给这个客户端，这个时候收到了消息则使用回调函数进行处理，处理完毕后，根据确认标志决定是否进行消息确认。
   管理操作：增删查
   2.2、信道管理模块

   客户端的信道与服务端的信道是逐一对应的，服务端信道提供的服务，客户端都有，相称于服务端为客户端提供服务，客户端为用户提供服务。
   全部提供的操作与服务端雷同，因为客户端给用户要提供什么服务，服务器就要给客户端提供什么服务。
   管理信息：

消费者管理句柄：每个信道都有本身相干的消费者
线程池句柄：对推送过来的消息进行回调处理，处理过程通过工作线程来进行
信道关联的连接

信道提供的服务：

声明/删除交换机
声明/删除交换机
绑定/解绑队列与交换机
发布消息/确认消息
订阅消息队列/取消订阅队列消息
创建/关闭信道

   信道的管理：信道的增删查
   2.3、连接管理模块

   对于用户来说，全部的服务都是通过信道完成的，信道在用户交角度就是一个通讯通道(而不是连接)。因此全部的哀求都是通过信道来完成的，连接的管理就包含了客户端资源的整合。
   客户端连接的管理，本质上就是对客户端TcpClient的二次封装和管理。面临用户。不必要有客户端的概念，连接对于用户来说就是客户端，通过连接创建信道，通过信道完成本身所必要的服务。因此，当前客户端这边的连接，就是一个资源的载体
   管理操作：

连接服务器
创建信道
关闭信道
关闭连接

管理的资源：

工作线程池
连接关联的信道管理句柄

2.4、异步线程池模块

TcpClient模块必要一个EventLoopThread模块进行IO事件监控。
收到推送消息后，必要对推送过来的消息进行处理，因此必要一个线程池来帮助我们完成消息处理的过成

   将异步工作线程池模块单独拎出来，原因是多个连接用一个EventLoopThread进行IO事件监控就够了，以及全部的推送消息也只必要有一个线程池就够了。并不必要每个连接都有一个EveentLoop，也不必要每个信道都有本身的线程池。
   2.4、基于以上三个模块封装实现：订阅/发布客户端

   订阅客户端：订阅一个队列的消息，收到推送过来消息进行处理。
   发布客户端：向一个交换机发布消息

项目创建

   mqdemo：编写⼀些功能⽤例时所在的⽬录
   mqcommon：公共模块代码(线程池，数据库访问，⽂件访问，⽇志打印，pb相干，以及其他的⼀些琐碎功能模块代码)
   mqclient: 客⼾端模块代码
   mqserver: 服务器模块代码
   mqtest：单元测试
   mqthird: ⽤到的第三⽅库存放⽬录
   服务端实现

   1、实现日志工具

                                 登录后复制

#pragma once
#include <iostream>
#include <ctime>
namespace lsxmq
{
#define DBG_LEVEL 0
#define INF_LEVEL 1
#define ERR_LEVEL 2
#define DEFAULT_LEVEL DBG_LEVEL
#define LOG(levelStr, level, format, ...) \
{ \
if (level >= DEFAULT_LEVEL) \
{ \
time_t t = time(nullptr); \
struct tm *ptm = localtime(&t); \
char timeStr[32]; \
strftime(timeStr, 31, "%H:%M:%S", ptm); \
printf("[%s][%s][%s:%d]" format "\n", levelStr, timeStr, __FILE__, __LINE__, ##__VA_ARGS__); \
} \
}
//__VA_ARGS__是使用可变参数，##是为了保证没有可变参数时忽略前面的逗号
// 宏函数必须在一行里，在换行时加上\是为了转义\n
#define DLOG(format, ...) LOG("DBG", DBG_LEVEL, format, ##__VA_ARGS__)
#define ILOG(format, ...) LOG("INF", INF_LEVEL, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG("ERR", ERR_LEVEL, format, ##__VA_ARGS__)
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

                     2、实现使用Helper工具

   2.1、Sqlite基础操作类

                                 登录后复制

#pragma once
// 封装实现一个SqliteHelper类，提供简单的sqlite数据库操作接口，完成对数据增删改查操作
// 1.创建/打开数据库文件
// 2.针对打开的数据库进行操作
// 2.1表的操作
// 2.2数据的操作
// 3.关闭数据库
class SqliteHelper
{
public:
typedef int (*SqliteCallback)(void *, int, char **, char **);
SqliteHelper(std::string dbfile)
: _dbfile(dbfile)
{
}
bool open(int safeLevel = SQLITE_OPEN_FULLMUTEX) // 打开数据库文件
{
// 默认安全级别为串行化模式
// int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs )
int ret = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE | safeLevel, nullptr);
if (ret != SQLITE_OK)
{
ELOG("创建/打开数据库文件失败：%s",sqlite3_errmsg(_handler));
return false;
}
return true;
}
bool exec(const std::string &sql, SqliteCallback cb, void *arg) // 进行操作
{
// int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),void* arg, char **err)
int ret = sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr);
if (ret != SQLITE_OK)
{
ELOG("%s执行失败：%s",sql.c_str(),sqlite3_errmsg(_handler));
return false;
}
return true;
}
void close() // 关闭数据库文件
{
// int sqlite3_close_v2(sqlite3*)
sqlite3_close_v2(_handler);
}
private:
sqlite3 *_handler; // 句柄
std::string _dbfile; // 数据库文件名
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.

                     2.2、字符串操作类

   分割字符串功能：
                                 登录后复制

static size_t split(const std::string& str,const std::string& sep,std::vector<std::string>& vs)
{
size_t idx=0;
while(idx<str.size())
{
size_t pos=str.find(sep,idx);
//找不到分隔符了
if(pos==std::string::npos)
{
vs.push_back(str.substr(idx,str.size()-idx));
return vs.size();
}
//pos的位置是一个分割符，而idx是上一个分割符的下一个位置
if(idx==pos)//意味着出现连续分割符，中间没有有效子串
{
idx+=sep.size();
continue;
}
vs.push_back(str.substr(idx,pos-idx));
idx=pos+sep.size();
}
return vs.size();
}

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.

                     2.3、UUID天生器类

   1.天生8个0~255之间的随机数
   2.取出一个8字节的序号
   通过以上数据，构成16字节的数据，转换为16进制字符，共32个
   随机数的天生
                                 登录后复制

class UUIDHelper
{
public:
static std::string uuid()
{
std::random_device rd; // 这是一个随机数生成装置
// rd();//这样就可以生成机器随机数，但是生成机器随机数效率太低，所以不使用这种方法
// 实际上是把这个机器随机数作为随机数种子来生成伪随机数
std::mt19937_64 gernator(rd()); // 使用梅森旋转算法，生成一个伪随机数
// 因为要生成的随机数范围是0~255，所以需要限定随机数的生成范围
std::uniform_int_distribution<int> distribution(0, 255);
std::stringstream uuid;
for (int i = 0; i < 8; ++i) // 生成16位16进制字符
{
int number = distribution(gernator); // 生成随机数
uuid << std::setw(2) << std::setfill('0') << std::hex << number;
if (i == 3 || i == 5 || i == 7)
{
uuid << "-";
}
}
// 生成序号
static std::atomic<size_t> seq(0);
size_t tmpSeq = seq.fetch_add(1); // 每次加一
for (int i = 7; i >= 0; --i)
{
uuid << std::setw(2) << std::setfill('0') << std::hex << (tmpSeq >> (i * 8) & 0xff);
if (i == 6)
{
uuid << "-";
}
}
return uuid.str();
}
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.

2.3、文件操作类

登录后复制

class FileHelper
{
public:
FileHelper(const std::string& filename)
:_filename(filename)
{}
bool exists()//文件是否存在
{
struct stat st;
//返回值为0表示能获取文件信息，文件自然就是存在的
return stat(_filename.c_str(),&st)==0;
}
size_t size()//文件大小
{
struct stat st;
if(stat(_filename.c_str(),&st)!=0)
{
return 0;
}
return st.st_size;
}
bool read(char* body,size_t offset,size_t len)//从文件指定位置读取指定长度内容
{
//1.打开文件
std::ifstream ifs(_filename.c_str(),std::ios_base::in|std::ios_base::binary);
if(ifs.is_open()==false)
{
ELOG("%s文件打开失败",_filename.c_str());
ifs.close();
return false;
}
//2.移动到指定位置
ifs.seekg(offset,std::ios_base::beg);
//3.读取文件内容
ifs.read(&body[0],len);
if(ifs.good()==false)
{
ELOG("%s文件读取失败",_filename.c_str());
ifs.close();
return false;
}
//4.关闭文件
ifs.close();
return true;
}
bool read(std::string& body)//从文件读取内容
{
size_t fsize=this->size();
//根据fsize调整body的大小
body.resize(fsize);
return read(&body[0],0,fsize);
}
bool write(const char* body,size_t offset,size_t len)//向文件指定位置写入内容
{
//1.打开文件
std::fstream fs(_filename.c_str(),std::ios_base::in|std::ios_base::out|std::ios_base::binary);
//这里不能ofstream
//想要跳转到文件的指定位置必须有读权限，istream可以fstream可以，但是ofstream不可以
if(fs.is_open()==false)
{
ELOG("%s文件打开失败",_filename.c_str());
fs.close();
return false;
}
//2.移动到指定位置
fs.seekg(offset,std::ios_base::beg);
//3.向文件写入内容
fs.write(body,len);
if(fs.good()==false)
{
ELOG("%s文件写入失败",_filename.c_str());
fs.close();
return false;
}
//4.关闭文件
fs.close();
return true;
}
bool write(const std::string& body)//向文件写入内容
{
return write(&body[0],0,body.size());
}
static std::string parentDirectory(const std::string& filename)//返回当前文件的父路径
{
// aaa/bbb/ccc/ddd/e.txt
int pos=filename.find_last_of('/');
if(pos==std::string::npos)//没找到
{
//e.txt
return "./";
}
return filename.substr(0,pos+1);
}
bool rename(const std::string& filename)//修改文件名称
{
return (::rename(_filename.c_str(),filename.c_str())==0);
}
static bool createFile(const std::string& filename)//创建文件
{
std::ofstream ofs(filename.c_str(),std::ios_base::binary|std::ios_base::out|std::ios_base::app);
if(ofs.is_open()==false)
{
ELOG("%s文件创建失败",filename.c_str());
ofs.close();
return false;
}
ofs.close();
return true;
}
static bool removeFile(const std::string& filename)//删除文件
{
return (::remove(filename.c_str())==0);
}
static bool createDirectory(const std::string& path)//创建目录
{
size_t idx=0;
while(idx<path.size())
{
int pos=path.find('/',idx);
if(pos==std::string::npos)
{
if(mkdir(path.c_str(),0775)!=0)
{
ELOG("%s目录创建失败",path.c_str());
return false;
}
return true;
}
std::string subPath=path.substr(0,pos);
if(mkdir(subPath.c_str(),0775)!=0)
{
ELOG("%s目录创建失败",subPath.c_str());
return false;
}
idx=pos+1;
}
return true;
}
static bool removeDirectory(const std::string& path)//删除目录
{
//rmdir必须保证文件夹是空的，所以不用这种方法
std::string cmd="rm -rf "+path;
return system(cmd.c_str())==0;
}
private:
std::string _filename;
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.

3、消息范例界说与交换机范例界说

登录后复制

syntax= "proto3";
package lsx;
enum ExchangType//交换机类型
{
UNKNOWTYPE=0;//因为必须从0开始，所以加一个未知类型
DIRECT=1;//直接
FANOUT=2;//广播
TOPIC=3;//主题
};
enum DeliveryMode//消息投递方式
{
UNKONWMODE=0;
DURABLE=1; //持久化
UNDURABLE=2; //非持久化
};
message BasicProperties //基本特性
{
string id=1; //消息ID
DeliveryMode deliveryMode=2; //是否持久化
string routing_key=3;
};
message Message
{
message Payload//有效载荷
{
BasicProperties properties=1; //基本属性
string body=2; //消息主体
string valid=3; //是否有效
//消息是否有效不用bool类型，而是使用字符的0/1
//因为bool类型序列化时true和false所占长度不同
//会导致修改文件中消息有效位后消息长度发生变化
}
Payload payload=1; //有效载荷
int32 offset=2; //偏移量
int32 len=3; //长度
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.

                     4、交换机数据管理

    界说交换机数据类

交换机名称
交换机范例
是否持久化标志
是否⾃动删除标志
其他参数

界说交换机数据持久化类(数据持久化的sqlite3数据库中)

创建/删除交换机数据表
新增交换机数据
移除交换机数据
查询全部交换机数据
查询指定交换机数据(根据名称)

界说交换机数据管理类

声明交换机，并添加管理(存在则OK，不存在则创建)
删除交换机
获取指定交换机
销毁全部交换机数据

登录后复制

namespace lsxmq
{
//1.定义交换机类
struct Exchange
{
using ptr=std::shared_ptr<Exchange>;
std::string name;//交换机名称
ExchangeType type;//交换机类型
bool durable;//是否持久化
bool autoDel;//是否自动删除
std::unordered_map<std::string,std::string> args;//其它参数
Exchange(const std::string& ename,ExchangeType etype
,bool edurable,bool eautoDel
,std::unordered_map<std::string,std::string> eargs)
:name(ename),type(etype),durable(edurable)
,autoDel(eautoDel),args(eargs)
{}
//args存储的是键值对，在存储数据库的时候，会组织一个格式进行存储：key=value&key=value
//将字符串反序列化为args
void setArgs(std::string strArgs);
//将args序列化为字符串
std::string getArgs();
};
//2.定义交换机数据持久化存储类--数据存储在sqlite数据库中
class ExchangeMapper
{
public:
ExchangeMapper(const std::string dbfile)
:_sqliteHelper(dbfile)
{}
bool createTable();//创建表
bool removeTable();//删除表
bool insert(Exchange::ptr& exchang);//向数据库插入一条交换机数据
bool erase(Exchange::ptr& exchang);//从数据库中删除一条交换机数据
std::unordered_map<std::string,Exchange::ptr> getAll();//获取数据库中所有的交换机数据
//name:ptr
private:
SqliteHelper _sqliteHelper;
};
//3.定义交换机数据管理类
class ExchangeManager
{
public:
ExchangeManager(const std::string dbfile)
:_mapper(dbfile)
{}
bool declareExchange(const std::string& ename,ExchangeType etype
,bool edurable,bool eautoDel
,std::unordered_map<std::string,std::string> eargs);//声明交换机
bool deleteExchange(const std::string& name);//删除交换机
Exchange::ptr selectExchange(const std::string& name);//获取指定交换机
bool exists(const std::string& anme);//指定交换机是否存在
void clear();//清理数据
private:
std::mutex _mutex;
ExchangeMapper _mapper;
std::unordered_map<std::string,Exchange::ptr> _exchanges;
};
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.

                     5、队列数据的管理

   当前队列数据的管理，本质上是队列描述信息的管理，描述当前服务器上有哪些队列。
   界说队列描述数据类

队列名称
是否持久化标志
是否独占标志
是否⾃动删除标志
其他参数

界说队列数据持久化类(数据持久化的sqlite3数据库中)

创建/删除队列数据表
新增队列数据
移除队列数据
查询全队伍列数据

界说队列数据管理类

创建队列，并添加管理(存在则OK，不存在则创建)
删除队列
获取指定队列
获取全队伍列
判定指定队列是否存在
获取队列数目
销毁全队伍列数据

登录后复制

#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/logger.hpp"
#include <unordered_map>
#include <memory>
#include <cassert>
#include <mutex>
namespace lsxmq
{
// 1.定义消息队列类
struct MessageQueue
{
using ptr = std::shared_ptr<MessageQueue>;
std::string name; // 队列名称
bool durable; // 是否持久化
bool exclusive; // 是否独占
bool autoDel; // 是否自动删除
std::unordered_map<std::string, std::string> args; // 其它参数
MessageQueue() {}
MessageQueue(const std::string &qname, bool qdurable, bool qexclusive, bool qautoDel, std::unordered_map<std::string, std::string> qargs)
: name(qname), durable(qdurable), exclusive(qexclusive), autoDel(qautoDel), args(qargs)
{
}
// args存储的是键值对，在存储数据库的时候，会组织一个格式进行存储：key=value&key=value
// 将字符串反序列化为args
void setArgs(std::string strArgs)
{
std::vector<std::string> subStr;
StrHelper::split(strArgs, "&", subStr); // 分割字符串
for (auto &str : subStr) // 继续分割
{
int pos = str.find("=");
if (pos != std::string::npos) // 加入args
{
std::string first = str.substr(0, pos);
std::string second = str.substr(pos + 1, str.size());
args.insert(std::make_pair(first, second));
}
}
}
// 将args序列化为字符串
std::string getArgs()
{
std::string res;
for (auto &a : args)
{
res += a.first;
res += "=";
res += a.second;
res += "&";
}
res.pop_back();
return res;
}
};
// 2.定义消息队列数据持久化存储类--数据存储在sqlite数据库中
using MessageQueueMap = std::unordered_map<std::string, MessageQueue::ptr>;
class MessageQueueMapper
{
public:
MessageQueueMapper(const std::string &dbfile)
: _sqliteHelper(dbfile)
{
std::string path = FileHelper::parentDirectory(dbfile);
if (FileHelper(path).exists() == false) // 父目录不存在就创建
{
FileHelper::createDirectory(path);
}
assert(_sqliteHelper.open());
this->createTable();
}
void createTable() // 创建一张消息队列数据管理表
{
#define CREATE_TABLE "create table if not exists messageQueueTable(\
name varchar(32) primary key,\
durable int,\
exclusive int,\
autoDel int,\
args varchar(128));"
bool ret = _sqliteHelper.exec(CREATE_TABLE, nullptr, nullptr);
if (ret == false)
{
ELOG("消息队列数据库表创建失败");
abort();
}
}
void removeTable() // 删除表
{
#define DROP_TABLE "drop table messageQueueTable"
bool ret = _sqliteHelper.exec(DROP_TABLE, nullptr, nullptr);
if (ret == false)
{
ELOG("消息队列数据库表删除失败");
abort();
}
}
bool insert(MessageQueue::ptr& mqp) // 向数据库插入一条消息队列数据
{
#define INSERT "insert into messageQueueTable values('%s',%d,%d,%d,'%s');"
char sql[256] = {0};
sprintf(sql, INSERT, mqp->name.c_str(), mqp->durable, mqp->exclusive, mqp->autoDel, mqp->getArgs().c_str());
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
bool erase(const std::string &qname) // 从数据库中删除一条消息队列数据
{
#define ERASE "delete from messageQueueTable where name='%s';"
char sql[256] = {0};
sprintf(sql, ERASE, qname.c_str());
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
MessageQueueMap recovery() // 获取数据库中所有的消息队列数据(恢复)
{
#define SELECT_ALL "select name,durable,exclusive,autoDel,args from messageQueueTable"
MessageQueueMap result;
bool ret = _sqliteHelper.exec(SELECT_ALL, sqliteCallback, &result);
if (ret == false)
{
ELOG("恢复消息队列数据失败");
abort();
}
return result;
}
~MessageQueueMapper()
{
_sqliteHelper.close();
}
private:
static int sqliteCallback(void *arg, int col, char **row, char **fields)
{
MessageQueueMap *result = (MessageQueueMap *)arg;
auto mqp = std::make_shared<MessageQueue>();
mqp->name = row[0];
mqp->durable = (bool)std::stoi(row[2]);
mqp->exclusive = (bool)std::stoi(row[2]);
mqp->autoDel = (bool)std::stoi(row[3]);
if (row[4])
mqp->setArgs(row[4]); // 其它参数可能不存在
result->insert(std::make_pair(mqp->name, mqp));
return 0;
}
private:
SqliteHelper _sqliteHelper;
};
// 3.定义消息队列数据管理类
class MessageQueueManager
{
public:
using ptr = std::shared_ptr<MessageQueueManager>;
MessageQueueManager(const std::string dbfile)
: _mapper(dbfile)
{
_messageQueues = _mapper.recovery(); // 恢复数据
}
bool declareMessageQueue(const std::string &name, bool durable,
bool exclusive, bool autoDel, std::unordered_map<std::string, std::string> args) // 声明消息队列
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
auto it = _messageQueues.find(name);
if (it != _messageQueues.end()) // 已经存在了
{
return true;
}
auto mqp = std::make_shared<MessageQueue>(name, durable,exclusive, autoDel, args);
if (durable == true) // 需要持久化
{
bool ret = _mapper.insert(mqp);
if (ret == false)
return false;
}
_messageQueues.insert(std::make_pair(name, mqp));
return true;
}
bool deleteMessageQueue(const std::string &name) // 删除消息队列
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
auto it = _messageQueues.find(name);
if (it == _messageQueues.end()) // 不存在
{
return true;
}
if (it->second->durable == true) // 从数据库中删除
{
bool ret = _mapper.erase(it->second->name);
if (ret == false)
return false;
}
_messageQueues.erase(name);
return true;
}
MessageQueue::ptr selectMessageQueue(const std::string &name) // 获取指定消息队列
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
auto it = _messageQueues.find(name);
if (it == _messageQueues.end()) // 不存在
{
return MessageQueue::ptr();
}
return it->second;
}
MessageQueueMap getall()//获取所有消息队列
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
return _messageQueues;
}
bool exists(const std::string &name) // 指定交换机是否存在
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
auto it = _messageQueues.find(name);
return it != _messageQueues.end();
}
void clear() // 清理数据
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
_mapper.removeTable(); // 先删磁盘内容
_messageQueues.clear(); // 再删内存内容
}
size_t size()
{
std::unique_lock<std::mutex> lock(_mutex); // 先加锁
return _messageQueues.size();
}
private:
std::mutex _mutex; // 多线程使用，注意线程安全
MessageQueueMapper _mapper;
MessageQueueMap _messageQueues;
};
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.
175.
176.
177.
178.
179.
180.
181.
182.
183.
184.
185.
186.
187.
188.
189.
190.
191.
192.
193.
194.
195.
196.
197.
198.
199.
200.
201.
202.
203.
204.
205.
206.
207.
208.
209.
210.
211.
212.
213.
214.
215.
216.
217.
218.
219.
220.
221.
222.
223.
224.
225.
226.
227.
228.
229.
230.

                     6、绑定信息(交换机-消息队列)管理

   绑定信息，本质上就是⼀个交换构造联了哪些队列的描述。
   界说绑定信息类

交换机名称
队列名称
binding_key(分发匹配规则-决定了哪些数据能被交换机放⼊队列)

界说绑定信息数据持久化类(数据持久化的sqlite3数据库中)

创建/删除绑定信息数据表
新增绑定信息数据
移除指定绑定信息数据
移除指定交换机相干绑定信息数据：移除交换机的时候会被调⽤
移除指定队列相干绑定信息数据：移除队列的时候会被调⽤
查询全部绑定信息数据：⽤于重启服务器时进⾏汗青数据恢复

界说绑定信息数据管理类

创建绑定信息，并添加管理(存在则OK，不存在则创建)
解除指定的绑定信息
删除指定队列的全部绑定信息
删除交换机相干的全部绑定信息
获取交换机相干的全部绑定信息：交换机收到消息后，必要分发给⾃⼰关联的队列
判定指定绑定信息是否存在
获取当前绑定信息数目
销毁全部绑定信息数据

登录后复制

#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/logger.hpp"
#include <unordered_map>
#include <memory>
#include <cassert>
#include <mutex>
namespace lsxmq
{
//1.定义绑定类
struct Binding
{
using ptr=std::shared_ptr<Binding>;
std::string exchangeName;//交换机名称
std::string messageQueueName;//消息队列名称
std::string binding_key;
bool durable;//是否持久化
Binding(){}
Binding(const std::string& ename,const std::string& qname,const std::string& bdkey,bool bdurable)
:exchangeName(ename),messageQueueName(qname),binding_key(bdkey),durable(bdurable)
{}
};
//2.定义持久化管理绑定类
//交换机名称-->多个消息队列
//消息队列-->多个绑定信息
using MessageQueueMap=std::unordered_map<std::string,Binding::ptr>;
using BindingMap=std::unordered_map<std::string,MessageQueueMap>;
class BindingMapper
{
public:
BindingMapper(const std::string dbfile)
:_sqliteHelper(dbfile)
{
std::string path = FileHelper::parentDirectory(dbfile);
if (FileHelper(path).exists() == false) // 父目录不存在就创建
{
FileHelper::createDirectory(path);
}
assert(_sqliteHelper.open());
this->createTable();
}
void createTable()//创建表
{
#define CREATE_TABLE "create table if not exists bindingTable(\
exchangeName varchar(32),\
messageQueueName varchar(32),\
binding_key varchar(128),\
durable int);"
bool ret = _sqliteHelper.exec(CREATE_TABLE, nullptr, nullptr);
if (ret == false)
{
ELOG("绑定数据库表创建失败");
abort();
}
}
void removeTable()//删除表
{
#define DROP_TABLE "drop table bindingTable"
bool ret = _sqliteHelper.exec(DROP_TABLE, nullptr, nullptr);
if (ret == false)
{
ELOG("绑定数据库表删除失败");
abort();
}
}
bool insert(Binding::ptr& bdp)//插入绑定数据
{
#define INSERT "insert into bindingTable values('%s','%s','%s',%d);"
char sql[256] = {0};
sprintf(sql, INSERT, bdp->exchangeName.c_str(),bdp->messageQueueName.c_str(),bdp->binding_key.c_str(),bdp->durable);
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
bool erase(const std::string& ename,const std::string qname)//删除绑定数据数据
{
#define ERASE "delete from bindingTable where exchangeName='%s' and messageQueueName='%s';"
char sql[256] = {0};
sprintf(sql, ERASE, ename.c_str(),qname.c_str());
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
bool removeExchangeBinding(const std::string& ename)//删除交换机的绑定数据
{
#define ERASE_EXCHANGE "delete from bindingTable where exchangeName='%s';"
char sql[256] = {0};
sprintf(sql, ERASE_EXCHANGE, ename.c_str());
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
bool removeMessageQueueBinding(const std::string& qname)//删除消息队列的绑定数据
{
#define ERASE_MESSAGEQUEUE "delete from bindingTable where messageQueueName='%s';"
char sql[256] = {0};
sprintf(sql, ERASE_MESSAGEQUEUE, qname.c_str());
return _sqliteHelper.exec(sql, nullptr, nullptr);
}
BindingMap recovery()//恢复数据
{
#define SELECT_ALL "select exchangeName,messageQueueName,binding_key,durable from bindingTable"
BindingMap result;
bool ret = _sqliteHelper.exec(SELECT_ALL, sqliteCallback, &result);
if (ret == false)
{
ELOG("恢复绑定数据失败");
abort();
}
return result;
}
~BindingMapper()
{
_sqliteHelper.close();
}
private:
static int sqliteCallback(void *arg, int col, char **row, char **fields)
{
BindingMap *result = (BindingMap *)arg;
auto bdp = std::make_shared<Binding>(row[0],row[1],row[2],row[3]);
//ename->(qname->binding)
//为了防止交换机相关的绑定信息已经存在，因此不能直接创建队列映射，进行添加，这样会覆盖历史数据
//因此得先获取交换机对应的映射对象，往里边添加数据
//但是，若这时候没有交换机对应的映射信息，因此这里的获取要使用引用(会保证不存在则自动创建)
//1.用ename拿出MessageQueueMap
MessageQueueMap& mqm=(*result)[row[0]];//已经有了就拿出来，没有就创建
mqm.insert(std::make_pair(row[1],bdp));
return 0;
}
private:
SqliteHelper _sqliteHelper;
};
//3.定义绑定内存管理类
class BindingManager
{
public:
using ptr=std::shared_ptr<BindingManager>;
BindingManager(const std::string& dbfile)
:_mapper(dbfile)
{
_bindings=_mapper.recovery();
}
bool bind(const std::string& ename,const std::string& qname,const std::string& bdkey,bool durable)//建立绑定关系
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
//先确定改绑定是否已存在
auto it=_bindings.find(ename);
if(it!=_bindings.end()&&it->second.find(qname)!=it->second.end())
{
return true;
}
Binding::ptr bdp=std::make_shared<Binding>(ename,qname,bdkey,durable);
//绑定信息是否需要持久化，取决于什么？交换机数据是持久化的，以及队列数据也是持久化的。
if(durable)//需要持久化
{
bool ret=_mapper.insert(bdp);
if(ret==false) return false;
}
auto& mqm=_bindings[ename];
mqm.insert(std::make_pair(qname,bdp));
return true;
}
bool unbind(const std::string& ename,const std::string& qname)//解除绑定关系
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
//先确定绑定是否不存在
auto it=_bindings.find(ename);
if(it==_bindings.end())//通过ename找不到MessageQueueMap
{
return true;
}
auto it2=it->second.find(qname);
if(it2==it->second.end())//通过qname找不到binding
{
return true;
}
//现在确定了要解除的绑定是存在的
if(it2->second->durable==true)//从数据库里删除
{
bool ret=_mapper.erase(ename,qname);
if(ret==false) return false;
}
//从内存中删除
it->second.erase(qname);
if(it->second.empty())
{
_bindings.erase(ename);
}
return true;
}
bool deleteExchangeBinding(const std::string& ename)//删除与交换机相关的绑定
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
_bindings.erase(ename);
return _mapper.removeExchangeBinding(ename);
}
bool deleteMessageQueueBinding(const std::string& qname)//删除与消息队列相关的绑定
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
for(auto it=_bindings.begin();it!=_bindings.end();)
{
it->second.erase(qname);
if(it->second.empty())
{
it=_bindings.erase(it);
}
else
{
++it;
}
}
return _mapper.removeMessageQueueBinding(qname);
}
MessageQueueMap getExchangeBinding(const std::string& ename)//获取交换机绑定的所有消息队列
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
//该交换机是否存在
auto it=_bindings.find(ename);
if(it==_bindings.end())//该交换机不存在
{
return MessageQueueMap();
}
return it->second;
}
Binding::ptr getBinding(const std::string& ename,const std::string& qname)
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
//先确定绑定是否不存在
auto it=_bindings.find(ename);
if(it==_bindings.end())//通过ename找不到MessageQueueMap
{
return Binding::ptr();
}
auto it2=it->second.find(qname);
if(it2==it->second.end())//通过qname找不到binding
{
return Binding::ptr();
}
//现在确定了要解除的绑定是存在的
return it2->second;
}
bool exists(const std::string& ename,const std::string& qname)//绑定信息是否存在
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
auto it=_bindings.find(ename);
if(it!=_bindings.end()&&it->second.find(qname)!=it->second.end())
{
return true;
}
return false;
}
size_t size()//绑定数量
{
std::unique_lock<std::mutex> lock(_mutex);//先加锁
size_t ret=0;
for(auto& it:_bindings)
{
ret+=it.second.size();
}
return ret;
}
void clear()//清空数据
{
ILOG("清空数据");
std::unique_lock<std::mutex> lock(_mutex);//先加锁
_mapper.removeTable();
_bindings.clear();
}
private:
std::mutex _mutex;
BindingMapper _mapper;
BindingMap _bindings;
};
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.
175.
176.
177.
178.
179.
180.
181.
182.
183.
184.
185.
186.
187.
188.
189.
190.
191.
192.
193.
194.
195.
196.
197.
198.
199.
200.
201.
202.
203.
204.
205.
206.
207.
208.
209.
210.
211.
212.
213.
214.
215.
216.
217.
218.
219.
220.
221.
222.
223.
224.
225.
226.
227.
228.
229.
230.
231.
232.
233.
234.
235.
236.
237.
238.
239.
240.
241.
242.
243.
244.
245.
246.
247.
248.
249.
250.
251.
252.
253.
254.
255.
256.
257.
258.
259.
260.
261.
262.
263.
264.
265.
266.
267.
268.
269.
270.
271.
272.
273.

                     7、队列消息管理

   文件数据管理

   消息的要素(决定了消息的数据结构)：

网络传输的消息要素：消息id、消息routing_key、消息的投递模式
服务器上的消息管理所需的额外要素：最主要的就是持久化原理

消息有用标志位：这个字段必要随着消息的持久化内容一起持久化。每一条消息都可能要进行持久化存储，等到推送给客户端就会删除掉，然而每一次删除一条消息就重写一次文件，效率太低下。如果设置了有用标志位，每次只必要将这个有用标志位对应的数据给修改为无效即可。
消息的垃圾回收机制：为了避免文件中消息存储越来越多(许多都是无效消息)，因此文件中消息的垃圾回收机制就很有必要了。
消息的现实存储位置(相对于文件起始位置的偏移量)：当要删除某条消息时，必要重写覆盖这条消息在文件的对应位置(将有用标志位设为无效)，这时候就必要可以或许找到这条消息
消息的长度：当恢复汗青消息以及读取消息内容的时候，必要解决粘包标题。

消息的持久化管理：

以队列为单元进行消息的持久化管理

当消息文件垃圾回收时，必要重新加载全部有用消息，重新天生新的数据文件，但是天生新的数据文件后，消息的存储位置发生了变化，这时候必要更新内存中的数据，这时候就必要对全部的队列数据进行加锁，然后进行更新---锁辩说频仍，效率低。因此，如果每个队列都有本身独立的数据文件，则每次只必要对操作的队列数据进行加锁即可。
为什么不使用数据库存储？有些消息较大，不得当数据库，其次消息的持久化主要是为了备份，而不是为了查询，因此直接使用平常文件进行存储。
既然是数据存储在文件中，那必然就会有数据格式的要求。4字节长度|数据|4字节长度|数据...... 。通过四字节长度描述消息现实存储长度，就可以解决粘包标题。

向外提供的操作：

消息文件的创建与删除
消息的新增持久化/删除消息的持久化(并不是真正的删除，只是将标志位置为无效)
汗青数据恢复/垃圾回收

什么环境下必要垃圾回收？因为每次删除数据都不是真正的删除，因此文件中的数据会越来越多，但是也不是每次删除都必要垃圾回收。当文件中有用消息凌驾2000条，且其中有用消息比例低于50%。
回收头脑：加载文件中全部有用消息，删除原文件，天生新的数据文件，将数据写入(存在风险，不采用)。加载文件中全部的有用消息，先写到一个临时文件中，然后再去删除原文件，将临时文件名称改为原文件名称。返回全部的有用消息(每条消息中都记录当前新的存储位置---用于更新内存中数据的内容)

要管理的数据：

队列名
根据队列名天生数据文件名称：队列名.mqd
根据队列名天生临时文件名称：队列名.mqd.tmp

   内存数据管理

   以队列为单元进行管理：

如果内存中的全部消息团体进行管理，则在进行垃圾回收以及恢复汗青消息上就会变得麻烦，因此每个队列都有一个消息数据的管理结构，终极向外提供一个总体的消息管理类。
队列消息管理：

构造对象时：创建/打开队列数据文件，恢复队列汗青消息数据
新增消息/确认消息(删除)
垃圾回收：但持久化数据总量凌驾2000且有用比例低于50%则进行垃圾回收
获取队首消息
删除队列全部消息
获取待推送消息数目
获取待确认消息数目
获取持久化消息数目

要管理的数据：

持久化的管理句柄
待推送消息链表：以头插尾删的头脑实现队列功能
持久化消息的hashmap：垃圾回收后，必要更新消息数据(现实存储位置)
待确认消息的hashmap：一条消息被推送给客户端，并不会立即真正删除，而是等到被确认后才会删除。一条消息被推送给客户端后，取出待推送链表，参加到待确认hashmap中，等到确认后再删除。
持久化文件中有用消息数目
持久化文件中总体消息数目：可以盘算出文件中的有用消息比例(根据比例和总量决定是否进行垃圾回收)
队列名称

实现一个对外的总体消息管理类

管理的时候每一个队列的消息
管理的成员：

互斥锁
每个队列的消息管理句柄：队列名称&队列消息管理句柄的hash表

提供的操作

初始化队列的消息管理句柄：创建队列的时候调用
销毁队列的消息管理句柄：删除队列的时候调用
向队列新增消息
获取队列队首消息
对队列进行消息确认
获取队列消息数目：可获取消息数目、持久化消息数目、有用待确认消息数目、总的持久化消息数目
恢复队列汗青消息

登录后复制

#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/logger.hpp"
#include "../mqcomm/message.pb.h"
#include <unordered_map>
#include <memory>
#include <cassert>
#include <list>
#include <mutex>
namespace lsxmq
{
using MessagePtr = std::shared_ptr<lsxmq::Message>;
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
#define NOTVALID "0"
#define ISVALID "1"
class MessageMapper
{
public:
MessageMapper(std::string &basedir, const std::string &qname)
: _qname(qname)
{
if (basedir.back() != '/')
{
basedir.push_back('/'); // 如果没有就往末尾加一个/
}
_datafile = basedir + qname + DATAFILE_SUBFIX;
_tmpfile = basedir + qname + TMPFILE_SUBFIX;
assert(FileHelper::createDirectory(basedir)); // 创建基本目录
createMessageFile();
}
bool createMessageFile() // 创建消息文件
{
if (FileHelper(_datafile).exists() == true)
{
return true;
}
if (FileHelper::createFile(_datafile) == false)
{
ELOG("创建消息文件失败");
return false;
}
return true;
}
bool removeMessageFile() // 删除消息文件
{
if (FileHelper(_datafile).exists() == false)
{
return true;
}
if (FileHelper::removeFile(_datafile) == false)
{
ELOG("删除消息文件失败");
return false;
}
return true;
}
bool insert(MessagePtr &ptr) // 新增消息
{
return insert(_datafile, ptr); // 向data文件写入内容
}
bool erase(MessagePtr &ptr) // 删除消息
{
// 1.将message的有效标志置为0
ptr->mutable_payload()->set_valid(NOTVALID);
// 2.进行序列化
std::string body = ptr->payload().SerializeAsString();
if (body.size() != ptr->len())
{
ELOG("新数据与原数据长度不一致")
return false;
}
// 3.将序列化的message覆盖文件中原本的message
FileHelper dataFileHelper(_datafile);
if (dataFileHelper.write(body.c_str(), ptr->offset(), body.size()) == false)
{
ELOG("消息重新写入文件失败");
return false;
}
return true;
}
std::list<MessagePtr> gc() // 垃圾回收
{
bool ret = true;
std::list<MessagePtr> result;
// 1.加载数据
ret = load(result);
if (ret == false)
{
ELOG("加载数据失败");
return result;
}
// 2.将数据写入到临时文件
FileHelper::createFile(_tmpfile);
FileHelper tmpDataHelper(_tmpfile);
for (auto &mp : result)
{
// 写入
ret = this->insert(_tmpfile, mp);
if (ret == false)
{
ELOG("消息写入临时文件失败");
return result;
}
}
// 3.删除原文件
ret = removeMessageFile();
if (ret == false)
{
ELOG("删除消息文件失败");
return result;
}
// 4.将原文件改为消息文件
ret = tmpDataHelper.rename(_datafile);
if (ret == false)
{
ELOG("修改临时文件名称失败");
return result;
}
return result;
}
private:
bool load(std::list<MessagePtr> &mplist) // 将文件中的消息加载进内存
{
size_t offset = 0;
FileHelper dataFileHelper(_datafile);
size_t fsize = dataFileHelper.size();
bool ret = true;
while (offset < fsize)
{
// 1.前四位为数据长度
size_t msize = 0;
ret = dataFileHelper.read((char *)&msize, offset, sizeof(size_t));
if (ret == false)
{
ELOG("读取信息长度失败");
return false;
}
// 2.开始读取数据
offset += sizeof(size_t);
std::string body(msize,'\0');
ret = dataFileHelper.read(&body[0], offset, msize);
if (ret == false)
{
ELOG("读取信息内容失败");
return false;
}
// 3.反序列化
MessagePtr mp = std::make_shared<Message>();
mp->mutable_payload()->ParseFromString(body);
// 4.剔除无效信息
if (mp->payload().valid() == NOTVALID)
{
offset += msize;
continue;
}
// 4.保存结果
mplist.push_back(mp);
offset += msize;
}
return true;
}
bool insert(const std::string filename, MessagePtr &ptr)
{
// 1.对消息进行序列化
std::string body = ptr->payload().SerializeAsString();
MessagePtr tmp=std::make_shared<Message>();
tmp->mutable_payload()->ParseFromString(body);
// 2.获取文件大小
FileHelper FileHelper(filename);
size_t fsize = FileHelper.size();
size_t msize = body.size();
// 3.先写入8字节长度
if (FileHelper.write((char*)&msize, fsize, sizeof(size_t)) == false)
{
ELOG("消息长度写入文件失败");
return false;
}
//4.在写入指定长度内容
if (FileHelper.write(body.c_str(), fsize+sizeof(size_t), msize) == false)
{
ELOG("消息内容写入文件失败");
return false;
}
// 4.更新message的实际存储位置
ptr->set_offset(fsize+sizeof(size_t));
ptr->set_len(msize);
return true;
}
private:
std::string _qname;
std::string _datafile;
std::string _tmpfile;
};
class QueueMessage
{
public:
using ptr = std::shared_ptr<QueueMessage>;
QueueMessage(std::string &basedir, const std::string &qname)
: _mapper(basedir, qname), _validCount(0), _totalCount(0), _qname(qname)
{}
void recovery() // 恢复历史消息
{
std::unique_lock<std::mutex> lock(_mutex);
_pushMessages = _mapper.gc();
for (auto &mp : _pushMessages)
{
_durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
}
_totalCount = _validCount = _durableMessages.size();
}
bool insert(BasicProperties *bp, const std::string& body, DeliveryMode deliveryMode) // 新增消息
{
// 1.构造message对象
MessagePtr mp = std::make_shared<Message>();
if (bp != nullptr)
{
mp->mutable_payload()->mutable_properties()->set_id(bp->id());
mp->mutable_payload()->mutable_properties()->set_deliverymode(bp->deliverymode());
mp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
mp->mutable_payload()->set_body(body);
mp->mutable_payload()->set_valid(ISVALID);
}
else
{
mp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
mp->mutable_payload()->mutable_properties()->set_deliverymode(deliveryMode);
mp->mutable_payload()->mutable_properties()->set_routing_key("");
mp->mutable_payload()->set_body(body);
mp->mutable_payload()->set_valid(ISVALID);
}
// 2.判断是否需要持久化
std::unique_lock<std::mutex> lock(_mutex);
bool ret = true;
if (mp->payload().properties().deliverymode() == DeliveryMode::DURABLE) // 需要
{
ret = _mapper.insert(mp);
if (ret == false)
{
ELOG("%s:消息持久化失败", body.c_str());
return false;
}
_durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
++_validCount;
++_totalCount;
}
// 3.在内存中管理
_pushMessages.push_back(mp);
return true;
}
MessagePtr front() // 获取队首消息
{
std::unique_lock<std::mutex> lock(_mutex);
if(_pushMessages.size()==0)//没有消息能拿出来
{
return MessagePtr();
}
// 获取队首消息
MessagePtr mp = _pushMessages.front();
// 将消息从待推送队列加到待确认hashmap中
_pushMessages.pop_front();
_waitAckMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
return mp;
}
bool erase(const std::string &id) // 确认消息,每次删除一条消息后确认是否要进行垃圾回收
{
std::unique_lock<std::mutex> lock(_mutex);
// 1.从待确认消息中找到这条消息
auto it = _waitAckMessages.find(id);
if (it == _waitAckMessages.end()) // 找不到
{
return true;
}
// 2.根据这条消息的投递模式决定是否从持久化中删除
bool ret = true;
if (it->second->payload().properties().deliverymode() == true) // 为持久化
{
// 3.从持久化中删除
ret = _mapper.erase(it->second);
if (ret == false)
{
ELOG("从文件中删除id=%s的消息失败", id.c_str());
return false;
}
_durableMessages.erase(id);
--_validCount;
// 进行垃圾回收
gc();
}
// 4.从内存中删除
_waitAckMessages.erase(id);
return true;
}
size_t pushCount()
{
std::unique_lock<std::mutex> lock(_mutex);
return _pushMessages.size();
}
size_t durableCount()
{
std::unique_lock<std::mutex> lock(_mutex);
return _durableMessages.size();
}
size_t waitAckCount()
{
std::unique_lock<std::mutex> lock(_mutex);
return _waitAckMessages.size();
}
size_t validCount()
{
std::unique_lock<std::mutex> lock(_mutex);
return _validCount;
}
size_t totalCount()
{
std::unique_lock<std::mutex> lock(_mutex);
return _totalCount;
}
void clear()
{
_mapper.removeMessageFile();
_pushMessages.clear();
_durableMessages.clear();
_waitAckMessages.clear();
_totalCount = _validCount = 0;
}
private:
bool NeedGC() // 判断是否需要进行垃圾回收
{
if (_totalCount > 2000 && _validCount * 10 / _totalCount < 5)
{
return true;
}
return false;
}
void gc() // 垃圾回收
{
if (NeedGC() == false)
return;
// 1.接收垃圾回收后的消息
std::list<MessagePtr> mplist = _mapper.gc();
// 2.修改内存中消息的数据
for (auto &mp : mplist)
{
// 从持久化消息中找到消息
auto it = _durableMessages.find(mp->payload().properties().id());
if (it == _durableMessages.end())
{
ILOG("出现了一条持久消息没有在内存中管理");
// 解决策略：重新添加
_pushMessages.push_back(mp);
_durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
continue;
}
// 更新消息与文件相关的数据
it->second->set_offset(mp->offset());
it->second->set_len(mp->len());
}
_totalCount = _validCount = _durableMessages.size();
}
private:
std::mutex _mutex;
MessageMapper _mapper;
std::list<MessagePtr> _pushMessages; // 待推送消息
std::unordered_map<std::string, MessagePtr> _durableMessages; // 持久化消息
std::unordered_map<std::string, MessagePtr> _waitAckMessages; // 待确认消息
size_t _validCount; // 持久化有效消息个数
size_t _totalCount; // 持久化总消息个数
std::string _qname;
};
class QueueMessageManager
{
public:
using ptr=std::shared_ptr<QueueMessageManager>;
QueueMessageManager(const std::string &basedir)
: _basedir(basedir)
{}
void clear()
{
std::unique_lock<std::mutex> lock(_mutex);
for(auto& a:_queueMessages)
{
a.second->clear();
}
}
void initQueueMessage(const std::string &qname) // 初始化队列
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否已存在
auto it = _queueMessages.find(qname);
if (it != _queueMessages.end()) // 已存在
{
return ;
}
qmp = std::make_shared<QueueMessage>(_basedir, qname);
_queueMessages.insert(std::make_pair(qname,qmp));
}
qmp->recovery();
}
void destroyQueueMessage(const std::string &qname)// 销毁队列
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
return ;
}
qmp=it->second;
_queueMessages.erase(qname);
}
qmp->clear();
}
bool insert(const std::string &qname, BasicProperties *bp, const std::string& body, DeliveryMode deliveryMode) // 向队列新增消息
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s新增消息失败，因为该队列不存在",qname.c_str());
return false;
}
qmp=it->second;
}
qmp->insert(bp,body,deliveryMode);
return true;
}
MessagePtr front(const std::string &qname)// 获取队首消息
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("从队列%s获取队首消息失败，因为该队列不存在",qname.c_str());
return MessagePtr();
}
qmp=it->second;
}
return qmp->front();
}
bool ack(const std::string &qname, const std::string &id)// 确认消息
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s确认消息失败，因为该队列不存在",qname.c_str());
return false;
}
qmp=it->second;
}
qmp->erase(id);
return true;
}
size_t pushCount(const std::string& qname)
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s获取待发送消息数量失败，因为该队列不存在",qname.c_str());
return 0;
}
qmp=it->second;
}
return qmp->pushCount();
}
size_t durableCount(const std::string& qname)
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s获取持久化消息数量失败，因为该队列不存在",qname.c_str());
return 0;
}
qmp=it->second;
}
return qmp->durableCount();
}
size_t waitAckCount(const std::string& qname)
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s获取等待确认消息数量失败，因为该队列不存在",qname.c_str());
return 0;
}
qmp=it->second;
}
return qmp->waitAckCount();
}
size_t validCount(const std::string& qname)
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s获取有效持久化消息数量失败，因为该队列不存在",qname.c_str());
return 0;
}
qmp=it->second;
}
return qmp->validCount();
}
size_t totalCount(const std::string& qname)
{
QueueMessage::ptr qmp;
{
std::unique_lock<std::mutex> lock(_mutex);
// 该队列是否不存在
auto it = _queueMessages.find(qname);
if (it == _queueMessages.end()) // 不存在
{
ELOG("向队列%s获取全部持久化消息数量失败，因为该队列不存在",qname.c_str());
return 0;
}
qmp=it->second;
}
return qmp->totalCount();
}
private:
std::mutex _mutex;
std::unordered_map<std::string, QueueMessage::ptr> _queueMessages;
std::string _basedir;
};
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.
175.
176.
177.
178.
179.
180.
181.
182.
183.
184.
185.
186.
187.
188.
189.
190.
191.
192.
193.
194.
195.
196.
197.
198.
199.
200.
201.
202.
203.
204.
205.
206.
207.
208.
209.
210.
211.
212.
213.
214.
215.
216.
217.
218.
219.
220.
221.
222.
223.
224.
225.
226.
227.
228.
229.
230.
231.
232.
233.
234.
235.
236.
237.
238.
239.
240.
241.
242.
243.
244.
245.
246.
247.
248.
249.
250.
251.
252.
253.
254.
255.
256.
257.
258.
259.
260.
261.
262.
263.
264.
265.
266.
267.
268.
269.
270.
271.
272.
273.
274.
275.
276.
277.
278.
279.
280.
281.
282.
283.
284.
285.
286.
287.
288.
289.
290.
291.
292.
293.
294.
295.
296.
297.
298.
299.
300.
301.
302.
303.
304.
305.
306.
307.
308.
309.
310.
311.
312.
313.
314.
315.
316.
317.
318.
319.
320.
321.
322.
323.
324.
325.
326.
327.
328.
329.
330.
331.
332.
333.
334.
335.
336.
337.
338.
339.
340.
341.
342.
343.
344.
345.
346.
347.
348.
349.
350.
351.
352.
353.
354.
355.
356.
357.
358.
359.
360.
361.
362.
363.
364.
365.
366.
367.
368.
369.
370.
371.
372.
373.
374.
375.
376.
377.
378.
379.
380.
381.
382.
383.
384.
385.
386.
387.
388.
389.
390.
391.
392.
393.
394.
395.
396.
397.
398.
399.
400.
401.
402.
403.
404.
405.
406.
407.
408.
409.
410.
411.
412.
413.
414.
415.
416.
417.
418.
419.
420.
421.
422.
423.
424.
425.
426.
427.
428.
429.
430.
431.
432.
433.
434.
435.
436.
437.
438.
439.
440.
441.
442.
443.
444.
445.
446.
447.
448.
449.
450.
451.
452.
453.
454.
455.
456.
457.
458.
459.
460.
461.
462.
463.
464.
465.
466.
467.
468.
469.
470.
471.
472.
473.
474.
475.
476.
477.
478.
479.
480.
481.
482.
483.
484.
485.
486.
487.
488.
489.
490.
491.
492.
493.
494.
495.
496.
497.
498.
499.
500.
501.
502.
503.
504.
505.
506.
507.
508.
509.
510.
511.
512.
513.
514.
515.
516.
517.
518.
519.
520.
521.
522.
523.
524.
525.
526.
527.
528.
529.
530.
531.
532.
533.
534.
535.
536.
537.
538.
539.
540.
541.
542.
543.
544.
545.
546.
547.
548.
549.
550.
551.
552.
553.
554.
555.
556.
557.
558.
559.

                     8、假造机管理

   假造机模块是对上述三个数据管理模块的整合，并基于数据之间的关联关系进⾏团结操作。
   界说假造机类包含以下成员：

交换机数据管理模块句柄
队列数据管理模块句柄
绑定数据管理模块句柄
消息数据管理模块句柄

假造机包含操作：

提供声明交换机的功能(存在则OK，不存在则创建)
提供删除交换机的功能(删除交换机的同时删除关联绑定信息)
提供声明队列的功能(存在则OK，不存在则创建，创建的同时创建队列关联消息管理对象)
提供删除队列的功能(删除队列的同时删除关联绑定信息，删除关联消息管理对象及队列全部消息)
提供交换机-队列绑定的功能
提供交换机-队列解绑的功能
提供获取交换机相干的全部绑定信息功能
提供新增消息的功能
提供获取指定队列队⾸消息的功能
提供消息确认删除的功能

假造机管理操作：可以有多台假造机，项目中并没有实现。

增删查

登录后复制

#pragma once
#include "binding.hpp"
#include "exchange.hpp"
#include "message.hpp"
#include "messageQueue.hpp"
namespace lsxmq
{
class VirtualHost//虚拟机
{
public:
using ptr=std::shared_ptr<VirtualHost>;
VirtualHost(const std::string& hname,const std::string& dbfile,const std::string& basedir)
:_hostName(hname)
,_bmp(std::make_shared<BindingManager>(dbfile))
,_emp(std::make_shared<ExchangeManager>(dbfile))
,_qmmp(std::make_shared<QueueMessageManager>(basedir))
,_mqmp(std::make_shared<MessageQueueManager>(dbfile))
{
//恢复队列消息
//1.获取所有队列
MessageQueueMap mqmap=_mqmp->getAll();
//2.用队列名称初始化队列消息
for(auto& it:mqmap)
{
_qmmp->initQueueMessage(it.second->name);
}
}
bool declareExchange(const std::string& name,ExchangeType type
,bool durable,bool autoDel
,std::unordered_map<std::string,std::string> args)//声明交换机
{
return _emp->declareExchange(name,type,durable,autoDel,args);
}
bool deleteExchange(const std::string& ename)//删除交换机
{
bool ret=_emp->deleteExchange(ename);
if(ret==false)
{
return false;
}
//删除交换机对应的绑定
return _bmp->deleteExchangeBinding(ename);
}
bool existsExchange(const std::string& ename)
{
return _emp->exists(ename);
}
bool declareMessageQueue(const std::string &name, bool durable, bool exclusive
, bool autoDel, std::unordered_map<std::string, std::string> args)//声明消息队列
{
bool ret=_mqmp->declareMessageQueue(name,durable,exclusive,autoDel,args);
if(ret==false)
{
return false;
}
_qmmp->initQueueMessage(name);
return true;
}
bool deleteMessageQueue(const std::string &qname)//删除消息队列
{
auto ret=_mqmp->deleteMessageQueue(qname);
if(ret==false)
{
return false;
}
//删除消息队列对应的绑定
ret=_bmp->deleteMessageQueueBinding(qname);
if(ret==false)
{
return false;
}
//删除对应的队列消息
_qmmp->destroyQueueMessage(qname);
return true;
}
bool existsMessageQueue(const std::string& qname)
{
return _mqmp->exists(qname);
}
bool bind(const std::string& ename,const std::string& qname,const std::string& binding_key)//绑定
{
//交换机是否存在
Exchange::ptr ep=_emp->selectExchange(ename);
if(ep.get()==nullptr)
{
ELOG("%s与%s绑定失败，因为交换机不存在",ename.c_str(),qname.c_str());
return false;
}
//队列是否存在
MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
if(mqp.get()==nullptr)
{
ELOG("%s与%s绑定失败，因为队列不存在",ename.c_str(),qname.c_str());
return false;
}
//根据交换机与队列的持久化方式决定绑定是否持久化
return _bmp->bind(ename,qname,binding_key,ep->durable&&mqp->durable);
}
bool unbind(const std::string& ename,const std::string& qname)//解绑
{
//交换机是否存在
Exchange::ptr ep=_emp->selectExchange(ename);
if(ep.get()==nullptr)
{
ELOG("%s与%s解绑失败，因为交换机不存在",ename.c_str(),qname.c_str());
return false;
}
//队列是否存在
MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
if(mqp.get()==nullptr)
{
ELOG("%s与%s解绑失败，因为队列不存在",ename.c_str(),qname.c_str());
return false;
}
return _bmp->unbind(ename,qname);
}
MessageQueueBindingMap exchangeBindings(const std::string& ename)//获取一台交换机的所有绑定
{
return _bmp->getExchangeBinding(ename);
}
bool existsBinding(const std::string &ename, const std::string &qname)
{
return _bmp->exists(ename,qname);
}
bool basicPublish(const std::string &qname, BasicProperties *bp
, const std::string& body)//发布一条消息
{
MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
return _qmmp->insert(qname,bp,body,mqp->durable);
}
MessagePtr basicConsume(const std::string& qname)//获取一条消息
{
return _qmmp->front(qname);
}
bool basicAck(const std::string &qname, const std::string &id)//确认一条消息
{
return _qmmp->ack(qname,id);
}
void clear()
{
_bmp->clear();
_emp->clear();
_qmmp->clear();
_mqmp->clear();
}
private:
std::string _hostName;//虚拟机名称
BindingManager::ptr _bmp;//绑定管理句柄
ExchangeManager::ptr _emp;//交换机管理句柄
QueueMessageManager::ptr _qmmp;//队列消息管理句柄
MessageQueueManager::ptr _mqmp;//消息队列管理句柄
};
};

复制代码

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.

                     9、交换机路由模块

   交换路由规则取决要素：

交换机范例

广播交换：将消息发布到全部绑定的队列中
直接交换：用户发布消息的时候，消息中有一个routing_key，与哪个队列的binding_key相同，则匹配成功
主题交换：按照匹配规则匹配

路由交换模块：

功能：判定一个消息中的routing_key与队列的binding_key是否匹配成功。
提供功能：

判定routing_key是否合法：必须是a~z、A~Z、.、_构成
判定binding_key是否合法：必须是a~z、A~Z、*、#、.、_构成，其中*#是通配符，必须独立存在，*和#不能连续出现
进行routing_key与binding_key的路由匹配

   binding_key：
   是由数字字⺟下划线构成的, 并且使⽤ . 分成若⼲部门，并⽀持 * 和 # 通配符。
   例如： news.music.# ，这⽤于表⽰交换机绑定的当前队列是⼀个⽤于发布⾳乐消息的队列。

⽀持 * 和 # 两种通配符，但是 * # 只能作为 . 切分出来的独⽴部门, 不能和其他数字字⺟混⽤,

⽐如 a.*.b 是合法的， a.*a.b 是不合法的
* 可以匹配任意⼀个单词(注意是单词不是字⺟)
# 可以匹配零个或者多个任意单词(注意是单词不是字⺟)

注意事项： a.#.b

⼀个单词中不能既出现 * ⼜出现 #，也就是，⼀个单词中只能有⼀个通配符，且必须独⽴存在
#通配符双方不能出现其他通配符，因为 # 可以匹配任意多个任意单词，因此连续出现是没故意义的。

   routing_key：
   是由数据、字⺟和下划线构成，并且可以使⽤ . 分别成若⼲部门。
   例如： news.music.pop ，这⽤于表⽰当前发布的消息是⼀个流⾏⾳乐的消息。
   ⽐如，在进⾏队列绑定时，某队列的binding_key约定为：news.music.#表⽰这个队列⽤于发布⾳乐消息。⽽这时候客⼾端发布了⼀条消息，其中routing_key为：news.music.pop 则可以匹配成功，⽽，如果发布消息的routing_key为：news.sport.football，这时候就会匹配失败。
   匹配算法头脑：

对binding_key和routing_key进行单词分割
按照元素个数，界说出对应巨细+1的二维数组，并且将dp[0][0]位置置为1
使用routing_key的每个元素，与binding_key的每个元素进行比对，在二维数组中进行标志
标志规则：

当两个单词匹配成功时：从左上方继承结果，dp[j]=dp[i-1][j-1]
当遇到#通配符匹配成功时，不仅可以从左上方继承，还可以从左方继承结果dp[j]=dp[i-1][j-1] || dp[j-1]
当遇到#通配符匹配成功时，不仅可以从左上方、左方继承，还可以从上方继承结果dp[j]=dp[i-1][j-1] || dp[j-1] || dp[i-1][j]
当binding_key以#起始时，必要将#对应行的第0列置为1

                                 登录后复制
#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/message.pb.h"
namespace lsxmq
{
class Router
{
public:
      static bool routingKeyIsLegal(const std::string& routingKey)//判断routing_key是否合法
      {
         //只需要确定是否有非法字符即可
         //合法字符：a~z、A~Z、.、_
         for(auto& ch:routingKey)
         {
            if((ch>='a'&&ch<='z')
                  ||(ch>='A'&&ch<='Z')
                  ||(ch>='0'&&ch<='9')
                  ||ch=='_'||ch=='.')
            {
                  continue;
            }
            else
            {
                  return false;//不合法
            }
         }
         return true;
      }
      static bool bindingKeyIsLegal(const std::string& bindingKey)
      {
         //1.确定是否有非法字符即可
         //合法字符：a~z、A~Z、.、_、*、#
         for(auto& ch:bindingKey)
         {
            if((ch>='a'&&ch<='z')
                  ||(ch>='A'&&ch<='Z')
                  ||(ch>='0'&&ch<='9')
                  ||ch=='_'||ch=='.'
                  ||ch=='#'||ch=='*')
            {
                  continue;
            }
            else
            {
                  return false;//不合法
            }
         }
         //2.#和*必须独立存在(aa.a#.b这种就是非法)
         std::vector<std::string> subStr;
         StrHelper::split(bindingKey,".",subStr);
         for(auto& word:subStr)
         {
            if(word.size()>1
                  &&word.find('#')!=std::string::npos
                  &&word.find('*')!=std::string::npos)
            {
                  return false;
            }
         }
         //3.#和*不连续出现，#的前面没有#和*，*的前面没有#
         for(int i=1;i<subStr.size();++i)
         {
            if(subStr[i]=="#"&&subStr[i-1]=="#")
            {
                  return false;
            }
            if(subStr[i]=="#"&&subStr[i-1]=="*")
            {
                  return false;
            }
            if(subStr[i]=="*"&&subStr[i-1]=="#")
            {
                  return false;
            }
         }
         return true;
      }
      static bool route(const std::string& routingKey,const std::string& bindingKey,ExchangeType routeType)
      {
         if(routeType==DIRECT)//直接路由
         {
            return routingKey==bindingKey;
         }
         if(routeType==FANOUT)//广播路由
         {
            return true;
         }
         if(routeType==TOPIC)//主题路由
         {
            return topicRoute(routingKey,bindingKey);
         }
         ELOG("位置交换机类型：%d",routeType);
         return false;
      }
private:
      static bool topicRoute(const std::string& routingKey,const std::string& bindingKey)
      {
         //1.先分割字符串
         std::vector<std::string> subBinding;
         StrHelper::split(bindingKey,".",subBinding);
         std::vector<std::string> subRouting;
         StrHelper::split(routingKey,".",subRouting);
         //2.构造dp表
         int row=subBinding.size(),col=subRouting.size();//行列
         std::vector<std::vector<int>> dp(row+1,std::vector<int>(col+1,0));
         dp[0][0]=1;
         if(subBinding[0]=="#") dp[1][0]=1;//当binding_key以#起始时，需要将#对应行的第0列置为1
         //3.开始匹配
         for(int i=1;i<row+1;++i)
         {
            for(int j=1;j<col+1;++j)
            {
                  if(subBinding[i-1]=="#")//遇到#通配符匹配成功时,可以从左上方、左方、上方继承
                  {
                     dp[i][j]=dp[i-1][j-1]||dp[i-1][j]||dp[i][j-1];
                  }
                  else if(subBinding[i-1]=="*"||(subBinding[i-1]==subRouting[j-1]))
                  {
                     //当两个单词匹配成功时,从左上方继承结果
                     dp[i][j]=dp[i-1][j-1];
                  }
            }
         }
         return dp[row][col];
      }
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

127.

128.

                     10、队列消费者/订阅者管理



   界说消费者信息结构：


消费者表现

订阅的队列名称

一个消息的处理回调函数(实现的是当发布一条消息到队列，则选择消费者进行消费，怎样消费？对于服务端来说就是调用这个回调函数进行处理，其内部逻辑就是找到消费者对应的连接，然后将数据发送给消费者对应的客户端)

是否自动应答标志(一个消息被消费者消费后，若自动应答，则直接移除待确认消息，否则等候客户端回应)

   消费者的管理以信道为单元照旧以队列为单元？


以信道为单元：一个信道关闭时全部关联的消费者都要删除

以队列为单元：一个队列收到了一条消息，必要找到订购了队列消息的消费者进行推送

   消费者管理(以队列为单元进行管理)：


操作：



新增消费者：信道提供的服务式订阅队列消息的时候创建

删除消费者：取消订阅/信道关闭/连接关闭的时候删除

获取消费者：从队列中全部的消费者中按序取出一个消费者进行消息推送

判定队列消费者是否为空

判定指定消费者是否存在

清理队列全部消费者



元素



消费者管理结构：vector

轮转序号：一个队列可能会有多个消费者，但是一条消息只必要被一个消费者消费即可，因此采用RR轮转

互斥锁：保证线程安全

队列名称

   对消费者进行同一管理


初始化/删除队列的消费者信息结构(创建/删除队列的时候初始化)

向指定队列新增消费者(客户端订阅指定队列消息的时候)新增完成时返回消费者对象，让信道能管理

从指定队列移除消费者(客户端取消订阅的时候)

移除指定队列的全部消费者(队列被删除时销毁)删除消费者的队列管理单元对象

从指定队列获取一个消费者(轮询获取-消费者轮询消费起到负载平衡作用)

判定队列消费者是否为空

判定队列中指定消费者是否存在

清理消费者

                                 登录后复制
#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/logger.hpp"
#include "../mqcomm/message.pb.h"
namespace lsxmq
{
using ConsumerCallBack=std::function<void(const std::string&,BasicProperties*,const std::string&)>;
                                                   //消费者标识    //基本属性          //消息内容
struct Consumer
{
      using ptr=std::shared_ptr<Consumer>;
      Consumer(){}
      Consumer(const std::string& ctag,const std::string& cqname,bool cautoAck,const ConsumerCallBack& ccallback)
         :tag(ctag),qname(cqname),autoAck(cautoAck),callback(ccallback)
      {}
      std::string tag;//消费者标识
      std::string qname;//队列名称
      bool autoAck;//自动确认标志
      ConsumerCallBack callback;//消费回调函数
};
//以队列为单元管理消费者
class QueueConsumer
{
public:
      using ptr=std::shared_ptr<QueueConsumer>;
      QueueConsumer(const std::string& qname)
         :_qname(qname),_rrSeq(0)
      {}
      Consumer::ptr create(const std::string& tag,bool autoAck,const ConsumerCallBack& callback)//创建一个消费者
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         //2.判断该消费者是否已存在
         for(auto& con:_consumers)
         {
            if(con->tag==tag)
            {
                  return con;
            }
         }
         //3.构造消费者
         Consumer::ptr cp=std::make_shared<Consumer>(tag,_qname,autoAck,callback);
         //4.添加管理并返回消费者
         _consumers.push_back(cp);
         return cp;
      }
      void remove(const std::string& tag)//删除一个消费者
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         //2.查找消费者并删除
         for(auto it=_consumers.begin();it!=_consumers.end();++it)
         {
            if((*it)->tag==tag)
            {
                  _consumers.erase(it);
                  return ;
            }
         }
      }
      Consumer::ptr choose()//选择一个消费者
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         if(_consumers.size()==0)//没有消费者
         {
            return Consumer::ptr();
         }
         //2.获取选择的消费者序号
         size_t id=_rrSeq%_consumers.size();
         ++_rrSeq;
         //3.返回消费者
         return _consumers[id];
      }
      bool empty()//判断队列中是否没有消费者
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         return _consumers.empty();
      }
      bool exists(const std::string& tag)//判断消费者是否存在
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         //2.判断消费者是否存在
         for(auto& con:_consumers)
         {
            if(con->tag==tag)
            {
                  return true;
            }
         }
         return false;
      }
      void clear()//清空
      {
         //1.先加锁
         std::unique_lock<std::mutex> lock(_mutex);
         _consumers.clear();
      }
private:
      std::vector<Consumer::ptr> _consumers;
      size_t _rrSeq;
      std::mutex _mutex;
      std::string _qname;
};
class QueueConsumerManager
{
public:
      using ptr=std::shared_ptr<QueueConsumerManager>;
      QueueConsumerManager(){}
      void initQueueConsumer(const std::string& qname)//初始化队列
      {
         //1.加锁
         std::unique_lock<std::mutex> lock(_mutex);
         //2.是否已存在
         auto it=_queueConsumers.find(qname);
         if(it!=_queueConsumers.end())
         {
            return ;
         }
         //2.构造队列
         QueueConsumer::ptr qcp=std::make_shared<QueueConsumer>(qname);
         //3.添加管理
         _queueConsumers.insert(std::make_pair(qname,qcp));
      }
      void destroyQueueConsumer(const std::string& qname)//销毁队列
      {
         //1.加锁
         std::unique_lock<std::mutex> lock(_mutex);
         //2.是否不存在
         auto it=_queueConsumers.find(qname);
         if(it==_queueConsumers.end())
         {
            return ;
         }
         //3.删除队列消费者
         //it->second->clear();//不需要，智能指针指向的对象会自动释放
         _queueConsumers.erase(qname);
      }
      Consumer::ptr create(const std::string& qname,const std::string& tag
            ,bool autoAck,const ConsumerCallBack& callback)//往某个队列创建消费者
      {
         QueueConsumer::ptr qcp;
         {
               //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断队列是否不存在
            auto it=_queueConsumers.find(qname);
            if(it==_queueConsumers.end())
            {
                  ELOG("向队列%s添加消费者%s失败，因为该队列不存在",qname.c_str(),tag.c_str());
                  return Consumer::ptr();
            }
            qcp=it->second;
         }
         //3.通过队列消费者句柄创建消费者
         return qcp->create(tag,autoAck,callback);
      }
      void remove(const std::string& qname,const std::string& tag)//从某个队列中删除一个消费者
      {
         QueueConsumer::ptr qcp;
         {
               //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断队列是否不存在
            auto it=_queueConsumers.find(qname);
            if(it==_queueConsumers.end())
            {
                  ELOG("从队列%s删除消费者%s失败，因为该队列不存在",qname.c_str(),tag.c_str());
                  return ;
            }
            qcp=it->second;
         }
         //3.通过队列消费者句柄创建消费者
         qcp->remove(tag);
      }
      Consumer::ptr choose(const std::string& qname)//选择一个消费者
      {
         QueueConsumer::ptr qcp;
         {
               //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断队列是否不存在
            auto it=_queueConsumers.find(qname);
            if(it==_queueConsumers.end())
            {
                  ELOG("从队列%s获取消费者失败，因为该队列不存在",qname.c_str());
                  return Consumer::ptr();
            }
            qcp=it->second;
         }
         //3.通过队列消费者句柄创建消费者
         return qcp->choose();
      }
      bool empty(const std::string& qname)
      {
         QueueConsumer::ptr qcp;
         {
               //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断队列是否不存在
            auto it=_queueConsumers.find(qname);
            if(it==_queueConsumers.end())
            {
                  ELOG("从队列%s获取是否为空失败，因为该队列不存在",qname.c_str());
                  return false;
            }
            qcp=it->second;
         }
         //3.通过队列消费者句柄创建消费者
         return qcp->empty();

      }
      bool exists(const std::string& qname,const std::string& tag)
      {
         QueueConsumer::ptr qcp;
         {
               //1.加锁
            std::unique_lock<std::mutex> lock(_mutex);
            //2.判断队列是否不存在
            auto it=_queueConsumers.find(qname);
            if(it==_queueConsumers.end())
            {
                  ELOG("从队列%s获取消费者失败，因为该队列不存在",qname.c_str());
                  return false;
            }
            qcp=it->second;
         }
         //3.通过队列消费者句柄创建消费者
         return qcp->exists(tag);
      }
      void clear()
      {
         //1.加锁
         std::unique_lock<std::mutex> lock(_mutex);
         _queueConsumers.clear();
      }
private:
      std::mutex _mutex;
      std::unordered_map<std::string,QueueConsumer::ptr> _queueConsumers;
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

127.

128.

129.

130.

131.

132.

133.

134.

135.

136.

137.

138.

139.

140.

141.

142.

143.

144.

145.

146.

147.

148.

149.

150.

151.

152.

153.

154.

155.

156.

157.

158.

159.

160.

161.

162.

163.

164.

165.

166.

167.

168.

169.

170.

171.

172.

173.

174.

175.

176.

177.

178.

179.

180.

181.

182.

183.

184.

185.

186.

187.

188.

189.

190.

191.

192.

193.

194.

195.

196.

197.

198.

199.

200.

201.

202.

203.

204.

205.

206.

207.

208.

209.

210.

211.

212.

213.

214.

215.

216.

217.

218.

219.

220.

221.

222.

223.

224.

225.

226.

227.

228.

229.

230.

231.

232.

233.

234.

235.

236.

237.

238.

239.

240.

241.

242.

243.

244.

245.

246.

                     11、信道管理模块

   管理信息：


信道id：信道的唯一标识

信道关联的消费者：用于消费者信道在关闭时取消订阅，删除订阅者信息

信道关联的连接：用于向客户端发送数据(相应/推送消息)

protobuf协议的处理句柄：网络通讯前的协议处理

消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息

假造机句柄：交换机/队列/绑定/消息数据管理

工作线程池句柄：一条消息被发布到队列后，必要将消息推送给订阅了对应队列的消费者，过程由线程池完成。

   管理操作：


提供声明&删除交换机操作(删除交换机的同时删除交换构造联的绑定信息)

提供声明&删除队列操作(删除队列的同时，删除队列关联的绑定信息，消息，消费者信息)

提供绑定&解绑队列操作

提供订阅&取消订阅队列消息操作

提供发布&确认消息操作

   信道管理：


信道的增删查

   应用层协议设计：
                                 登录后复制
syntax= "proto3";
package lsxmq;
import "message.proto";
//信道的打开与关闭
message OpenChannelRequest  //打开信道
{
string rid = 1;  //请求id,对应不同的请求类型
string cid = 2;  //信道id
};
message CloseChannelRequest  //关闭信道
{
string rid = 1;
string cid = 2;
};
//交换机的声明与删除
message DeclareExchangeRequest  //声明交换机
{
string rid = 1;
string cid = 2;
string exchange_name = 3;  //交换机名称
ExchangeType exchange_type = 4;  //交换机类型
bool durable = 5;  //持久化
bool auto_delete = 6;  //自动删除
map<string, string> args = 7;  //其它参数
};
message DeleteExchangeRequest  //删除交换机
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
};
//消息队列的声明与删除
message DeclareMessageQueueRequest  //声明消息队列
{
string rid = 1;
string cid = 2;
string queue_name = 3;  //队列名称
bool exclusive = 4;  //是否独占
bool durable = 5;
bool auto_delete = 6;
map<string, string> args = 7;
};
message DeleteMessageQueueRequest  //删除消息队列
{
string rid = 1;
string cid = 2;
string queue_name = 3;
};
//绑定与解绑
message BindRequest  //绑定
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
string queue_name = 4;
string binding_key = 5;
};
message UnbindRequest  //解绑
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
string queue_name = 4;
};
//订阅与取消订阅
message BasicConsumeRequest  //订阅
{
string rid = 1;
string cid = 2;
string consumer_tag  =3;  //消费者标识
string queue_name = 4;
bool auto_ack = 5;  //是否自动确认
};
message BasicCancelRequest  //取消订阅
{
string rid = 1;
string cid = 2;
string consumer_tag = 3;
string queue_name = 4;
};
//发布与确认消息
message BasicPublishRequest //发布消息
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
string body = 4;  //消息内容
BasicProperties properties = 5;  //消息基本属性
};
message BasicAckRequest //确认消息
{
string rid = 1;
string cid = 2;
string queue_name = 3;
string mid = 4;
};
//消息的推送
message basicConsumeResponse {
string cid = 1;
string consumer_tag = 2;
string body = 3;
BasicProperties properties = 4;
};
//通用响应
message BasicCommonResponse
{
string rid = 1;
string cid = 2;
bool ok = 3;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

                     12、连接管理模块

   向⽤⼾提供⼀个⽤于实现⽹络通讯的Connection对象，从其内部可创建出粒度更轻的Channel对象，⽤于与客⼾端进⾏⽹络通讯。
   成员信息：


连接关联的信道管理句柄(实现信道的增删查)

连接关联的现实⽤于通讯的muduo::net::Connection连接

protobuf协议处理的句柄(ProtobufCodec对象)

消费者管理句柄

假造机句柄

异步⼯作线程池句柄

   连接操作：


提供创建Channel信道的操作

提供删除Channel信道的操作

   连接管理：


连接的增删查

                                 登录后复制
#pragma once
#include "channel.hpp"
namespace lsxmq
{
class Connection
{
public:
      using ptr=std::shared_ptr<Connection>;
      Connection(const muduo::net::TcpConnectionPtr &conn
         , const ProtobufCodecPtr &codec
         , const QueueConsumerManager::ptr &qcmp
         , const VirtualHost::ptr &vhp
         , const ThreadPool::ptr &pool)
         : _conn(conn)
         , _codec(codec)
         , _qcmp(qcmp)
         , _vhp(vhp)
         , _pool(pool)
         , _chmp(std::make_shared<ChannelManager>())
      {
         DLOG("new Connection: %p",this);
      }
      ~Connection()
      {
         DLOG("delete Connection: %p",this);
      }
      void openChannel(const OpenChannelRequestPtr &req)
      {
         // 1.判断信道是否已打开
         bool ret = _chmp->openChannel(req->cid(),_conn,_codec,_qcmp,_vhp,_pool);
         if(ret==false)
         {
            basicResponse(req->rid(),req->cid(),false);
            return;
         }
         //2.响应
         basicResponse(req->rid(),req->cid(),true);
      }
      void closeChannel(const CloseChannelRequestPtr &req)
      {
         _chmp->closeChannel(req->cid());
         basicResponse(req->rid(),req->cid(),true);
      }
      Channel::ptr getChannel(const std::string& cid)
      {
         return _chmp->getChannel(cid);
      }
private:
      void basicResponse(const std::string &rid, const std::string &cid, bool ok)
      {
         BasicCommonResponse rsp;
         rsp.set_rid(rid);
         rsp.set_cid(cid);
         rsp.set_ok(ok);
         _codec->send(_conn, rsp);
      }
private:
      muduo::net::TcpConnectionPtr _conn; // 关联的连接
      ProtobufCodecPtr _codec;          // protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
      QueueConsumerManager::ptr _qcmp; // 消费者管理句柄
      VirtualHost::ptr _vhp;             // 虚拟机管理句柄
      ThreadPool::ptr _pool;             // 线程池管理句柄
      ChannelManager::ptr _chmp;       // 信道管理句柄
};
class ConnectionManager
{
public:
      using ptr=std::shared_ptr<ConnectionManager>;
      ConnectionManager()
      {
         DLOG("new ConnectionManager: %p",this);
      }
      ~ConnectionManager()
      {
         DLOG("delete ConnectionManager: %p",this);
      }
      void newConnection(const muduo::net::TcpConnectionPtr &conn
         , const ProtobufCodecPtr &codec
         , const QueueConsumerManager::ptr &qcmp
         , const VirtualHost::ptr &vhp
         , const ThreadPool::ptr &pool)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //连接是否已存在
         auto it=_conns.find(conn);
         if(it!=_conns.end())
         {
            return;
         }
         //构建对象
         Connection::ptr cop=std::make_shared<Connection>(conn,codec,qcmp,vhp,pool);
         //添加管理
         _conns[conn]=cop;
      }

      void deleteConnection(const muduo::net::TcpConnectionPtr &conn)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //连接是否不存在
         auto it=_conns.find(conn);
         if(it==_conns.end())
         {
            return;
         }
         //删除
         _conns.erase(it);
      }
      Connection::ptr getConnection(const muduo::net::TcpConnectionPtr &conn)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //连接是否不存在
         auto it=_conns.find(conn);
         if(it==_conns.end())
         {
            return Connection::ptr();
         }
         return it->second;
      }
private:
      std::mutex _mutex;
      std::unordered_map<muduo::net::TcpConnectionPtr,Connection::ptr> _conns;
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

                     13、服务器模块

   服务器模块借助muduo网络库来实现


_server：Muduo库提供的⼀个通⽤TCP服务器，可以封装这个服务器进⾏TCP通讯

_baseloop：主事件循环器，⽤于相应IO事件和定时器事件，主loop主要是为了相应监听描述符的IO事件

_codec: ⼀个protobuf编解码器，在TCP服务器上设计了⼀层应⽤层协议，这个编解码器主要就是负责实现应⽤层协议的解析和封装，下边详细讲解

_dispatcher：⼀个消息分发器，当Socket接收到⼀个报⽂消息后，我必要按照消息的范例，即上⾯提到的typeName进⾏消息分发，会不不同范例的消息分发相对应的的处理函数中，下边详细讲解

_consumer: 服务器中的消费者信息管理句柄。

_threadpool: 异步⼯作线程池，主要⽤于队列消息的推送⼯作。

_connections: 连接管理句柄，管理当前服务器上的全部已经建⽴的通讯连接。

_virtual_host：服务器持有的假造主机。队列、交换机、绑定、消息等数据都是通过假造主机管理。

                                 登录后复制
#pragma once
#include "../mqthird/include/muduo/proto/codec.h"
#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/base/Logging.h"
#include "../mqthird/include/muduo/base/Mutex.h"
#include "../mqthird/include/muduo/net/EventLoop.h"
#include "../mqthird/include/muduo/net/TcpServer.h"
#include "connection.hpp"
namespace lsxmq
{
//typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
#define DBFILE "/meta.db"
#define HOSTNAME "lsxhost"
class Server
{
public:
      Server(uint16_t port,const std::string& basedir)
         : _server(&_loop, muduo::net::InetAddress(port)
         , "ProtobufServer", muduo::net::TcpServer::kReusePort)
         , _dispatcher(std::bind(&Server::onUnknownMessage
                  , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
         , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher
                  , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
         ,_consumerManager(std::make_shared<QueueConsumerManager>())
         ,_virtualHost(std::make_shared<VirtualHost>(HOSTNAME,basedir+DBFILE,basedir))
         ,_threadPool(std::make_shared<ThreadPool>())
         ,_connectionManager(std::make_shared<ConnectionManager>())
      {
         //需要初始化队列的消费者结构,因为消息队列是持久化的,构造时会自动加载,但消费者的队列结构不会
         //1.获取所有队列
         auto mqmap=_virtualHost->allQueue();
         for(auto& mqp:mqmap)
         {
            _consumerManager->initQueueConsumer(mqp.second->name);
         }
         // 注册业务处理请求函数
         _dispatcher.registerMessageCallback<OpenChannelRequest>(std::bind(&Server::onOpenChannel
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<CloseChannelRequest>(std::bind(&Server::onCloseChannel
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<DeclareExchangeRequest>(std::bind(&Server::onDeclareExchange
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<DeleteExchangeRequest>(std::bind(&Server::onDeleteExchange
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<DeclareMessageQueueRequest>(std::bind(&Server::onDeclareMessageQueue
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<DeleteMessageQueueRequest>(std::bind(&Server::onDeleteMessageQueue
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<BindRequest>(std::bind(&Server::onBind
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<UnbindRequest>(std::bind(&Server::onUnbind
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<BasicConsumeRequest>(std::bind(&Server::onBasicConsume
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<BasicCancelRequest>(std::bind(&Server::onBasicCancel
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<BasicPublishRequest>(std::bind(&Server::onBasicPublish
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<BasicAckRequest>(std::bind(&Server::onBasicAckRequest
            , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         //注册服务器的回调函数
         _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
         _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
      }
      void start()
      {
         _server.start();
         _loop.loop();
      }
private:
      //打开信道
      void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const OpenChannelRequestPtr &message, muduo::Timestamp)
      {
         //获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("打开信道失败，因为连接不存在");
            return ;
         }
         selfConnP->openChannel(message);
      }
      //关闭信道
      void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const CloseChannelRequestPtr &message, muduo::Timestamp)
      {
         //获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("关闭信道失败，因为连接不存在");
            return ;
         }
         selfConnP->closeChannel(message);
      }
      //声明交换机
      void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const DeclareExchangeRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("声明交换机失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("声明交换机失败，因为信道不存在");
            return ;
         }
         //声明交换机
         selfChannelP->declareExchange(message);
      }
      //删除交换机
      void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const DeleteExchangeRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("删除交换机失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("删除交换机失败，因为信道不存在");
            return ;
         }
         //删除交换机
         selfChannelP->deleteExchange(message);
      }
      //声明消息队列
      void onDeclareMessageQueue(const muduo::net::TcpConnectionPtr &conn, const DeclareMessageQueueRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("声明消息队列失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("声明消息队列失败，因为信道不存在");
            return ;
         }
         //声明消息队列
         selfChannelP->declareMessageQueue(message);
      }
      //删除消息队列
      void onDeleteMessageQueue(const muduo::net::TcpConnectionPtr &conn, const DeleteMessageQueueRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("删除消息队列失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("删除消息队列失败，因为信道不存在");
            return ;
         }
         //删除消息队列
         selfChannelP->deleteMessageQueue(message);
      }
      //绑定
      void onBind(const muduo::net::TcpConnectionPtr &conn, const BindRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("绑定失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("绑定失败，因为信道不存在");
            return ;
         }
         //绑定
         selfChannelP->bind(message);
      }
      //解绑
      void onUnbind(const muduo::net::TcpConnectionPtr &conn, const UnbindRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("解绑失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("解绑失败，因为信道不存在");
            return ;
         }
         //解绑
         selfChannelP->unbind(message);
      }
      //订阅
      void onBasicConsume(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("订阅失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("订阅失败，因为信道不存在");
            return ;
         }
         //订阅
         selfChannelP->basicConsume(message);
      }
      //取消订阅
      void onBasicCancel(const muduo::net::TcpConnectionPtr &conn, const BasicCancelRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("取消订阅失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("取消订阅失败，因为信道不存在");
            return ;
         }
         //取消订阅
         selfChannelP->basicCancel(message);
      }
      //发布消息
      void onBasicPublish(const muduo::net::TcpConnectionPtr &conn, const BasicPublishRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("发布消息失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("发布消息失败，因为信道不存在");
            return ;
         }
         //发布消息
         selfChannelP->basicPublish(message);
      }
      //确认消息
      void onBasicAckRequest(const muduo::net::TcpConnectionPtr &conn, const BasicAckRequestPtr &message, muduo::Timestamp)
      {
         //1.获取连接
         auto selfConnP=_connectionManager->getConnection(conn);
         if(selfConnP.get()==nullptr)
         {
            ELOG("确认消息失败，因为连接不存在");
            return ;
         }
         //2.获取信道
         auto selfChannelP=selfConnP->getChannel(message->cid());
         if(selfChannelP.get()==nullptr)
         {
            ELOG("确认消息失败，因为信道不存在");
            return ;
         }
         //确认消息
         selfChannelP->basicAck(message);
      }
      void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,const std::shared_ptr<google::protobuf::Message>& message,muduo::Timestamp)
      {
         //收到未知消息
         LOG_INFO<<message->GetTypeName();
         conn->shutdown();
      }
      void onConnection(const muduo::net::TcpConnectionPtr &conn)
      {
         if (conn->connected())
         {
            LOG_INFO << "连接成功";
            _connectionManager->newConnection(conn,_codec,_consumerManager,_virtualHost,_threadPool);
         }
         else
         {
            LOG_INFO << "断开连接";
            _connectionManager->deleteConnection(conn);
         }
      }
private:
      muduo::net::EventLoop _loop;
      muduo::net::TcpServer _server;  // 服务器
      ProtobufDispatcher _dispatcher; // 请求分发器对象--要向其中注册请求处理函数
      ProtobufCodecPtr _codec;          // protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
      QueueConsumerManager::ptr _consumerManager;    // 消费者管理句柄
      VirtualHost::ptr _virtualHost;    // 虚拟机管理句柄
      ThreadPool::ptr _threadPool;       // 线程池管理句柄
      ConnectionManager::ptr _connectionManager;    // 连接管理句柄
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

127.

128.

129.

130.

131.

132.

133.

134.

135.

136.

137.

138.

139.

140.

141.

142.

143.

144.

145.

146.

147.

148.

149.

150.

151.

152.

153.

154.

155.

156.

157.

158.

159.

160.

161.

162.

163.

164.

165.

166.

167.

168.

169.

170.

171.

172.

173.

174.

175.

176.

177.

178.

179.

180.

181.

182.

183.

184.

185.

186.

187.

188.

189.

190.

191.

192.

193.

194.

195.

196.

197.

198.

199.

200.

201.

202.

203.

204.

205.

206.

207.

208.

209.

210.

211.

212.

213.

214.

215.

216.

217.

218.

219.

220.

221.

222.

223.

224.

225.

226.

227.

228.

229.

230.

231.

232.

233.

234.

235.

236.

237.

238.

239.

240.

241.

242.

243.

244.

245.

246.

247.

248.

249.

250.

251.

252.

253.

254.

255.

256.

257.

258.

259.

260.

261.

262.

263.

264.

265.

266.

267.

268.

269.

270.

271.

272.

273.

274.

275.

276.

277.

278.

279.

280.

281.

282.

283.

284.

285.

286.

287.

288.

289.

290.

291.

292.

293.

294.

295.

296.

297.

298.

299.

300.

301.

302.

303.

304.

305.

306.

307.

308.

309.

310.

311.

312.

313.

314.

315.

316.

317.

318.

319.

320.

321.

322.

323.

324.

325.

326.

327.

328.

329.

                     客户端实现

   1、订阅者模块

   ⼀个并不直接对⽤⼾展⽰的模块，其在客⼾端体现的作⽤就是对于⻆⾊的描述，表⽰这是⼀个消费者。
                                 登录后复制
#pragma once
#include "../mqcomm/helper.hpp"
#include "../mqcomm/logger.hpp"
#include "../mqcomm/message.pb.h"
namespace lsxmq
{
using ConsumerCallBack=std::function<void(const std::string&,BasicProperties*,const std::string&)>;
                                                   //消费者标识    //基本属性          //消息内容
struct Consumer
{
      using ptr=std::shared_ptr<Consumer>;
      Consumer()
      {
         DLOG("new Consumer: %p",this);
      }
      ~Consumer()
      {
         DLOG("delete Consumer: %p",this);
      }
      Consumer(const std::string& ctag,const std::string& cqname,bool cautoAck,const ConsumerCallBack& ccallback)
         :tag(ctag),qname(cqname),autoAck(cautoAck),callback(ccallback)
      {}
      std::string tag;//消费者标识
      std::string qname;//队列名称
      bool autoAck;//自动确认标志
      ConsumerCallBack callback;//消费回调函数
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

                     2、信道管理模块

   信道信息：


信道ID

信道关联的网络通讯连接对象

protobuf协议处理对象

信道关联的消费者

请结构<哀求id,相应>，方便查找)

互斥锁&条件变量(大部门哀求是壅闭操作，但muduo库的通讯是异步的，因此在收到应答后，通过判定是否是等候的指定相应来进行同步)：发送消息时，还没有声明交换机成功怎么办？以是要同步

   信道操作：


打开信道

关闭信道

声明交换机

删除交换机

声明队列

删除队列

绑定

解绑

订阅

取消订阅

发布消息

确认消息

                                 登录后复制
#pragma once
#include "../mqcomm/logger.hpp"
#include "../mqcomm/threadpool.hpp"
#include "../mqcomm/proto.pb.h"
#include "../mqcomm/message.pb.h"
#include "consumer.hpp"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
namespace lsxmq
{
using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;
using BasicCommonResponsePtr=std::shared_ptr<BasicCommonResponse>;
using BasicConsumeResponsePtr=std::shared_ptr<basicConsumeResponse>;
class Channel
{
public:
      using ptr=std::shared_ptr<Channel>;
      Channel(const muduo::net::TcpConnectionPtr& conn
            ,const ProtobufCodecPtr& codec)
            :_cid(UUIDHelper::uuid()),_conn(conn),_codec(codec)
      {}
      ~Channel()
      {
         basicCancel();//析构时取消订阅
      }
      std::string cid()
      {
         return _cid;
      }
      //打开信道
      bool openChannel()
      {
         //构造请求
         OpenChannelRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("打开信道失败");
         }
         return rsp->ok();
      }
      //关闭信道
      bool closeChannel()
      {
         //构造请求
         CloseChannelRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("关闭信道失败");
         }
         return rsp->ok();
      }
      //声明交换机
      bool declareExchange(const std::string& exchangeName
                        ,ExchangeType exchangeType
                        ,bool durable
                        ,bool autoDelete
                        ,google::protobuf::Map<std::string, std::string>& args)
      {
         //构造请求
         DeclareExchangeRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_exchange_name(exchangeName);
         req.set_exchange_type(exchangeType);
         req.set_durable(durable);
         req.set_auto_delete(autoDelete);
         req.mutable_args()->swap(args);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("声明交换机失败");
         }
         return rsp->ok();
      }
      //删除交换机
      bool deleteExchange(const std::string& exchangeName)
      {
         //构造请求
         DeleteExchangeRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_exchange_name(exchangeName);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("删除交换机失败");
         }
         return rsp->ok();
      }
      //声明消息队列
      bool declareMessageQueue(const std::string& queueName
                              ,bool exclusive
                              ,bool durable
                              ,bool autoDelete
                              ,google::protobuf::Map<std::string, std::string>& args)
      {
         //构造请求
         DeclareMessageQueueRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_queue_name(queueName);
         req.set_exclusive(exclusive);
         req.set_durable(durable);
         req.set_auto_delete(autoDelete);
         req.mutable_args()->swap(args);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("声明消息队列失败");
         }
         return rsp->ok();
      }
      //删除消息队列
      bool deleteMessageQueue(const std::string& queueName)
      {
         //构造请求
         DeleteMessageQueueRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_queue_name(queueName);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("删除消息队列失败");
         }
         return rsp->ok();
      }
      //绑定
      bool bind(const std::string& exchangeName
                  ,const std::string& queueName
                  ,const std::string& bindingKey)
      {
         //构造请求
         BindRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_exchange_name(exchangeName);
         req.set_queue_name(queueName);
         req.set_binding_key(bindingKey);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("绑定失败");
         }
         return rsp->ok();
      }
      //解绑
      bool unbind(const std::string& exchangeName
                  ,const std::string& queueName)
      {
         //构造请求
         BindRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_exchange_name(exchangeName);
         req.set_queue_name(queueName);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("解绑失败");
         }
         return rsp->ok();
      }
      //订阅
      bool basicConsume(const std::string& consumerTag
                     ,const std::string& queueName
                     ,bool autoAck
                     ,const ConsumerCallBack& cb)
      {
         //构造请求
         if(_consumer.get()!=nullptr)
         {
            DLOG("已订阅，不可重复订阅")
            return false;
         }
         BasicConsumeRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_consumer_tag(consumerTag);
         req.set_queue_name(queueName);
         req.set_auto_ack(autoAck);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("订阅失败");
            return false;
         }
         //订阅成功，要构造对应的消费者
         _consumer=std::make_shared<Consumer>(consumerTag,queueName,autoAck,cb);
         return rsp->ok();
      }
      //取消订阅
      bool basicCancel()
      {
         if(_consumer.get()==nullptr)
         {
            return true;
         }
         //构造请求
         BasicCancelRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_consumer_tag(_consumer->tag);
         req.set_queue_name(_consumer->qname);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("取消订阅失败");
            return false;
         }
         //取消订阅时清理消费者对象
         _consumer.reset();
         return rsp->ok();
      }
      //发布消息
      bool basicPublish(const std::string& exchangeName
                        ,const std::string& body
                        ,BasicProperties* bp)
      {
         //构造请求
         BasicPublishRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_exchange_name(exchangeName);
         req.set_body(body);
         if(bp!=nullptr)
         {
            req.mutable_properties()->set_id(bp->id());
            req.mutable_properties()->set_deliverymode(bp->deliverymode());
            req.mutable_properties()->set_routing_key(bp->routing_key());
         }
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("发布消息失败");
         }
         return rsp->ok();
      }
      //确认消息
      bool basicAck(const std::string& mid)
      {
         if(_consumer.get()==nullptr)
         {
            ELOG("消息确认时，没有找到消费者")
            return false;
         }
         //构造请求
         BasicAckRequest req;
         req.set_rid(UUIDHelper::uuid());
         req.set_cid(_cid);
         req.set_queue_name(_consumer->qname);
         req.set_mid(mid);
         //发送请求并等待
         _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
         auto rsp=waitBasicResponse(req.rid());
         if(rsp->ok()==false)
         {
            ELOG("确认消息失败");
         }
         return rsp->ok();
      }
public:
      //收到基础响应时，向hash map中添加响应
      void putBasicResponse(const BasicCommonResponsePtr& rsp)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         _basicRsp.insert(std::make_pair(rsp->rid(),rsp));
         _cond.notify_all();//唤醒所有等待线程，只有满足条件的才能竞争锁
      }
      //收到消息推送后，需要信道找到对应的消费者，通过回调函数进行处理
      void consume(const BasicConsumeResponsePtr& rsp)
      {
         if(_consumer.get()==nullptr)
         {
            ELOG("消费者不存在");
            return;
         }
         if(rsp->consumer_tag()!=_consumer->tag)
         {
            ELOG("收到错误的响应，消费者标志匹配失败");
            return;
         }
         //执行回调函数
         _consumer->callback(rsp->consumer_tag(),rsp->mutable_properties(),rsp->body());
      }
private:
      //等待响应
      BasicCommonResponsePtr waitBasicResponse(const std::string& rid)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         _cond.wait(lock,[&rid,this](){
            return _basicRsp.find(rid)!=_basicRsp.end();//等到收到了对应的响应才退出
         });
         //拿到响应，然后再删除
         BasicCommonResponsePtr rsp=_basicRsp[rid];
         _basicRsp.erase(rid);
         return rsp;
      }
private:
      std::string _cid;  //信道id
      muduo::net::TcpConnectionPtr _conn;  //信道关联的网络通信连接
      ProtobufCodecPtr _codec;  //protobuf协议处理器
      Consumer::ptr _consumer;  //信道对应的消费者,当确定了信道角色后构造(订阅操作后)
      std::mutex _mutex;
      std::condition_variable _cond;
      std::unordered_map<std::string,BasicCommonResponsePtr> _basicRsp;  //<请求id,响应>
};
class ChannelManager
{
public:
      using ptr=std::shared_ptr<ChannelManager>;
      ChannelManager()
      {
         DLOG("new ChannelManager: %p",this);
      }
      ~ChannelManager()
      {
         DLOG("delete ChannelManager: %p",this);
      }
      //打开信道
      Channel::ptr createChannel(const muduo::net::TcpConnectionPtr& conn
         ,const ProtobufCodecPtr& codec)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //队列是否已存在？不需要担心，因为创建信道会自动生成id
         //构造信道对象
         Channel::ptr chp=std::make_shared<Channel>(conn,codec);
         //添加管理
         _channels[chp->cid()]=chp;
         return chp;
      }
      void deleteChannel(const std::string& cid)//关闭信道
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //队列是否不存在
         auto it=_channels.find(cid);
         if(it==_channels.end())
         {
            return;
         }
         //删除信道
         _channels.erase(cid);
      }
      Channel::ptr getChannel(const std::string& cid)
      {
         std::unique_lock<std::mutex> lock(_mutex);
         //队列是否不存在
         auto it=_channels.find(cid);
         if(it==_channels.end())
         {
            return Channel::ptr();
         }
         return it->second;
      }
private:
      std::mutex _mutex;
      std::unordered_map<std::string,Channel::ptr> _channels;
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

127.

128.

129.

130.

131.

132.

133.

134.

135.

136.

137.

138.

139.

140.

141.

142.

143.

144.

145.

146.

147.

148.

149.

150.

151.

152.

153.

154.

155.

156.

157.

158.

159.

160.

161.

162.

163.

164.

165.

166.

167.

168.

169.

170.

171.

172.

173.

174.

175.

176.

177.

178.

179.

180.

181.

182.

183.

184.

185.

186.

187.

188.

189.

190.

191.

192.

193.

194.

195.

196.

197.

198.

199.

200.

201.

202.

203.

204.

205.

206.

207.

208.

209.

210.

211.

212.

213.

214.

215.

216.

217.

218.

219.

220.

221.

222.

223.

224.

225.

226.

227.

228.

229.

230.

231.

232.

233.

234.

235.

236.

237.

238.

239.

240.

241.

242.

243.

244.

245.

246.

247.

248.

249.

250.

251.

252.

253.

254.

255.

256.

257.

258.

259.

260.

261.

262.

263.

264.

265.

266.

267.

268.

269.

270.

271.

272.

273.

274.

275.

276.

277.

278.

279.

280.

281.

282.

283.

284.

285.

286.

287.

288.

289.

290.

291.

292.

293.

294.

295.

296.

297.

298.

299.

300.

301.

302.

303.

304.

305.

306.

307.

308.

309.

310.

311.

312.

313.

314.

315.

316.

317.

318.

319.

320.

321.

322.

323.

324.

325.

326.

327.

328.

329.

330.

331.

332.

333.

334.

335.

336.

337.

338.

339.

340.

341.

342.

343.

344.

345.

346.

347.

348.

349.

350.

351.

352.

353.

354.

355.

356.

357.

358.

359.

360.

361.

362.

363.

364.

365.

366.

367.

368.

369.

370.

371.

372.

373.

374.

375.

376.

377.

378.

379.

380.

381.

382.

383.

384.

385.

386.

387.

388.

389.

                     3、异步工作线程池模块

   客⼾端这边存在两个异步⼯作线程，


⼀个是muduo库中客⼾端连接的异步循环线程EventLoopThread，

⼀个是当收到消息后进⾏异步处理的⼯作线程池。

   这两项都不是以连接为单元进⾏创建的，⽽是创建后，可以⽤以多个连接中，因此单独进⾏封装。
                                 登录后复制
#pragma once
#include "muduo/net/EventLoopThreadPool.h"
#include "../mqcomm/logger.hpp"
#include "../mqcomm/threadpool.hpp"
namespace lsxmq
{
class AsyncWorker
{
public:
      using ptr=std::shared_ptr<AsyncWorker>;
      muduo::net::EventLoopThreadPool _loopthread;
      ThreadPool _pool;
};
};
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

                     4、连接管理模块

   在客⼾端这边，RabbitMQ弱化了客⼾端的概念，因为⽤⼾所需的服务都是通过信道来提供的，因此操作头脑转换为先创建连接，通过连接创建信道，通过信道提供服务这⼀流程。
   这个模块同样是针对muduo库客⼾端连接的⼆次封装，向⽤⼾提供创建channel信道的接⼝，创建信道后，可以通过信道来获取指定服务。
                                 登录后复制
#pragma once
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/base/CountDownLatch.h"
#include "muduo/net/EventLoopThread.h"
#include "worker.hpp"
#include "channel.hpp"
#include <string>
namespace lsxmq
{
typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
class Connection
{
public:
using ptr=std::shared_ptr<Connection>;
Connection(const std::string &serverIp, uint16_t serverPort,const AsyncWorker::ptr& worker)
         : _latch(1)
         ,_worker(worker)
         , _client(_worker->_loopthread.startLoop()
            , muduo::net::InetAddress(serverIp, serverPort), "ProtobufClient")
         , _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1
            , std::placeholders::_2, std::placeholders::_3))
         , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher
            , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
         ,_channels(std::make_shared<ChannelManager>())
      {
         // 注册处理函数
         _dispatcher.registerMessageCallback<BasicCommonResponse>(std::bind(&Connection::onBasicCommonResponse, this
            , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::onBasicConsumeResponse, this
            , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
         // 客户端注册回调函数
         _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
         _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get()
            , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

         //连接服务器
         _client.connect();
         _latch.wait(); // 等待连接服务器成功
      }
      //打开信道
      Channel::ptr openChannel()
      {
         //构造一个信道
         auto channel=_channels->createChannel(_conn,_codec);
         //打开信道
         auto ret=channel->openChannel();
         if(ret==false)
         {
            ELOG("打开信道失败");
            _channels->deleteChannel(channel->cid());
            return Channel::ptr();
         }
         return channel;
      }
      //关闭信道
      bool closeChannel(const Channel::ptr& channel)
      {
         auto ret=channel->closeChannel();
         if(ret==false)
         {
            ELOG("关闭信道失败");
            return false;
         }
         _channels->deleteChannel(channel->cid());
         return true;
      }
private:
      //收到普通响应
      void onBasicCommonResponse(const muduo::net::TcpConnectionPtr &conn, const BasicCommonResponsePtr &message, muduo::Timestamp)
      {
         //1.获取信道
         auto channel=_channels->getChannel(message->cid());
         if(channel.get()==nullptr)
         {
            ELOG("处理普通响应失败，因为获取信道失败");
            return;
         }
         //2.将响应加入hash map
         channel->putBasicResponse(message);
      }
      //收到推送消息
      void onBasicConsumeResponse(const muduo::net::TcpConnectionPtr &conn,const BasicConsumeResponsePtr& message, muduo::Timestamp)
      {
         //1.获取信道
         auto channel=_channels->getChannel(message->cid());
         if(channel.get()==nullptr)
         {
            ELOG("处理普通响应失败，因为获取信道失败");
            return;
         }
         //2.对收到的消息进行处理，由工作线程池完成
         auto task=std::bind(&Channel::consume,channel.get(),message);
         _worker->_pool.push(task);
      }
      void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,const MessagePtr& message,muduo::Timestamp)
      {
         LOG_INFO<<message->GetTypeName();
         //收到了未定义的消息，断开连接
         conn->shutdown();
      }
      void onConnection(const muduo::net::TcpConnectionPtr &conn)
      {
         if (conn->connected())
         {
            LOG_INFO << "连接成功";
            _conn = conn;
            // 解除阻塞
            _latch.countDown();
         }
         else
         {
            LOG_INFO << "断开连接";
         }
      }
private:
      muduo::CountDownLatch _latch;          // 用于同步等待连接服务器成功
      AsyncWorker::ptr _worker;  //异步工作线程
      muduo::net::TcpClient _client;          // 客户端
      ProtobufDispatcher _dispatcher;       // 派发器
      ProtobufCodecPtr _codec;                   // protobuf协议处理器
      muduo::net::TcpConnectionPtr _conn;
      ChannelManager::ptr _channels;  //信道管理句柄
};
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

54.

55.

56.

57.

58.

59.

60.

61.

62.

63.

64.

65.

66.

67.

68.

69.

70.

71.

72.

73.

74.

75.

76.

77.

78.

79.

80.

81.

82.

83.

84.

85.

86.

87.

88.

89.

90.

91.

92.

93.

94.

95.

96.

97.

98.

99.

100.

101.

102.

103.

104.

105.

106.

107.

108.

109.

110.

111.

112.

113.

114.

115.

116.

117.

118.

119.

120.

121.

122.

123.

124.

125.

126.

127.

128.

129.

130.

131.

132.

133.

134.

                     综合测试

   广播交换模式下：
                                 登录后复制
//订阅者
#include "connection.hpp"
void callBack(const lsxmq::Channel::ptr& channel,const std::string& tag,lsxmq::BasicProperties* bp,const std::string& body)
{
std::cout<<"消费者"<<tag<<"收到一条消息："<<body<<std::endl;
//发送响应
channel->basicAck(bp->id());
}
int main(int argc,char* argv[])
{
if(argc!=2)
{
      std::cout<<"格式错误"<<std::endl;
      return 0;
}
std::string qname=argv[1];
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::FANOUT,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","queue1");
channel->bind("exchange1","queue2","news.music.#");
      //4.5订阅一个队列
auto cb=std::bind(callBack,channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
channel->basicConsume("consumer1",qname,false,cb);
//5.等待
while(1)
{
      std::this_thread::sleep_for(std::chrono::seconds(3));
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

                                                   登录后复制
//发布者
#include "connection.hpp"
int main()
{
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::FANOUT,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","queue1");
channel->bind("exchange1","queue2","news.music.#");
      //4.5循环向交换机发布消息
for(int i=11;i<=20;++i)
{
      //交换机类型是广播交换，所以不需要BasicProperties
      channel->basicPublish("exchange1","hello-world-"+std::to_string(i),nullptr);
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

                     直接交换模式：
                                 登录后复制
//订阅者
#include "connection.hpp"
void callBack(const lsxmq::Channel::ptr& channel,const std::string& tag,lsxmq::BasicProperties* bp,const std::string& body)
{
std::cout<<"消费者"<<tag<<"收到一条消息："<<body<<std::endl;
//发送响应
channel->basicAck(bp->id());
}
int main(int argc,char* argv[])
{
if(argc!=2)
{
      std::cout<<"格式错误"<<std::endl;
      return 0;
}
std::string qname=argv[1];
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::DIRECT,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","queue1");
channel->bind("exchange1","queue2","queue2");
      //4.5订阅一个队列
auto cb=std::bind(callBack,channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
channel->basicConsume("consumer1",qname,false,cb);
//5.等待
while(1)
{
      std::this_thread::sleep_for(std::chrono::seconds(3));
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

                                                   登录后复制
//发布者
#include "connection.hpp"
int main()
{
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::DIRECT,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","queue1");
channel->bind("exchange1","queue2","queue2");
      //4.5循环向交换机发布消息
for(int i=1;i<=10;++i)
{
      lsxmq::BasicProperties bp;
      bp.set_id(lsxmq::UUIDHelper::uuid());
      bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
      bp.set_routing_key("queue1");
      channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
}
for(int i=11;i<=15;++i)
{
      lsxmq::BasicProperties bp;
      bp.set_id(lsxmq::UUIDHelper::uuid());
      bp.set_deliverymode(lsxmq::DeliveryMode::UNDURABLE);
      bp.set_routing_key("queue2");
      channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

                     主题交换模式：
                                 登录后复制
//订阅者
#include "connection.hpp"
void callBack(const lsxmq::Channel::ptr& channel,const std::string& tag,lsxmq::BasicProperties* bp,const std::string& body)
{
std::cout<<"消费者"<<tag<<"收到一条消息："<<body<<std::endl;
//发送响应
channel->basicAck(bp->id());
}
int main(int argc,char* argv[])
{
if(argc!=2)
{
      std::cout<<"格式错误"<<std::endl;
      return 0;
}
std::string qname=argv[1];
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::TOPIC,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","sport.#");
channel->bind("exchange1","queue2","music.#");
      //4.5订阅一个队列
auto cb=std::bind(callBack,channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
channel->basicConsume("consumer1",qname,false,cb);
//5.等待
while(1)
{
      std::this_thread::sleep_for(std::chrono::seconds(3));
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

                                                   登录后复制
//发布者
#include "connection.hpp"
int main()
{
//1.构造异步工作者
lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
//2.构造连接
lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
//3.打开信道
lsxmq::Channel::ptr channel=conn->openChannel();
//4.通过信道进行操作
      //4.1.声明一台交换机
google::protobuf::Map<std::string, std::string> args;
channel->declareExchange("exchange1",lsxmq::ExchangeType::TOPIC,true,false,args);
      //4.2.声明队列queue1
channel->declareMessageQueue("queue1",false,true,false,args);
      //4.3.声明队列queue2
channel->declareMessageQueue("queue2",false,true,false,args);
      //4.4.将队列与交换机绑定
channel->bind("exchange1","queue1","sport.#");
channel->bind("exchange1","queue2","music.#");
      //4.5循环向交换机发布消息
for(int i=1;i<=10;++i)
{
      lsxmq::BasicProperties bp;
      bp.set_id(lsxmq::UUIDHelper::uuid());
      bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
      bp.set_routing_key("music.pop");
      channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
}
for(int i=1;i<=5;++i)
{
      lsxmq::BasicProperties bp;
      bp.set_id(lsxmq::UUIDHelper::uuid());
      bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
      bp.set_routing_key("music.pop.ajc");
      channel->basicPublish("exchange1","good-morning-"+std::to_string(i),&bp);
}
for(int i=1;i<=5;++i)
{
      lsxmq::BasicProperties bp;
      bp.set_id(lsxmq::UUIDHelper::uuid());
      bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
      bp.set_routing_key("news.gassip");
      channel->basicPublish("exchange1","have-nice-day-"+std::to_string(i),&bp);
}
//5.关闭信道
      conn->closeChannel(channel);
return 0;
}
复制代码


1.

2.

3.

4.

5.

6.

7.

8.

9.

10.

11.

12.

13.

14.

15.

16.

17.

18.

19.

20.

21.

22.

23.

24.

25.

26.

27.

28.

29.

30.

31.

32.

33.

34.

35.

36.

37.

38.

39.

40.

41.

42.

43.

44.

45.

46.

47.

48.

49.

50.

51.

52.

53.

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

		自动登录	找回密码
密码			立即注册

模拟RabbitMQ实现消息队列

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块