ToB企服应用市场:ToB评测及商务社交产业平台

标题: 模拟RabbitMQ实现消息队列 [打印本页]
作者: 老婆出轨    时间: 7 天前
标题: 模拟RabbitMQ实现消息队列
代码地址： 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

     是序列化反序列化的框架
     特点：语言无关、平台无关；高效；扩展性、兼容性好。
     使用方法：
     
     语法规范：
     
     语法：
     指定proto3语法，syntax = "proto3"。
     声明命名空间，package 声明符，示.proto文件的命名空间。
     界说结构对象描述message 消息体名称{}，名称用驼峰
     界说消息字段 字段范例 字段名 = 字段唯一编号
     

     

     编译：protoc --cpp_out=路径 proto文件名称
     protoc --cpp_out=. contacts.proto
                                   登录后复制                        
  1. syntax= "proto3";//声明语法版本
  3. package contacts;//声明代码的命名空间
  5. message contact{
  6.     //各个字段  字段类型 字段名 = 字段唯一编号
  7.     uint64 sn = 1;
  8.     string name = 2;
  9.     float score = 3;
  10. };
复制代码
      
                                                     登录后复制                        
  1. main:main.cpp contacts.pb.cc
  2.         g++ -o $@ $^ -std=c++11 -lprotobuf
复制代码
      
                                                     登录后复制                        
  1. #include "contacts.pb.h"
  2. #include <iostream>
  3. int main()
  4. {
  5.     //创建通讯录对象
  6.     contacts::contact con;
  7.     //添加字段
  8.     con.set_name("Nesta");
  9.     con.set_sn(13);
  10.     con.set_score(99);
  11.     //序列化
  12.     std::string str=con.SerializeAsString();
  13.     //反序列化
  14.     contacts::contact conR;
  15.     conR.ParseFromString(str);
  16.     //打印信息
  17.     std::cout<<"name="<<conR.name()<<std::endl;
  18.     std::cout<<"sn="<<conR.sn()<<std::endl;
  19.     std::cout<<"socre="<<conR.score()<<std::endl;
  20.     return 0;
  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：
                                   登录后复制                        
  1. #include "./include/muduo/net/TcpServer.h"
  2. #include "./include/muduo/net/EventLoop.h"
  3. #include "./include/muduo/net/TcpConnection.h"
  4. #include <iostream>
  5. #include <string>
  6. #include <functional>
  7. #include <unordered_map>
  8. class TranslateServer
  9. {
  10. public:
  11.     TranslateServer(uint16_t port)
  12.         :_server(&_loop
  13.         ,muduo::net::InetAddress("0.0.0.0",8888)
  14.         ,"translateServer"
  15.         ,muduo::net::TcpServer::kNoReusePort)
  16.         {
  17.             //注册回调函数
  18.             auto ccb=std::bind(&TranslateServer::OnConnection,this,std::placeholders::_1);
  19.             auto mcb=std::bind(&TranslateServer::OnMessage,this,std::placeholders::_1
  20.             ,std::placeholders::_2,std::placeholders::_3);
  21.             _server.setConnectionCallback(ccb);
  22.             _server.setMessageCallback(mcb);
  24.             _server.setConnectionCallback(ccb);
  25.             _server.setMessageCallback(mcb);
  26.         }
  27.     void Start()//启动服务器
  28.     {
  29.         
  30.         _server.start();
  31.         _loop.loop();//这是一个死循环
  32.     }
  33. private:
  34.     void OnConnection(const muduo::net::TcpConnectionPtr& conn)
  35.     {
  36.         //对连接的回调
  37.         //建立连接和断开连接时都会触发
  38.         if(conn->connected())//确定是否连接成功
  39.         {
  40.             std::cout<<"连接成功"<<std::endl;
  41.         }
  42.         else
  43.         {
  44.             std::cout<<"连接失败"<<std::endl;
  45.         }
  46.     }
  47.     std::string Translate(std::string key)
  48.     {
  49.         std::unordered_map<std::string,std::string> dict={
  50.             {"你好","hello"}
  51.             ,{"hello","你好"}
  52.             ,{"world","世界"}
  53.             ,{"世界","world"}
  54.         };//字典
  55.         auto it=dict.find(key);
  56.         if(it!=dict.end())//找到了
  57.         {
  58.             return it->second;
  59.         }
  60.         else//没找到
  61.         {
  62.             return "不认识这个词语";
  63.         }
  64.     }
  65.     void OnMessage(const muduo::net::TcpConnectionPtr& ptr,muduo::net::Buffer* buffer,muduo::Timestamp)
  66.     {
  67.         //对消息的回调
  69.         //1.从缓冲区里拿到数据
  70.         std::string key=buffer->retrieveAllAsString();
  71.         //2.翻译
  72.         std::string val=Translate(key);
  73.         //3.将翻译结果发回客户端
  74.         muduo::net::Buffer message;
  75.         message.append(val);
  76.         ptr->send(&message);
  77.     }
  78.     //loop是对epool的事件监听，触发事件后进行IO操作
  79.     muduo::net::EventLoop _loop;
  80.     //_server主要用于设置回调函数，告诉服务器收到什么消息进行怎样的处理
  81.     muduo::net::TcpServer _server;
  82. };
  83. int main()
  84. {
  85.     TranslateServer svr(8888);
  86.     svr.Start();
  87.     return 0;
  88. }
复制代码
      
                       Client
                                   登录后复制                        
  1. #include "include/muduo/net/TcpClient.h"
  2. #include "include/muduo/net/Buffer.h"
  3. #include "include/muduo/net/EventLoopThread.h"
  4. #include "include/muduo/net/InetAddress.h"
  5. #include "include/muduo/base/CountDownLatch.h"
  6. #include <string>
  7. #include <iostream>
  8. #include <functional>
  9. class TranslateClient
  10. {
  11. public:
  12.     TranslateClient(std::string serverIp,uint16_t serverPort)
  13.         :_client(_loopThread.startLoop()
  14.         ,muduo::net::InetAddress(serverIp,serverPort)
  15.         ,"translateClient")
  16.         ,_latch(1)//一定要设为1
  17.     {
  18.         //注册回调函数
  19.         auto ccb=std::bind(&TranslateClient::OnConnection,this,std::placeholders::_1);
  20.         auto mcb=std::bind(&TranslateClient::OnMessage,this
  21.         ,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
  22.         
  23.         _client.setConnectionCallback(ccb);
  24.         _client.setMessageCallback(mcb);
  25.         std::cout<<"初始化完成"<<std::endl;
  26.     }
  27.     void Connect()//连接服务器
  28.     {
  29.         _client.connect();//进行连接
  30.         //之所以要阻塞就是为了方式没有连接成功就进行了其他操作，导致错误
  31.         std::cout<<"开始连接"<<std::endl;
  32.         _latch.wait();//阻塞等待连接成功
  33.     }
  34.     void Send(std::string& message)//发送消息
  35.     {
  36.         if(_conn->connected())//确认连接成功再进行IO
  37.         {
  38.             _conn->send(message);
  39.         }
  40.         
  41.     }
  42. private:
  43.     void OnConnection(const muduo::net::TcpConnectionPtr& conn)
  44.     {
  45.         if(conn->connected())//连接上了
  46.         {
  47.             _latch.countDown();//解除阻塞
  48.             _conn=conn;//保存连接
  49.         }
  50.         else
  51.         {
  52.             //......
  53.         }
  54.     }
  55.     void OnMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer,muduo::Timestamp ts)
  56.     {
  57.         //来消息了就打印出来
  58.         std::string message=buffer->retrieveAllAsString();
  59.         std::cout<<"Server Say> "<<message<<std::endl;
  60.     }
  61. private:
  62.     muduo::CountDownLatch _latch;//用于同步
  63.     muduo::net::EventLoopThread _loopThread;//创建线程进行IO
  64.     muduo::net::TcpClient _client;
  65.     muduo::net::TcpConnectionPtr _conn;//客户端与服务器的连接
  66. };
  67. int main()
  68. {
  69.     TranslateClient cli("127.0.0.1",8888);
  70.     cli.Connect();//建立连接
  71.     while(1)
  72.     {
  73.         std::string message;
  74.         std::cin>>message;
  75.         cli.Send(message);
  76.     }
  77.     return 0;
  78. }
复制代码
      
                       上面的服务器与客户端没有协议，直接通讯。
     下面是基于protobuf来进行通讯。
     

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

     SQLite3

     是什么？进程内的轻量级数据库，是本地的，不必要服务器，它给我们提供了一系列接口。
     官方文档： List Of SQLite Functions
                                   登录后复制                        
  1. sqlite3操作流程：0. 查看当前数据库在编译阶段是否启动了线程安全
  2.                 int sqlite3_threadsafe()； 0-未启⽤； 1-启⽤
  3.                 需要注意的是sqlite3是有三种安全等级的：
  4.                 1. ⾮线程安全模式--不使用任何锁，效率最高
  5.     2. 线程安全模式(不同的连接在不同的线程/进程间是安全的，即⼀个句柄不能⽤于多线程
  6.                 间)
  7.                 3. 串⾏化模式(可以在不同的线程/进程间使⽤同⼀个句柄)
  8. 1. 创建/打开数据库⽂件，并返回操作句柄
  9.                 int sqlite3_open(const char *filename, sqlite3 **ppDb) 成功返回SQLITE_OK
  10.                 //若在编译阶段启动了线程安全，则在程序运⾏阶段可以通过参数选择线程安全等级
  11.                 int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const
  12.                 char *zVfs );
  13.                 flag:
  14.                 SQLITE_OPEN_READWRITE -- 以可读可写⽅式打开数据库⽂件
  15.                 SQLITE_OPEN_CREATE -- 不存在数据库⽂件则创建
  16.                 SQLITE_OPEN_NOMUTEX--多线程模式，只要不同的线程使⽤不同的连接即可保证线程
  17.                 安全
  18.                 SQLITE_OPEN_FULLMUTEX--串⾏化模式
  19.                 返回：SQLITE_OK表⽰成功，否则有问题
  20. 2. 执⾏语句
  21.                 int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),
  22.                 void* arg, char **err)
  23.         err是错误信息
  24.         
  25.                 int (*callback)(void*,int,char**,char**)
  26.                 void* : 是设置的在回调时传⼊的arg参数
  27.                 int：⼀⾏中数据的列数
  28.                 char**：存储⼀⾏数据的字符指针数组
  29.                 char**：每⼀列的字段名称
  30.                 这个回调函数有个int返回值，成功处理的情况下必须返回0，返回⾮0会触发ABORT退出程序
  31.                 返回：SQLITE_OK表⽰成功
  32. 3. 销毁句柄
  33.                 int sqlite3_close(sqlite3* db); 成功返回SQLITE_OK
  34.                 int sqlite3_close_v2(sqlite3*); 推荐使⽤--⽆论如何都会返回SQLITE_OK
  35. 获取错误信息
  36.                 const char *sqlite3_errmsg(sqlite3* db);
复制代码
      
                       sqlite3 test.db//打开数据库文件
     .table//查看表
                                   登录后复制                        
  1. //封装实现一个SqliteHelper类，提供简单的sqlite数据库操作接口，完成对数据增删改查操作
  2. //1.创建/打开数据库文件
  3. //2.针对打开的数据库进行操作
  4.     //2.1表的操作
  5.     //2.2数据的操作
  6. //3.关闭数据库
  8. #include <sqlite3.h>
  9. #include <string>
  10. #include <iostream>
  12. class SqliteHelper
  13. {
  14. public:
  15.     typedef int(*SqliteCallback)(void*,int,char**,char**);
  16.     SqliteHelper(std::string dbfile)
  17.         :_dbfile(dbfile)
  18.     {}
  19.     bool open(int safeLevel=SQLITE_OPEN_FULLMUTEX)//打开数据库文件
  20.     {
  21.         //默认安全级别为串行化模式
  22.         //int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs )
  23.         int ret=sqlite3_open_v2(_dbfile.c_str(),&_handler,SQLITE_OPEN_CREATE|SQLITE_OPEN_READWRITE|safeLevel,nullptr);
  24.         if(ret!=SQLITE_OK)
  25.         {
  26.             std::cout<<"创建/打开数据库文件失败："<<sqlite3_errmsg(_handler)<<std::endl;
  27.             return false;
  28.         }
  29.         return true;
  30.     }
  31.     bool exec(const std::string& sql,SqliteCallback cb,void* arg)//进行操作
  32.     {
  33.         //int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),void* arg, char **err)
  34.         int ret=sqlite3_exec(_handler,sql.c_str(),cb,arg,nullptr);
  35.         if(ret!=SQLITE_OK)
  36.         {
  37.             std::cout<<sql<<"执行失败："<<sqlite3_errmsg(_handler)<<std::endl;
  38.             return false;
  39.         }
  40.         return true;
  41.     }
  42.     void close()//关闭数据库文件
  43.     {
  44.         //int sqlite3_close_v2(sqlite3*)
  45.         sqlite3_close_v2(_handler);
  46.     }
  47. private:
  48.     sqlite3* _handler;//句柄
  49.     std::string _dbfile;//数据库文件名
  50. };
复制代码
      
                                                     登录后复制                        
  1. #include "sqlite.hpp"
  2. #include <vector>
  3. #include <cassert>
  5. int select_stu_callback(void* arg,int col_count,char** result,char** fields_name) {
  6.     std::vector<std::string> *arry = (std::vector<std::string>*)arg;
  7.     arry->push_back(result[0]);
  8.     return 0; //必须有！！！
  9. }
  10. int main()
  11. {
  12.     SqliteHelper helper("./test.db");
  13.     //1. 创建/打开库文件
  14.     assert(helper.open());
  15.     //2. 创建表(不存在则创建)， 学生信息： 学号，姓名，年龄
  16.     const char *ct = "create table if not exists student(sn int primary key, name varchar(32), age int);";
  17.     assert(helper.exec(ct, nullptr, nullptr));
  18.     //3. 新增数据 ， 修改， 删除， 查询
  19.     // const char *insert_sql = "insert into student values(1, '小明', 18), (2, '小黑', 19), (3, '小红', 18);";
  20.     // assert(helper.exec(insert_sql, nullptr, nullptr));
  21.     // const char *update_sql = "update student set name='张小明' where sn=1";
  22.     // assert(helper.exec(update_sql, nullptr, nullptr));
  23.     // const char *delete_sql = "delete from student where sn=3";
  24.     // assert(helper.exec(delete_sql, nullptr, nullptr));
  26.     const char *select_sql = "select name from student;";
  27.     std::vector<std::string> arry;
  28.     assert(helper.exec(select_sql, select_stu_callback, &arry));
  29.     for (auto &name : arry) {
  30.         std::cout << name << std::endl;
  31.     }
  32.     //4. 关闭数据库
  33.     helper.close();
  34.     return 0;
  35. }
复制代码
      
                       GTest

     使用：
     
     GTest中的断⾔的宏可以分为两类：
     
     常用断言：
                                   登录后复制                        
  1. // bool值检查
  2. ASSERT_TRUE(参数)，期待结果是true
  3. ASSERT_FALSE(参数)，期待结果是false
  4. //数值型数据检查
  5. ASSERT_EQ(参数1，参数2)，传⼊的是需要⽐较的两个数 equal
  6. ASSERT_NE(参数1，参数2)，not equal，不等于才返回true
  7. ASSERT_LT(参数1，参数2)，less than，⼩于才返回true
  8. ASSERT_GT(参数1，参数2)，greater than，⼤于才返回true
  9. ASSERT_LE(参数1，参数2)，less equal，⼩于等于才返回true
  10. ASSERT_GE(参数1，参数2)，greater equal，⼤于等于才返回true
复制代码
      
                       TEST(参数1,参数2){//......}
     参数1是测试名称，第二个参数是详细名称
     RUN_ALL_TESTS();//运行全部测试
                                   登录后复制                        
  1. #include<iostream>
  2. #include<gtest/gtest.h>
  3. int abs(int x)
  4. {
  5.         return x > 0 ? x : -x;
  6. }
  7. TEST(abs_test, test1)
  8. {
  9.         ASSERT_TRUE(abs(1) == 1) << "abs(1)=1";
  10.         ASSERT_TRUE(abs(-1) == 1);
  11.         ASSERT_FALSE(abs(-2) == -2);
  12.         ASSERT_EQ(abs(1),abs(-1));
  13.         ASSERT_NE(abs(-1),0);
  14.         ASSERT_LT(abs(-1),2);
  15.         ASSERT_GT(abs(-1),0);
  16.         ASSERT_LE(abs(-1),2);
  17.         ASSERT_GE(abs(-1),0);
  18. }
  19. int main(int argc,char *argv[])
  20. {
  21.         //将命令⾏参数传递给gtest
  22.         testing::InitGoogleTest(&argc, argv);
  23.         // 运⾏所有测试案例
  24.         return RUN_ALL_TESTS();
  25. }
复制代码
      
                       

     全局事件：针对整个测试程序。实现全局的事件机制，必要创建⼀个⾃⼰的类，然后继承 testing::Environment类，然后分别实现成员函数 SetUp 和 TearDown ，同时在main函数内进⾏调⽤ testing::AddGlobalTestEnvironment(new MyEnvironment); 函数添加全局的事件机制
     全局测试套件，其实就是用户本身界说一个全局的测试环境类
                                   登录后复制                        
  1. #include <gtest/gtest.h>
  2. #include <iostream>
  3. #include <unordered_map>
  4. //全局事件:针对整个测试程序,提供全局事件机制,能够在测试之前配置测试环境数据,测试完毕后清理数据
  5. //先定义环境类,通过继承testing::Environment的派⽣类来完成
  6. //重写的虚函数接⼝SetUp会在测试之前被调⽤; TearDown会在测试完毕后调⽤.
  7. std::unordered_map<std::string,std::string> dict;
  8. class MyMapTest:public testing::Environment
  9. {
  10. public:
  11.     virtual void SetUp() override
  12.     {
  13.         std::cout<<"所有单元开始测试"<<std::endl;
  14.         dict.insert(std::make_pair("hello","你好"));
  15.         dict.insert(std::make_pair("世界","world"));
  16.     }
  17.     virtual void TearDown() override
  18.     {
  19.         std::cout<<"所有单元测试结束"<<std::endl;
  20.         dict.clear();
  21.     }
  22. };
  25. TEST(MyMapTest,test1)
  26. {
  27.     ASSERT_EQ(dict.size(),2);
  28.     dict.erase("hello");
  29. }
  30. TEST(MyMapTest,test2)
  31. {
  32.     ASSERT_EQ(dict.size(),2);
  33. }
  35. int main(int argc,char* argv[])
  36. {
  37.     testing::InitGoogleTest(&argc,argv);
  38.     testing::AddGlobalTestEnvironment(new MyMapTest());
  39.     RUN_ALL_TESTS();
  40.     return 0;
  41. }
复制代码
      
                       TestSuite事件：针对⼀个个测试套件。测试套件的事件机制我们同样必要去创建⼀个类，继承⾃testing::Test ，实现两个静态函数 SetUpTestCase 和 TearDownTestCase ，测试套件的事件机制不必要像全局事件机制⼀样在 main 注册，⽽是必要将我们平常使⽤的 TEST 宏改为 TEST_F 宏。
     
                                   登录后复制                        
  1. #include <iostream>
  2. #include <gtest/gtest.h>
  3. #include <unordered_map>
  4. class MyTest:public testing::Test
  5. {
  6. public:
  7.     static void SetUpTestCase()//所有单元测试开始时初始化
  8.     {
  9.         //假设有一个全局的变量，公共的测试数据就可以在这里初始化
  10.         std::cout<<"所有单元测试开始"<<std::endl;
  11.     }
  12.     virtual void SetUp() override//每次单元测试时初始化
  13.     {
  14.         //假设有一个全局的变量，在这里插入每个单元测试所需的独立的测试数据
  15.         std::cout<<"单元测试开始"<<std::endl;
  16.         mymap.insert(std::make_pair("hello","world"));  
  17.     }
  18.     virtual void TearDown() override//每次单元测试结束时清理
  19.     {
  20.         std::cout<<"单元测试结束"<<std::endl;
  21.         mymap.clear();
  22.     }
  23.     static void TearDownTestCase()//所有单元测试结束时清理
  24.     {
  25.         std::cout<<"所有单元测试结束"<<std::endl;
  26.     }
  27. public:
  28.     std::unordered_map<std::string,std::string> mymap;
  29. };
  31. //必须和类同名
  32. TEST_F(MyTest,test1)
  33. {
  34.     mymap.insert(std::make_pair("aaa","aaa"));
  35.     ASSERT_EQ(mymap.size(),2);
  36. }
  38. TEST_F(MyTest,test2)
  39. {
  40.     ASSERT_EQ(mymap.size(),1);
  41. }
  43. int main(int argc,char* argv[])
  44. {
  45.     testing::InitGoogleTest(&argc,argv);
  46.     RUN_ALL_TESTS();
  47.     return 0;
  48. }
复制代码
      
                       C++11异步操作实现线程池

     怎样获取线程池里的线程完成使命后的结果？
     std::future：获取使命结果
     std::future是C++11标准库中的⼀个模板类，它表⽰q⼀个异步操作的结果。当我们在多线程编程中使⽤异步使命时，std::future可以帮助我们在必要的时候获取使命的执⾏结果。std::future的⼀个重要特性是可以或许壅闭当前线程，直到异步操作完成，从⽽确保我们在获取结果时不会遇到未完成的操作。
     asyn
                                   登录后复制                        
  1. #include <iostream>
  2. #include <future>
  3. #include <thread>
  4. #include <chrono>
  6. int add(int num1,int num2)
  7. {
  8.     std::cout<<"aaaaaaaaaaaaaaaaaaaaa"<<std::endl;
  9.     std::cout<<"加法任务"<<std::endl;
  10.      std::this_thread::sleep_for(std::chrono::seconds(5));
  11.     std::cout<<"bbbbbbbbbbbbbbbbbbbbb"<<std::endl;
  12.     return num1+num2;
  13. }
  14. int main()
  15. {
  16.     //std::async(func, ...)      std::async(policy, func, ...)
  17.     //std::launch::deferred  在执行get获取异步结果的时候，才会执行异步任务
  18.     //std::launch::async   内部会创建工作线程，异步的完成任务
  19.     std::cout<<"----------1----------"<<std::endl;
  20.     std::future<int> result=std::async(std::launch::deferred,add,2,3);//把任务交它，它会创建一个工作线程来执行任务
  21.     std::this_thread::sleep_for(std::chrono::seconds(1));
  22.     std::cout<<"----------2----------"<<std::endl;
  23.     int ret=result.get();//此时如果任务还未完成，会阻塞等待
  24.     std::cout<<"----------3----------"<<std::endl;
  25.     std::cout<<"result="<<ret<<std::endl;
  26.     return 0;
  27. }
复制代码
      
                       promise
                                   登录后复制                        
  1. #include <future>
  2. #include <thread>
  3. #include <chrono>
  4. #include <iostream>
  5. //通过在线程中对promise对象设置数据，其他线程中通过future获取设置数据的方式实现获取异步任务执行结果的功能
  6. void add(int num1,int num2,std::promise<int>& prom)
  7. {
  8.     prom.set_value(num1+num2);
  9.     std::this_thread::sleep_for(std::chrono::seconds(3));
  10. }
  11. int main()
  12. {
  13.     std::promise<int> prom;
  14.     std::future<int> fut=prom.get_future();//二者从此绑定了关系
  15.     std::thread th(add,11,12,std::ref(prom));//创建线程执行任务
  16.     std::cout<<fut.get()<<std::endl;//get会进行同步，等任务完成了才得到结果
  17.     th.join();//等待线程
  18.     return 0;
  19. }
复制代码
      
                       packaged_task
                                   登录后复制                        
  1. #include <iostream>
  2. #include <future>
  3. #include <memory>
  4. #include <thread>
  5. //pakcaged_task的使用
  6. //pakcaged_task 是一个模板类，实例化的对象可以对一个函数进行二次封装，
  7. //pakcaged_task可以通过get_future获取一个future对象，来获取封装的这个函数的异步执行结果
  8. int add(int num1,int num2)
  9. {
  10.     return num1+num2;
  11. }
  12. int main()
  13. {
  14.     //task(11, 22);  task可以当作一个可调用对象来调用执行任务
  15.     //但是它又不能完全的当作一个函数来使用
  16.     //std::async(std::launch::async, task, 11, 22);
  17.     //std::thread thr(task, 11, 22);
  19.     //但是我们可以把task定义成为一个指针，传递到线程中，然后进行解引用执行
  20.     //但是如果单纯指针指向一个对象，存在生命周期的问题，很有可能出现风险
  21.     //思想就是在堆上new对象，用智能指针管理它的生命周期
  22.     auto pTask=std::make_shared<std::packaged_task<int(int,int)>>(add);//make_shared效率更高
  23.     std::future<int> fut=pTask->get_future();//建立同步关系
  24.     std::thread th([pTask](){
  25.         (*pTask)(1,2);
  26.     });
  27.     std::cout<<"结果为："<<fut.get()<<std::endl;
  28.     th.join();
  29.     return 0;
  30. }
复制代码
      
                       实现线程池
     packaged_task+futur
                                   登录后复制                        
  1. #include <thread>
  2. #include <future>
  3. #include <vector>
  4. #include <mutex>
  5. #include <atomic>
  6. #include <condition_variable>
  7. #include <functional>
  8. #include <memory>
  9. #include <iostream>
  11. class ThreadPool
  12. {
  13. public:
  14.     using Func = std::function<void(void)>;
  15.     ThreadPool(int threadCount = 1) // 构造函数
  16.         : _stop(false)
  17.     {
  18.         for (int i = 0; i < threadCount; ++i) // 创建指定数量的线程
  19.         {
  20.             _threads.emplace_back(std::thread(&ThreadPool::entry, this));
  21.         }
  22.     }
  23.     ~ThreadPool() // 析构函数
  24.     {
  25.         stop();
  26.     }
  27.     template <class F, class... Args>
  28.     // func是用户传过来的函数，接下来就是不定参数
  29.     // push函数会把func封装成一个异步任务(packaged_task)，抛入任务池中，由工作线程取出处理
  30.     // push函数会返回一个future对象，让用户能得到任务结果
  31.     auto push(F &&func, Args &&...args) -> std::future<decltype(func(args...))> // 用户添加任务
  32.     {
  33.         // 1.将用户发过来的函数封装成packaged_task
  34.         using retType = decltype(func(args...));
  35.         auto tmp_func = std::bind(std::forward<F>(func), std::forward<Args>(args)...); // 将参数绑定，注意使用完美转发
  36.         auto task = std::make_shared<std::packaged_task<retType()>>(tmp_func);
  37.         std::future<retType> fut = task->get_future();
  38.         // 2.构造匿名lambda表达式，里面执行task
  39.         {
  40.             std::unique_lock<std::mutex> lock(_mutex); // 加锁
  41.             // 将lambda表达式抛入任务池中
  42.             _tasks.push_back([task]()
  43.                            { (*task)(); });
  44.             _cond.notify_one(); // 唤醒一个线程
  45.         }
  46.         return fut;
  47.     }
  48.     void stop() // 线程池停止工作
  49.     {
  50.         // 0.如果已经停止
  51.         if (_stop == true)
  52.             return;
  53.         // 1.将停止标志设为true
  54.         _stop = true;
  55.         // 2.唤醒所有线程
  56.         _cond.notify_all();
  57.         // 3.等待所有线程
  58.         for (auto &th : _threads)
  59.         {
  60.             th.join();
  61.         }
  62.     }
  64. private:
  65.     // 线程入口函数，内部不断从任务池中取出任务执行
  66.     void entry()
  67.     {
  68.         while (!_stop)
  69.         {
  70.             std::vector<Func> tmpTasks;
  71.             {
  72.                 std::unique_lock<std::mutex> lock(_mutex); // 加锁
  73.                 _cond.wait(lock, [this]()
  74.                            { return _stop || !_tasks.empty(); }); // 等待唤醒
  75.                 tmpTasks.swap(_tasks);//取出任务
  76.             }
  77.             std::cout<<"线程开始执行任务"<<std::endl;
  78.             for(auto& task:tmpTasks)//执行任务
  79.             {
  80.                 task();
  81.             }
  82.         }
  83.     }
  85. private:
  86.     std::atomic<bool> _stop;           // 线程池是否停止运行
  87.     std::mutex _mutex;                 // 互斥
  88.     std::condition_variable _cond;     // 同步
  89.     std::vector<Func> _tasks;          // 任务池
  90.     std::vector<std::thread> _threads; // 线程池
  91. };
复制代码
      
                       需求分析

     1、焦点概念

     其中， Broker Server是最核⼼的部门, 负责消息的存储和转发。
     ⽽在AMQP(Advanced Message Queuing Protocol-⾼级消息队列协议，⼀个提供统⼀消息服务的应⽤层标准⾼级消息队列协议，为⾯向消息的中间件设计，使得服从该规范的客⼾端应⽤和消息中间件服 务器的全功能互操作成为可能)模型中，也就是消息中间件服务器Broker中，⼜存在以下概念：
     
     所谓的 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 来说, 主要⽀持四种交换机范例：
     
     其中 Header 这种⽅式⽐较复杂, ⽐较少⻅。常⽤的是前三种交换机范例，项⽬中也主要实现这三种
     
     这三种操作就像给 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 中可以包含多个 Channel。Channel 和 Channel 之间的数据是独⽴的，不会相互⼲扰。如许做主要是为了可以或许更好的复⽤ TCP 连接, 达到⻓连接的结果, 避免频仍的创建关闭 TCP 连接。
     Connection 可以理解成⼀根⽹线. Channel 则是⽹线⾥详细的线缆.
     

     6、消息应答

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

     1、服务端模块

     1.1、数据管理模块

     1.1.1、交换机数据管理模块

     要管理的数据：描述了一个交换机应该有什么数据
     
           直接交换：binding_key与routing_key相同时，则将消息放入队列
      广播交换：将消息放入交换机绑定的全队伍列中
      主题交换：routing_key与多个绑定队列的binding_key有个匹配规则，匹配成功了则放入
         
     对交换机的管理操作：
     
     1.1.2、队列数据管理模块

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

     描述一下将哪个队列与哪个交换机绑定到了一起
     管理的数据：
     
           由数字、字符、_、#、。、*构成      binding_key:  news.music.#       消息中的routing_key:   news.music.pop
          管理的操作：
     
     1.1.4、消息数据管理模块

     
     
     
     
     以下是服务端为了管理所添加的信息
     
     
     
     
     
     
     
     
     以上四个模块分别实现数据的管理(增删查)，以及持久化的存储
     1.2、假造机数据管理模块

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

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

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

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

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

     对以上全部模块的整合，整合成一个服务器
          管理信息：
     
     2、客户端模块

     2.1、消费者管理模块

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

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

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

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

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

     项目创建

     

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

     1、实现日志工具

                                   登录后复制                        
  1. #pragma once
  2. #include <iostream>
  3. #include <ctime>
  5. namespace lsxmq
  6. {
  7. #define DBG_LEVEL 0
  8. #define INF_LEVEL 1
  9. #define ERR_LEVEL 2
  10. #define DEFAULT_LEVEL DBG_LEVEL
  11. #define LOG(levelStr, level, format, ...)                                                                \
  12.     {                                                                                                    \
  13.         if (level >= DEFAULT_LEVEL)                                                                      \
  14.         {                                                                                                \
  15.             time_t t = time(nullptr);                                                                    \
  16.             struct tm *ptm = localtime(&t);                                                              \
  17.             char timeStr[32];                                                                            \
  18.             strftime(timeStr, 31, "%H:%M:%S", ptm);                                                      \
  19.             printf("[%s][%s][%s:%d]" format "\n", levelStr, timeStr, __FILE__, __LINE__, ##__VA_ARGS__); \
  20.         }                                                                                                \
  21.     }
  23.     //__VA_ARGS__是使用可变参数，##是为了保证没有可变参数时忽略前面的逗号
  24.     // 宏函数必须在一行里，在换行时加上\是为了转义\n
  26. #define DLOG(format, ...) LOG("DBG", DBG_LEVEL, format, ##__VA_ARGS__)
  27. #define ILOG(format, ...) LOG("INF", INF_LEVEL, format, ##__VA_ARGS__)
  28. #define ELOG(format, ...) LOG("ERR", ERR_LEVEL, format, ##__VA_ARGS__)
  30. }
复制代码
      
                       2、实现使用Helper工具

     2.1、Sqlite基础操作类

                                   登录后复制                        
  1. #pragma once
  2. // 封装实现一个SqliteHelper类，提供简单的sqlite数据库操作接口，完成对数据增删改查操作
  3.     // 1.创建/打开数据库文件
  4.     // 2.针对打开的数据库进行操作
  5.     // 2.1表的操作
  6.     // 2.2数据的操作
  7.     // 3.关闭数据库
  8.     class SqliteHelper
  9.     {
  10.     public:
  11.         typedef int (*SqliteCallback)(void *, int, char **, char **);
  12.         SqliteHelper(std::string dbfile)
  13.             : _dbfile(dbfile)
  14.         {
  15.         }
  16.         bool open(int safeLevel = SQLITE_OPEN_FULLMUTEX) // 打开数据库文件
  17.         {
  18.             // 默认安全级别为串行化模式
  19.             // int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs )
  20.             int ret = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE | safeLevel, nullptr);
  21.             if (ret != SQLITE_OK)
  22.             {
  23.                 ELOG("创建/打开数据库文件失败：%s",sqlite3_errmsg(_handler));
  24.                 return false;
  25.             }
  26.             return true;
  27.         }
  28.         bool exec(const std::string &sql, SqliteCallback cb, void *arg) // 进行操作
  29.         {
  30.             // int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**),void* arg, char **err)
  31.             int ret = sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr);
  32.             if (ret != SQLITE_OK)
  33.             {
  34.                 ELOG("%s执行失败：%s",sql.c_str(),sqlite3_errmsg(_handler));
  35.                 return false;
  36.             }
  37.             return true;
  38.         }
  39.         void close() // 关闭数据库文件
  40.         {
  41.             // int sqlite3_close_v2(sqlite3*)
  42.             sqlite3_close_v2(_handler);
  43.         }
  45.     private:
  46.         sqlite3 *_handler;   // 句柄
  47.         std::string _dbfile; // 数据库文件名
  48.     };
复制代码
      
                       2.2、字符串操作类

     分割字符串功能：
                                   登录后复制                        
  1. static size_t split(const std::string& str,const std::string& sep,std::vector<std::string>& vs)
  2.     {
  3.         size_t idx=0;
  4.         while(idx<str.size())
  5.         {
  6.             size_t pos=str.find(sep,idx);
  7.             //找不到分隔符了
  8.             if(pos==std::string::npos)
  9.             {
  10.                 vs.push_back(str.substr(idx,str.size()-idx));
  11.                 return vs.size();
  12.             }
  13.             //pos的位置是一个分割符，而idx是上一个分割符的下一个位置
  14.             if(idx==pos)//意味着出现连续分割符，中间没有有效子串
  15.             {
  16.                 idx+=sep.size();
  17.                 continue;
  18.             }
  19.             vs.push_back(str.substr(idx,pos-idx));
  20.             idx=pos+sep.size();
  21.         }
  22.         return vs.size();
  23.     }
复制代码
      
                       2.3、UUID天生器类

     1.天生8个0~255之间的随机数
     2.取出一个8字节的序号
     通过以上数据，构成16字节的数据，转换为16进制字符，共32个
     随机数的天生
                                   登录后复制                        
  1. class UUIDHelper
  2.     {
  3.     public:
  4.         static std::string uuid()
  5.         {
  6.             std::random_device rd; // 这是一个随机数生成装置
  7.             // rd();//这样就可以生成机器随机数，但是生成机器随机数效率太低，所以不使用这种方法
  8.             // 实际上是把这个机器随机数作为随机数种子来生成伪随机数
  9.             std::mt19937_64 gernator(rd()); // 使用梅森旋转算法，生成一个伪随机数
  10.             // 因为要生成的随机数范围是0~255，所以需要限定随机数的生成范围
  11.             std::uniform_int_distribution<int> distribution(0, 255);
  12.             std::stringstream uuid;
  13.             for (int i = 0; i < 8; ++i) // 生成16位16进制字符
  14.             {
  15.                 int number = distribution(gernator); // 生成随机数
  16.                 uuid << std::setw(2) << std::setfill('0') << std::hex << number;
  17.                 if (i == 3 || i == 5 || i == 7)
  18.                 {
  19.                     uuid << "-";
  20.                 }
  21.             }
  22.             // 生成序号
  23.             static std::atomic<size_t> seq(0);
  24.             size_t tmpSeq = seq.fetch_add(1); // 每次加一
  25.             for (int i = 7; i >= 0; --i)
  26.             {
  27.                 uuid << std::setw(2) << std::setfill('0') << std::hex << (tmpSeq >> (i * 8) & 0xff);
  28.                 if (i == 6)
  29.                 {
  30.                     uuid << "-";
  31.                 }
  32.             }
  33.             return uuid.str();
  34.         }
  35.     };
复制代码
      
                       2.3、文件操作类

                                   登录后复制                        
  1. class FileHelper
  2.     {
  3.     public:
  4.         FileHelper(const std::string& filename)
  5.             :_filename(filename)
  6.         {}
  7.         bool exists()//文件是否存在
  8.         {
  9.             struct stat st;
  10.             //返回值为0表示能获取文件信息，文件自然就是存在的
  11.             return stat(_filename.c_str(),&st)==0;
  12.         }
  13.         size_t size()//文件大小
  14.         {
  15.             struct stat st;
  16.             if(stat(_filename.c_str(),&st)!=0)
  17.             {
  18.                 return 0;
  19.             }
  20.             return st.st_size;
  21.         }
  22.         bool read(char* body,size_t offset,size_t len)//从文件指定位置读取指定长度内容
  23.         {
  24.             //1.打开文件
  25.             std::ifstream ifs(_filename.c_str(),std::ios_base::in|std::ios_base::binary);
  26.             if(ifs.is_open()==false)
  27.             {
  28.                 ELOG("%s文件打开失败",_filename.c_str());
  29.                 ifs.close();
  30.                 return false;
  31.             }
  32.             //2.移动到指定位置
  33.             ifs.seekg(offset,std::ios_base::beg);
  34.             //3.读取文件内容
  35.             ifs.read(&body[0],len);
  36.             if(ifs.good()==false)
  37.             {
  38.                 ELOG("%s文件读取失败",_filename.c_str());
  39.                 ifs.close();
  40.                 return false;
  41.             }
  42.             //4.关闭文件
  43.             ifs.close();
  44.             return true;
  45.         }
  46.         bool read(std::string& body)//从文件读取内容
  47.         {
  48.             size_t fsize=this->size();
  49.             //根据fsize调整body的大小
  50.             body.resize(fsize);
  51.             return read(&body[0],0,fsize);
  52.         }
  53.         bool write(const char* body,size_t offset,size_t len)//向文件指定位置写入内容
  54.         {
  55.             //1.打开文件
  56.             std::fstream fs(_filename.c_str(),std::ios_base::in|std::ios_base::out|std::ios_base::binary);
  57.             //这里不能ofstream
  58.             //想要跳转到文件的指定位置必须有读权限，istream可以fstream可以，但是ofstream不可以
  59.             if(fs.is_open()==false)
  60.             {
  61.                 ELOG("%s文件打开失败",_filename.c_str());
  62.                 fs.close();
  63.                 return false;
  64.             }
  65.             //2.移动到指定位置
  66.             fs.seekg(offset,std::ios_base::beg);
  67.             //3.向文件写入内容
  68.             fs.write(body,len);
  69.             if(fs.good()==false)
  70.             {
  71.                 ELOG("%s文件写入失败",_filename.c_str());
  72.                 fs.close();
  73.                 return false;
  74.             }
  75.             //4.关闭文件
  76.             fs.close();
  77.             return true;
  78.         }
  79.         bool write(const std::string& body)//向文件写入内容
  80.         {
  81.             return write(&body[0],0,body.size());
  82.         }
  83.         static std::string parentDirectory(const std::string& filename)//返回当前文件的父路径
  84.         {
  85.             //   aaa/bbb/ccc/ddd/e.txt
  86.             int pos=filename.find_last_of('/');
  87.             if(pos==std::string::npos)//没找到
  88.             {
  89.                 //e.txt
  90.                 return "./";
  91.             }
  92.             return filename.substr(0,pos+1);
  93.         }
  94.         bool rename(const std::string& filename)//修改文件名称
  95.         {
  96.             return (::rename(_filename.c_str(),filename.c_str())==0);
  97.         }
  98.         static bool createFile(const std::string& filename)//创建文件
  99.         {
  100.             std::ofstream ofs(filename.c_str(),std::ios_base::binary|std::ios_base::out|std::ios_base::app);
  101.             if(ofs.is_open()==false)
  102.             {
  103.                 ELOG("%s文件创建失败",filename.c_str());
  104.                 ofs.close();
  105.                 return false;
  106.             }
  107.             ofs.close();
  108.             return true;
  109.         }
  110.         static bool removeFile(const std::string& filename)//删除文件
  111.         {
  112.             return (::remove(filename.c_str())==0);
  113.         }
  114.         static bool createDirectory(const std::string& path)//创建目录
  115.         {
  116.             size_t idx=0;
  117.             while(idx<path.size())
  118.             {
  119.                 int pos=path.find('/',idx);
  120.                 if(pos==std::string::npos)
  121.                 {
  122.                     if(mkdir(path.c_str(),0775)!=0)
  123.                     {
  124.                         ELOG("%s目录创建失败",path.c_str());
  125.                         return false;
  126.                     }
  127.                     return true;
  128.                 }
  129.                 std::string subPath=path.substr(0,pos);
  130.                 if(mkdir(subPath.c_str(),0775)!=0)
  131.                 {
  132.                     ELOG("%s目录创建失败",subPath.c_str());
  133.                     return false;
  134.                 }
  135.                 idx=pos+1;
  136.             }
  137.             return true;
  138.         }
  139.         static bool removeDirectory(const std::string& path)//删除目录
  140.         {
  141.             //rmdir必须保证文件夹是空的，所以不用这种方法
  142.             std::string cmd="rm -rf "+path;
  143.             return system(cmd.c_str())==0;
  144.         }
  145.         
  146.     private:
  147.         std::string _filename;
  148.     };
复制代码
      
                       3、消息范例界说与交换机范例界说

                                   登录后复制                        
  1. syntax= "proto3";
  2. package lsx;
  4. enum ExchangType//交换机类型
  5. {
  6.     UNKNOWTYPE=0;//因为必须从0开始，所以加一个未知类型
  7.     DIRECT=1;//直接
  8.     FANOUT=2;//广播
  9.     TOPIC=3;//主题
  10. };
  12. enum DeliveryMode//消息投递方式
  13. {
  14.     UNKONWMODE=0;
  15.     DURABLE=1;   //持久化
  16.     UNDURABLE=2;   //非持久化
  17. };
  19. message BasicProperties //基本特性
  20. {
  21.     string id=1;   //消息ID
  22.     DeliveryMode deliveryMode=2;   //是否持久化
  23.     string routing_key=3;
  24. };
  26. message Message
  27. {
  28.     message Payload//有效载荷
  29.     {
  30.         BasicProperties properties=1;   //基本属性
  31.         string body=2;   //消息主体
  32.         string valid=3;   //是否有效
  33.         //消息是否有效不用bool类型，而是使用字符的0/1
  34.         //因为bool类型序列化时true和false所占长度不同
  35.         //会导致修改文件中消息有效位后消息长度发生变化
  36.     }
  37.     Payload payload=1;   //有效载荷
  38.     int32 offset=2;   //偏移量
  39.     int32 len=3;   //长度
  40. };
复制代码
      
                       4、交换机数据管理

      界说交换机数据类
     
     界说交换机数据持久化类(数据持久化的sqlite3数据库中)
     
     界说交换机数据管理类
     
                                   登录后复制                        
  1. namespace lsxmq
  2. {
  3.     //1.定义交换机类
  4.     struct Exchange
  5.     {
  6.         using ptr=std::shared_ptr<Exchange>;
  8.         std::string name;//交换机名称
  9.         ExchangeType type;//交换机类型
  10.         bool durable;//是否持久化
  11.         bool autoDel;//是否自动删除
  12.         std::unordered_map<std::string,std::string> args;//其它参数
  14.         Exchange(const std::string& ename,ExchangeType etype
  15.         ,bool edurable,bool eautoDel
  16.         ,std::unordered_map<std::string,std::string> eargs)
  17.             :name(ename),type(etype),durable(edurable)
  18.             ,autoDel(eautoDel),args(eargs)
  19.         {}
  21.         //args存储的是键值对，在存储数据库的时候，会组织一个格式进行存储：key=value&key=value
  22.         //将字符串反序列化为args
  23.         void setArgs(std::string strArgs);
  24.         //将args序列化为字符串
  25.         std::string getArgs();
  26.     };
  28.     //2.定义交换机数据持久化存储类--数据存储在sqlite数据库中
  29.     class ExchangeMapper
  30.     {
  31.     public:
  32.         ExchangeMapper(const std::string dbfile)
  33.             :_sqliteHelper(dbfile)
  34.         {}
  35.         bool createTable();//创建表
  36.         bool removeTable();//删除表
  37.         bool insert(Exchange::ptr& exchang);//向数据库插入一条交换机数据
  38.         bool erase(Exchange::ptr& exchang);//从数据库中删除一条交换机数据
  39.         std::unordered_map<std::string,Exchange::ptr> getAll();//获取数据库中所有的交换机数据
  40.         //name:ptr
  41.     private:
  42.         SqliteHelper _sqliteHelper;
  43.     };
  45.     //3.定义交换机数据管理类
  46.     class ExchangeManager
  47.     {
  48.     public:
  49.         ExchangeManager(const std::string dbfile)
  50.             :_mapper(dbfile)
  51.         {}
  52.         bool declareExchange(const std::string& ename,ExchangeType etype
  53.         ,bool edurable,bool eautoDel
  54.         ,std::unordered_map<std::string,std::string> eargs);//声明交换机
  55.         bool deleteExchange(const std::string& name);//删除交换机
  56.         Exchange::ptr selectExchange(const std::string& name);//获取指定交换机
  57.         bool exists(const std::string& anme);//指定交换机是否存在
  58.         void clear();//清理数据
  59.     private:
  60.         std::mutex _mutex;
  61.         ExchangeMapper _mapper;
  62.         std::unordered_map<std::string,Exchange::ptr> _exchanges;
  63.         
  64.     };
  65. };
复制代码
      
                       5、队列数据的管理

     当前队列数据的管理，本质上是队列描述信息的管理，描述当前服务器上有哪些队列。
     界说队列描述数据类
     
     界说队列数据持久化类(数据持久化的sqlite3数据库中)
     
     界说队列数据管理类
     
                                   登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/helper.hpp"
  3. #include "../mqcomm/logger.hpp"
  4. #include <unordered_map>
  5. #include <memory>
  6. #include <cassert>
  7. #include <mutex>
  9. namespace lsxmq
  10. {
  11.     // 1.定义消息队列类
  12.     struct MessageQueue
  13.     {
  14.         using ptr = std::shared_ptr<MessageQueue>;
  15.         std::string name;                                  // 队列名称
  16.         bool durable;                                      // 是否持久化
  17.         bool exclusive;                                    // 是否独占
  18.         bool autoDel;                                      // 是否自动删除
  19.         std::unordered_map<std::string, std::string> args; // 其它参数
  21.         MessageQueue() {}
  22.         MessageQueue(const std::string &qname, bool qdurable, bool qexclusive, bool qautoDel, std::unordered_map<std::string, std::string> qargs)
  23.             : name(qname), durable(qdurable), exclusive(qexclusive), autoDel(qautoDel), args(qargs)
  24.         {
  25.         }
  27.         // args存储的是键值对，在存储数据库的时候，会组织一个格式进行存储：key=value&key=value
  28.         // 将字符串反序列化为args
  29.         void setArgs(std::string strArgs)
  30.         {
  31.             std::vector<std::string> subStr;
  32.             StrHelper::split(strArgs, "&", subStr); // 分割字符串
  33.             for (auto &str : subStr)                // 继续分割
  34.             {
  35.                 int pos = str.find("=");
  36.                 if (pos != std::string::npos) // 加入args
  37.                 {
  38.                     std::string first = str.substr(0, pos);
  39.                     std::string second = str.substr(pos + 1, str.size());
  40.                     args.insert(std::make_pair(first, second));
  41.                 }
  42.             }
  43.         }
  44.         // 将args序列化为字符串
  45.         std::string getArgs()
  46.         {
  47.             std::string res;
  48.             for (auto &a : args)
  49.             {
  50.                 res += a.first;
  51.                 res += "=";
  52.                 res += a.second;
  53.                 res += "&";
  54.             }
  55.             res.pop_back();
  56.             return res;
  57.         }
  58.     };
  59.     // 2.定义消息队列数据持久化存储类--数据存储在sqlite数据库中
  60.     using MessageQueueMap = std::unordered_map<std::string, MessageQueue::ptr>;
  61.     class MessageQueueMapper
  62.     {
  63.     public:
  64.         MessageQueueMapper(const std::string &dbfile)
  65.             : _sqliteHelper(dbfile)
  66.         {
  67.             std::string path = FileHelper::parentDirectory(dbfile);
  68.             if (FileHelper(path).exists() == false) // 父目录不存在就创建
  69.             {
  70.                 FileHelper::createDirectory(path);
  71.             }
  72.             assert(_sqliteHelper.open());
  73.             this->createTable();
  74.         }
  75.         void createTable() // 创建一张消息队列数据管理表
  76.         {
  77. #define CREATE_TABLE "create table if not exists messageQueueTable(\
  78.             name varchar(32) primary key,\
  79.             durable int,\
  80.             exclusive int,\
  81.             autoDel int,\
  82.             args varchar(128));"
  83.             bool ret = _sqliteHelper.exec(CREATE_TABLE, nullptr, nullptr);
  84.             if (ret == false)
  85.             {
  86.                 ELOG("消息队列数据库表创建失败");
  87.                 abort();
  88.             }
  89.         }
  90.         void removeTable() // 删除表
  91.         {
  92. #define DROP_TABLE "drop table messageQueueTable"
  93.             bool ret = _sqliteHelper.exec(DROP_TABLE, nullptr, nullptr);
  94.             if (ret == false)
  95.             {
  96.                 ELOG("消息队列数据库表删除失败");
  97.                 abort();
  98.             }
  99.         }
  100.         bool insert(MessageQueue::ptr& mqp) // 向数据库插入一条消息队列数据
  101.         {
  102. #define INSERT "insert into messageQueueTable values('%s',%d,%d,%d,'%s');"
  103.             char sql[256] = {0};
  104.             sprintf(sql, INSERT, mqp->name.c_str(), mqp->durable, mqp->exclusive, mqp->autoDel, mqp->getArgs().c_str());
  105.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  106.         }
  107.         bool erase(const std::string &qname) // 从数据库中删除一条消息队列数据
  108.         {
  109. #define ERASE "delete from  messageQueueTable where name='%s';"
  110.             char sql[256] = {0};
  111.             sprintf(sql, ERASE, qname.c_str());
  112.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  113.         }
  114.         MessageQueueMap recovery() // 获取数据库中所有的消息队列数据(恢复)
  115.         {
  116. #define SELECT_ALL "select name,durable,exclusive,autoDel,args from  messageQueueTable"
  117.             MessageQueueMap result;
  118.             bool ret = _sqliteHelper.exec(SELECT_ALL, sqliteCallback, &result);
  119.             if (ret == false)
  120.             {
  121.                 ELOG("恢复消息队列数据失败");
  122.                 abort();
  123.             }
  124.             return result;
  125.         }
  126.         ~MessageQueueMapper()
  127.         {
  128.             _sqliteHelper.close();
  129.         }
  130.     private:
  131.         static int sqliteCallback(void *arg, int col, char **row, char **fields)
  132.         {
  133.             MessageQueueMap *result = (MessageQueueMap *)arg;
  134.             auto mqp = std::make_shared<MessageQueue>();
  135.             mqp->name = row[0];
  136.             mqp->durable = (bool)std::stoi(row[2]);
  137.             mqp->exclusive = (bool)std::stoi(row[2]);
  138.             mqp->autoDel = (bool)std::stoi(row[3]);
  139.             if (row[4])
  140.                 mqp->setArgs(row[4]); // 其它参数可能不存在
  141.             result->insert(std::make_pair(mqp->name, mqp));
  142.             return 0;
  143.         }
  144.     private:
  145.         SqliteHelper _sqliteHelper;
  146.     };
  147.     // 3.定义消息队列数据管理类
  148.     class MessageQueueManager
  149.     {
  150.     public:
  151.         using ptr = std::shared_ptr<MessageQueueManager>;
  152.         MessageQueueManager(const std::string dbfile)
  153.             : _mapper(dbfile)
  154.         {
  155.             _messageQueues = _mapper.recovery(); // 恢复数据
  156.         }
  157.         bool declareMessageQueue(const std::string &name, bool durable,
  158.         bool exclusive, bool autoDel, std::unordered_map<std::string, std::string> args) // 声明消息队列
  159.         {
  160.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  161.             auto it = _messageQueues.find(name);
  162.             if (it != _messageQueues.end()) // 已经存在了
  163.             {
  164.                 return true;
  165.             }
  166.             auto mqp = std::make_shared<MessageQueue>(name, durable,exclusive, autoDel, args);
  167.             if (durable == true) // 需要持久化
  168.             {
  169.                 bool ret = _mapper.insert(mqp);
  170.                 if (ret == false)
  171.                     return false;
  172.             }
  173.             _messageQueues.insert(std::make_pair(name, mqp));
  174.             return true;
  175.         }
  176.         bool deleteMessageQueue(const std::string &name) // 删除消息队列
  177.         {
  178.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  179.             auto it = _messageQueues.find(name);
  180.             if (it == _messageQueues.end()) // 不存在
  181.             {
  182.                 return true;
  183.             }
  184.             if (it->second->durable == true) // 从数据库中删除
  185.             {
  186.                 bool ret = _mapper.erase(it->second->name);
  187.                 if (ret == false)
  188.                     return false;
  189.             }
  190.             _messageQueues.erase(name);
  191.             return true;
  192.         }
  193.         MessageQueue::ptr selectMessageQueue(const std::string &name) // 获取指定消息队列
  194.         {
  195.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  196.             auto it = _messageQueues.find(name);
  197.             if (it == _messageQueues.end()) // 不存在
  198.             {
  199.                 return MessageQueue::ptr();
  200.             }
  201.             return it->second;
  202.         }
  203.         MessageQueueMap getall()//获取所有消息队列
  204.         {
  205.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  206.             return _messageQueues;
  207.         }
  208.         bool exists(const std::string &name) // 指定交换机是否存在
  209.         {
  210.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  211.             auto it = _messageQueues.find(name);
  212.             return it != _messageQueues.end();
  213.         }
  214.         void clear() // 清理数据
  215.         {
  216.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  217.             _mapper.removeTable();                     // 先删磁盘内容
  218.             _messageQueues.clear();                    // 再删内存内容
  219.         }
  220.         size_t size()
  221.         {
  222.             std::unique_lock<std::mutex> lock(_mutex); // 先加锁
  223.             return _messageQueues.size();
  224.         }
  225.     private:
  226.         std::mutex _mutex; // 多线程使用，注意线程安全
  227.         MessageQueueMapper _mapper;
  228.         MessageQueueMap _messageQueues;
  229.     };
  230. };
复制代码
      
                       6、绑定信息(交换机-消息队列)管理

     绑定信息，本质上就是⼀个交换构造联了哪些队列的描述。
     界说绑定信息类
     
     界说绑定信息数据持久化类(数据持久化的sqlite3数据库中)
     
      界说绑定信息数据管理类
     
                                   登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/helper.hpp"
  3. #include "../mqcomm/logger.hpp"
  4. #include <unordered_map>
  5. #include <memory>
  6. #include <cassert>
  7. #include <mutex>
  9. namespace lsxmq
  10. {
  11.     //1.定义绑定类
  12.     struct Binding
  13.     {
  14.         using ptr=std::shared_ptr<Binding>;
  15.         std::string exchangeName;//交换机名称
  16.         std::string messageQueueName;//消息队列名称
  17.         std::string binding_key;
  18.         bool durable;//是否持久化
  19.         Binding(){}
  20.         Binding(const std::string& ename,const std::string& qname,const std::string& bdkey,bool bdurable)
  21.             :exchangeName(ename),messageQueueName(qname),binding_key(bdkey),durable(bdurable)
  22.         {}
  23.     };
  25.     //2.定义持久化管理绑定类
  27.     //交换机名称-->多个消息队列
  28.     //消息队列-->多个绑定信息
  29.     using MessageQueueMap=std::unordered_map<std::string,Binding::ptr>;
  30.     using BindingMap=std::unordered_map<std::string,MessageQueueMap>;
  32.     class BindingMapper
  33.     {
  34.     public:
  35.         BindingMapper(const std::string dbfile)
  36.             :_sqliteHelper(dbfile)
  37.         {
  38.             std::string path = FileHelper::parentDirectory(dbfile);
  39.             if (FileHelper(path).exists() == false) // 父目录不存在就创建
  40.             {
  41.                 FileHelper::createDirectory(path);
  42.             }
  43.             assert(_sqliteHelper.open());
  44.             this->createTable();
  45.         }
  46.         void createTable()//创建表
  47.         {
  48.             #define CREATE_TABLE "create table if not exists bindingTable(\
  49.             exchangeName varchar(32),\
  50.             messageQueueName varchar(32),\
  51.             binding_key varchar(128),\
  52.             durable int);"
  53.             bool ret = _sqliteHelper.exec(CREATE_TABLE, nullptr, nullptr);
  54.             if (ret == false)
  55.             {
  56.                 ELOG("绑定数据库表创建失败");
  57.                 abort();
  58.             }
  59.         }
  60.         void removeTable()//删除表
  61.         {
  62.             #define DROP_TABLE "drop table bindingTable"
  63.             bool ret = _sqliteHelper.exec(DROP_TABLE, nullptr, nullptr);
  64.             if (ret == false)
  65.             {
  66.                 ELOG("绑定数据库表删除失败");
  67.                 abort();
  68.             }
  69.         }
  70.         bool insert(Binding::ptr& bdp)//插入绑定数据
  71.         {
  72.             #define INSERT "insert into bindingTable values('%s','%s','%s',%d);"
  73.             char sql[256] = {0};
  74.             sprintf(sql, INSERT, bdp->exchangeName.c_str(),bdp->messageQueueName.c_str(),bdp->binding_key.c_str(),bdp->durable);
  75.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  76.         }
  77.         bool erase(const std::string& ename,const std::string qname)//删除绑定数据数据
  78.         {
  79.             #define ERASE "delete from  bindingTable where exchangeName='%s' and messageQueueName='%s';"
  80.             char sql[256] = {0};
  81.             sprintf(sql, ERASE, ename.c_str(),qname.c_str());
  82.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  83.         }
  84.         bool removeExchangeBinding(const std::string& ename)//删除交换机的绑定数据
  85.         {
  86.             #define ERASE_EXCHANGE "delete from  bindingTable where exchangeName='%s';"
  87.             char sql[256] = {0};
  88.             sprintf(sql, ERASE_EXCHANGE, ename.c_str());
  89.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  90.         }
  91.         bool removeMessageQueueBinding(const std::string& qname)//删除消息队列的绑定数据
  92.         {
  93.             #define ERASE_MESSAGEQUEUE "delete from  bindingTable where messageQueueName='%s';"
  94.             char sql[256] = {0};
  95.             sprintf(sql, ERASE_MESSAGEQUEUE, qname.c_str());
  96.             return _sqliteHelper.exec(sql, nullptr, nullptr);
  97.         }
  98.         BindingMap recovery()//恢复数据
  99.         {
  100.             #define SELECT_ALL "select exchangeName,messageQueueName,binding_key,durable from  bindingTable"
  101.             BindingMap result;
  102.             bool ret = _sqliteHelper.exec(SELECT_ALL, sqliteCallback, &result);
  103.             if (ret == false)
  104.             {
  105.                 ELOG("恢复绑定数据失败");
  106.                 abort();
  107.             }
  108.             return result;
  109.         }
  110.         ~BindingMapper()
  111.         {
  112.             _sqliteHelper.close();
  113.         }
  114.     private:
  115.         static int sqliteCallback(void *arg, int col, char **row, char **fields)
  116.         {
  117.             BindingMap *result = (BindingMap *)arg;
  118.             auto bdp = std::make_shared<Binding>(row[0],row[1],row[2],row[3]);
  119.             //ename->(qname->binding)
  120.             //为了防止 交换机相关的绑定信息已经存在，因此不能直接创建队列映射，进行添加，这样会覆盖历史数据
  121.             //因此得先获取交换机对应的映射对象，往里边添加数据
  122.             //但是，若这时候没有交换机对应的映射信息，因此这里的获取要使用引用(会保证不存在则自动创建)
  123.             //1.用ename拿出MessageQueueMap
  124.             MessageQueueMap& mqm=(*result)[row[0]];//已经有了就拿出来，没有就创建
  125.             mqm.insert(std::make_pair(row[1],bdp));
  126.             return 0;
  127.         }
  128.     private:
  129.         SqliteHelper _sqliteHelper;
  130.     };
  132.     //3.定义绑定内存管理类
  133.     class BindingManager
  134.     {
  135.     public:
  136.         using ptr=std::shared_ptr<BindingManager>;
  137.         BindingManager(const std::string& dbfile)
  138.             :_mapper(dbfile)
  139.         {
  140.             _bindings=_mapper.recovery();
  141.         }
  142.         bool bind(const std::string& ename,const std::string& qname,const std::string& bdkey,bool durable)//建立绑定关系
  143.         {
  144.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  145.             //先确定改绑定是否已存在
  146.             auto it=_bindings.find(ename);
  147.             if(it!=_bindings.end()&&it->second.find(qname)!=it->second.end())
  148.             {
  149.                 return true;
  150.             }
  151.             Binding::ptr bdp=std::make_shared<Binding>(ename,qname,bdkey,durable);
  152.             //绑定信息是否需要持久化，取决于什么？ 交换机数据是持久化的，以及队列数据也是持久化的。
  153.             if(durable)//需要持久化
  154.             {
  155.                 bool ret=_mapper.insert(bdp);
  156.                 if(ret==false) return false;
  157.             }
  158.             auto& mqm=_bindings[ename];
  159.             mqm.insert(std::make_pair(qname,bdp));
  160.             return true;
  161.         }
  162.         bool unbind(const std::string& ename,const std::string& qname)//解除绑定关系
  163.         {
  164.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  165.             //先确定绑定是否不存在
  166.             auto it=_bindings.find(ename);
  167.             if(it==_bindings.end())//通过ename找不到MessageQueueMap
  168.             {
  169.                 return true;
  170.             }
  171.             auto it2=it->second.find(qname);
  172.             if(it2==it->second.end())//通过qname找不到binding
  173.             {
  174.                 return true;
  175.             }
  176.             //现在确定了要解除的绑定是存在的
  177.             if(it2->second->durable==true)//从数据库里删除
  178.             {
  179.                 bool ret=_mapper.erase(ename,qname);
  180.                 if(ret==false) return false;
  181.             }
  182.             //从内存中删除
  183.             it->second.erase(qname);
  184.             if(it->second.empty())
  185.             {
  186.                 _bindings.erase(ename);
  187.             }
  188.             return true;
  189.         }
  190.         bool deleteExchangeBinding(const std::string& ename)//删除与交换机相关的绑定
  191.         {
  192.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  193.             _bindings.erase(ename);
  194.             return _mapper.removeExchangeBinding(ename);
  195.         }
  196.         bool deleteMessageQueueBinding(const std::string& qname)//删除与消息队列相关的绑定
  197.         {
  198.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  199.             for(auto it=_bindings.begin();it!=_bindings.end();)
  200.             {
  201.                 it->second.erase(qname);
  202.                 if(it->second.empty())
  203.                 {
  204.                     it=_bindings.erase(it);
  205.                 }
  206.                 else
  207.                 {
  208.                     ++it;
  209.                 }
  210.             }
  211.             return _mapper.removeMessageQueueBinding(qname);
  212.         }
  213.         MessageQueueMap getExchangeBinding(const std::string& ename)//获取交换机绑定的所有消息队列
  214.         {
  215.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  216.             //该交换机是否存在
  217.             auto it=_bindings.find(ename);
  218.             if(it==_bindings.end())//该交换机不存在
  219.             {
  220.                 return MessageQueueMap();
  221.             }
  222.             return it->second;
  223.         }
  224.         Binding::ptr getBinding(const std::string& ename,const std::string& qname)
  225.         {
  226.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  227.             //先确定绑定是否不存在
  228.             auto it=_bindings.find(ename);
  229.             if(it==_bindings.end())//通过ename找不到MessageQueueMap
  230.             {
  231.                 return Binding::ptr();
  232.             }
  233.             auto it2=it->second.find(qname);
  234.             if(it2==it->second.end())//通过qname找不到binding
  235.             {
  236.                 return Binding::ptr();
  237.             }
  238.             //现在确定了要解除的绑定是存在的
  239.             return it2->second;
  240.         }
  241.         bool exists(const std::string& ename,const std::string& qname)//绑定信息是否存在
  242.         {
  243.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  244.             auto it=_bindings.find(ename);
  245.             if(it!=_bindings.end()&&it->second.find(qname)!=it->second.end())
  246.             {
  247.                 return true;
  248.             }
  249.             return false;
  250.         }
  251.         size_t size()//绑定数量
  252.         {
  253.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  254.             size_t ret=0;
  255.             for(auto& it:_bindings)
  256.             {
  257.                 ret+=it.second.size();
  258.             }
  259.             return ret;
  260.         }
  261.         void clear()//清空数据
  262.         {
  263.             ILOG("清空数据");
  264.             std::unique_lock<std::mutex> lock(_mutex);//先加锁
  265.             _mapper.removeTable();
  266.             _bindings.clear();
  267.         }
  268.     private:
  269.         std::mutex _mutex;
  270.         BindingMapper _mapper;
  271.         BindingMap _bindings;
  272.     };
  273. };
复制代码
      
                       7、队列消息管理

     文件数据管理

     消息的要素(决定了消息的数据结构)：
               消息的持久化管理：
               向外提供的操作：
               要管理的数据：
          内存数据管理

     以队列为单元进行管理：
                         实现一个对外的总体消息管理类

                                                       登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/helper.hpp"
  3. #include "../mqcomm/logger.hpp"
  4. #include "../mqcomm/message.pb.h"
  5. #include <unordered_map>
  6. #include <memory>
  7. #include <cassert>
  8. #include <list>
  9. #include <mutex>
  11. namespace lsxmq
  12. {
  13.     using MessagePtr = std::shared_ptr<lsxmq::Message>;
  14. #define DATAFILE_SUBFIX ".mqd"
  15. #define TMPFILE_SUBFIX ".mqd.tmp"
  16. #define NOTVALID "0"
  17. #define ISVALID "1"
  18.     class MessageMapper
  19.     {
  20.     public:
  21.         MessageMapper(std::string &basedir, const std::string &qname)
  22.             : _qname(qname)
  23.         {
  24.             if (basedir.back() != '/')
  25.             {
  26.                 basedir.push_back('/'); // 如果没有就往末尾加一个/
  27.             }
  28.             _datafile = basedir + qname + DATAFILE_SUBFIX;
  29.             _tmpfile = basedir + qname + TMPFILE_SUBFIX;
  30.             assert(FileHelper::createDirectory(basedir)); // 创建基本目录
  31.             createMessageFile();
  32.         }
  33.         bool createMessageFile() // 创建消息文件
  34.         {
  35.             if (FileHelper(_datafile).exists() == true)
  36.             {
  37.                 return true;
  38.             }
  39.             if (FileHelper::createFile(_datafile) == false)
  40.             {
  41.                 ELOG("创建消息文件失败");
  42.                 return false;
  43.             }
  44.             return true;
  45.         }
  46.         bool removeMessageFile() // 删除消息文件
  47.         {
  48.             if (FileHelper(_datafile).exists() == false)
  49.             {
  50.                 return true;
  51.             }
  52.             if (FileHelper::removeFile(_datafile) == false)
  53.             {
  54.                 ELOG("删除消息文件失败");
  55.                 return false;
  56.             }
  57.             return true;
  58.         }
  59.         bool insert(MessagePtr &ptr) // 新增消息
  60.         {
  61.             return insert(_datafile, ptr); // 向data文件写入内容
  62.         }
  63.         bool erase(MessagePtr &ptr) // 删除消息
  64.         {
  65.             // 1.将message的有效标志置为0
  66.             ptr->mutable_payload()->set_valid(NOTVALID);
  67.             // 2.进行序列化
  68.             std::string body = ptr->payload().SerializeAsString();
  69.             if (body.size() != ptr->len())
  70.             {
  71.                 ELOG("新数据与原数据长度不一致")
  72.                 return false;
  73.             }
  74.             // 3.将序列化的message覆盖文件中原本的message
  75.             FileHelper dataFileHelper(_datafile);
  76.             if (dataFileHelper.write(body.c_str(), ptr->offset(), body.size()) == false)
  77.             {
  78.                 ELOG("消息重新写入文件失败");
  79.                 return false;
  80.             }
  81.             return true;
  82.         }
  83.         std::list<MessagePtr> gc() // 垃圾回收
  84.         {
  85.             bool ret = true;
  86.             std::list<MessagePtr> result;
  87.             // 1.加载数据
  88.             ret = load(result);
  89.             if (ret == false)
  90.             {
  91.                 ELOG("加载数据失败");
  92.                 return result;
  93.             }
  94.             // 2.将数据写入到临时文件
  95.             FileHelper::createFile(_tmpfile);
  96.             FileHelper tmpDataHelper(_tmpfile);
  97.             for (auto &mp : result)
  98.             {
  99.                 // 写入
  100.                 ret = this->insert(_tmpfile, mp);
  101.                 if (ret == false)
  102.                 {
  103.                     ELOG("消息写入临时文件失败");
  104.                     return result;
  105.                 }
  106.             }
  107.             // 3.删除原文件
  108.             ret = removeMessageFile();
  109.             if (ret == false)
  110.             {
  111.                 ELOG("删除消息文件失败");
  112.                 return result;
  113.             }
  114.             // 4.将原文件改为消息文件
  115.             ret = tmpDataHelper.rename(_datafile);
  116.             if (ret == false)
  117.             {
  118.                 ELOG("修改临时文件名称失败");
  119.                 return result;
  120.             }
  121.             return result;
  122.         }
  124.     private:
  125.         bool load(std::list<MessagePtr> &mplist) // 将文件中的消息加载进内存
  126.         {
  127.             size_t offset = 0;
  128.             FileHelper dataFileHelper(_datafile);
  129.             size_t fsize = dataFileHelper.size();
  130.             bool ret = true;
  131.             while (offset < fsize)
  132.             {
  133.                 // 1.前四位为数据长度
  134.                 size_t msize = 0;
  135.                 ret = dataFileHelper.read((char *)&msize, offset, sizeof(size_t));
  136.                 if (ret == false)
  137.                 {
  138.                     ELOG("读取信息长度失败");
  139.                     return false;
  140.                 }
  141.                 // 2.开始读取数据
  142.                 offset += sizeof(size_t);
  143.                 std::string body(msize,'\0');
  144.                 ret = dataFileHelper.read(&body[0], offset, msize);
  145.                 if (ret == false)
  146.                 {
  147.                     ELOG("读取信息内容失败");
  148.                     return false;
  149.                 }
  150.                 // 3.反序列化
  151.                 MessagePtr mp = std::make_shared<Message>();
  152.                 mp->mutable_payload()->ParseFromString(body);
  153.                 // 4.剔除无效信息
  154.                 if (mp->payload().valid() == NOTVALID)
  155.                 {
  156.                     offset += msize;
  157.                     continue;
  158.                 }
  159.                 // 4.保存结果
  160.                 mplist.push_back(mp);
  161.                 offset += msize;
  162.             }
  163.             return true;
  164.         }
  165.         bool insert(const std::string filename, MessagePtr &ptr)
  166.         {
  167.             // 1.对消息进行序列化
  168.             std::string body = ptr->payload().SerializeAsString();
  169.             MessagePtr tmp=std::make_shared<Message>();
  170.             tmp->mutable_payload()->ParseFromString(body);
  171.             // 2.获取文件大小
  172.             FileHelper FileHelper(filename);
  173.             size_t fsize = FileHelper.size();
  174.             size_t msize = body.size();
  175.             // 3.先写入8字节长度
  176.             if (FileHelper.write((char*)&msize, fsize, sizeof(size_t)) == false)
  177.             {
  178.                 ELOG("消息长度写入文件失败");
  179.                 return false;
  180.             }
  181.             //4.在写入指定长度内容
  182.             if (FileHelper.write(body.c_str(), fsize+sizeof(size_t), msize) == false)
  183.             {
  184.                 ELOG("消息内容写入文件失败");
  185.                 return false;
  186.             }
  187.             // 4.更新message的实际存储位置
  188.             ptr->set_offset(fsize+sizeof(size_t));
  189.             ptr->set_len(msize);
  190.             return true;
  191.         }
  192.     private:
  193.         std::string _qname;
  194.         std::string _datafile;
  195.         std::string _tmpfile;
  196.     };
  198.     class QueueMessage
  199.     {
  200.     public:
  201.         using ptr = std::shared_ptr<QueueMessage>;
  202.         QueueMessage(std::string &basedir, const std::string &qname)
  203.             : _mapper(basedir, qname), _validCount(0), _totalCount(0), _qname(qname)
  204.         {}
  205.         void recovery() // 恢复历史消息
  206.         {
  207.             std::unique_lock<std::mutex> lock(_mutex);
  208.             _pushMessages = _mapper.gc();
  209.             for (auto &mp : _pushMessages)
  210.             {
  211.                 _durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
  212.             }
  213.             _totalCount = _validCount = _durableMessages.size();
  214.         }
  215.         bool insert(BasicProperties *bp, const std::string& body, DeliveryMode deliveryMode) // 新增消息
  216.         {
  217.             // 1.构造message对象
  218.             MessagePtr mp = std::make_shared<Message>();
  219.             if (bp != nullptr)
  220.             {
  221.                 mp->mutable_payload()->mutable_properties()->set_id(bp->id());
  222.                 mp->mutable_payload()->mutable_properties()->set_deliverymode(bp->deliverymode());
  223.                 mp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
  224.                 mp->mutable_payload()->set_body(body);
  225.                 mp->mutable_payload()->set_valid(ISVALID);
  226.             }
  227.             else
  228.             {
  229.                 mp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
  230.                 mp->mutable_payload()->mutable_properties()->set_deliverymode(deliveryMode);
  231.                 mp->mutable_payload()->mutable_properties()->set_routing_key("");
  232.                 mp->mutable_payload()->set_body(body);
  233.                 mp->mutable_payload()->set_valid(ISVALID);
  234.             }
  235.             // 2.判断是否需要持久化
  236.             std::unique_lock<std::mutex> lock(_mutex);
  237.             bool ret = true;
  238.             if (mp->payload().properties().deliverymode() == DeliveryMode::DURABLE) // 需要
  239.             {
  240.                 ret = _mapper.insert(mp);
  241.                 if (ret == false)
  242.                 {
  243.                     ELOG("%s:消息持久化失败", body.c_str());
  244.                     return false;
  245.                 }
  246.                 _durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
  247.                 ++_validCount;
  248.                 ++_totalCount;
  249.             }
  250.             // 3.在内存中管理
  251.             _pushMessages.push_back(mp);
  252.             return true;
  253.         }
  254.         MessagePtr front() // 获取队首消息
  255.         {
  256.             std::unique_lock<std::mutex> lock(_mutex);
  257.             if(_pushMessages.size()==0)//没有消息能拿出来
  258.             {
  259.                 return MessagePtr();
  260.             }
  261.             // 获取队首消息
  262.             MessagePtr mp = _pushMessages.front();
  263.             // 将消息从待推送队列加到待确认hashmap中
  264.             _pushMessages.pop_front();
  265.             _waitAckMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
  266.             return mp;
  267.         }
  268.         bool erase(const std::string &id) // 确认消息,每次删除一条消息后确认是否要进行垃圾回收
  269.         {
  270.             std::unique_lock<std::mutex> lock(_mutex);
  271.             // 1.从待确认消息中找到这条消息
  272.             auto it = _waitAckMessages.find(id);
  273.             if (it == _waitAckMessages.end()) // 找不到
  274.             {
  275.                 return true;
  276.             }
  277.             // 2.根据这条消息的投递模式决定是否从持久化中删除
  278.             bool ret = true;
  279.             if (it->second->payload().properties().deliverymode() == true) // 为持久化
  280.             {
  281.                 // 3.从持久化中删除
  282.                 ret = _mapper.erase(it->second);
  283.                 if (ret == false)
  284.                 {
  285.                     ELOG("从文件中删除id=%s的消息失败", id.c_str());
  286.                     return false;
  287.                 }
  288.                 _durableMessages.erase(id);
  289.                 --_validCount;
  290.                 // 进行垃圾回收
  291.                 gc();
  292.             }
  293.             // 4.从内存中删除
  294.             _waitAckMessages.erase(id);
  295.             return true;
  296.         }
  297.         size_t pushCount()
  298.         {
  299.             std::unique_lock<std::mutex> lock(_mutex);
  300.             return _pushMessages.size();
  301.         }
  302.         size_t durableCount()
  303.         {
  304.             std::unique_lock<std::mutex> lock(_mutex);
  305.             return _durableMessages.size();
  306.         }
  307.         size_t waitAckCount()
  308.         {
  309.             std::unique_lock<std::mutex> lock(_mutex);
  310.             return _waitAckMessages.size();
  311.         }
  312.         size_t validCount()
  313.         {
  314.             std::unique_lock<std::mutex> lock(_mutex);
  315.             return _validCount;
  316.         }
  317.         size_t totalCount()
  318.         {
  319.             std::unique_lock<std::mutex> lock(_mutex);
  320.             return _totalCount;
  321.         }
  322.         void clear()
  323.         {
  324.             _mapper.removeMessageFile();
  325.             _pushMessages.clear();
  326.             _durableMessages.clear();
  327.             _waitAckMessages.clear();
  328.             _totalCount = _validCount = 0;
  329.         }
  331.     private:
  332.         bool NeedGC() // 判断是否需要进行垃圾回收
  333.         {
  334.             if (_totalCount > 2000 && _validCount * 10 / _totalCount < 5)
  335.             {
  336.                 return true;
  337.             }
  338.             return false;
  339.         }
  340.         void gc() // 垃圾回收
  341.         {
  342.             if (NeedGC() == false)
  343.                 return;
  344.             // 1.接收垃圾回收后的消息
  345.             std::list<MessagePtr> mplist = _mapper.gc();
  346.             // 2.修改内存中消息的数据
  347.             for (auto &mp : mplist)
  348.             {
  349.                 // 从持久化消息中找到消息
  350.                 auto it = _durableMessages.find(mp->payload().properties().id());
  351.                 if (it == _durableMessages.end())
  352.                 {
  353.                     ILOG("出现了一条持久消息没有在内存中管理");
  354.                     // 解决策略：重新添加
  355.                     _pushMessages.push_back(mp);
  356.                     _durableMessages.insert(std::make_pair(mp->payload().properties().id(), mp));
  357.                     continue;
  358.                 }
  359.                 // 更新消息与文件相关的数据
  360.                 it->second->set_offset(mp->offset());
  361.                 it->second->set_len(mp->len());
  362.             }
  363.             _totalCount = _validCount = _durableMessages.size();
  364.         }
  366.     private:
  367.         std::mutex _mutex;
  368.         MessageMapper _mapper;
  369.         std::list<MessagePtr> _pushMessages;                          // 待推送消息
  370.         std::unordered_map<std::string, MessagePtr> _durableMessages; // 持久化消息
  371.         std::unordered_map<std::string, MessagePtr> _waitAckMessages; // 待确认消息
  372.         size_t _validCount;                                           // 持久化有效消息个数
  373.         size_t _totalCount;                                           // 持久化总消息个数
  374.         std::string _qname;
  375.     };
  377.     class QueueMessageManager
  378.     {
  379.     public:
  380.         using ptr=std::shared_ptr<QueueMessageManager>;
  381.         QueueMessageManager(const std::string &basedir)
  382.             : _basedir(basedir)
  383.         {}
  384.         void clear()
  385.         {
  386.             std::unique_lock<std::mutex> lock(_mutex);
  387.             for(auto& a:_queueMessages)
  388.             {
  389.                 a.second->clear();
  390.             }
  391.         }
  392.         void initQueueMessage(const std::string &qname) // 初始化队列
  393.         {
  394.             QueueMessage::ptr qmp;
  395.             {
  396.                 std::unique_lock<std::mutex> lock(_mutex);
  397.                 // 该队列是否已存在
  398.                 auto it = _queueMessages.find(qname);
  399.                 if (it != _queueMessages.end()) // 已存在
  400.                 {
  401.                     return ;
  402.                 }
  403.                 qmp = std::make_shared<QueueMessage>(_basedir, qname);
  404.                 _queueMessages.insert(std::make_pair(qname,qmp));
  405.             }
  406.             qmp->recovery();
  407.         }
  408.         void destroyQueueMessage(const std::string &qname)// 销毁队列
  409.         {
  410.             QueueMessage::ptr qmp;
  411.             {
  412.                 std::unique_lock<std::mutex> lock(_mutex);
  413.                 // 该队列是否不存在
  414.                 auto it = _queueMessages.find(qname);
  415.                 if (it == _queueMessages.end()) // 不存在
  416.                 {
  417.                     return ;
  418.                 }
  419.                 qmp=it->second;
  420.                 _queueMessages.erase(qname);
  421.             }
  422.             qmp->clear();
  423.         }
  424.         bool insert(const std::string &qname, BasicProperties *bp, const std::string& body, DeliveryMode deliveryMode) // 向队列新增消息
  425.         {
  426.             QueueMessage::ptr qmp;
  427.             {
  428.                 std::unique_lock<std::mutex> lock(_mutex);
  429.                 // 该队列是否不存在
  430.                 auto it = _queueMessages.find(qname);
  431.                 if (it == _queueMessages.end()) // 不存在
  432.                 {
  433.                     ELOG("向队列%s新增消息失败，因为该队列不存在",qname.c_str());
  434.                     return false;
  435.                 }
  436.                 qmp=it->second;
  437.             }
  438.             qmp->insert(bp,body,deliveryMode);
  439.             return true;
  440.         }
  441.         MessagePtr front(const std::string &qname)// 获取队首消息
  442.         {
  443.             QueueMessage::ptr qmp;
  444.             {
  445.                 std::unique_lock<std::mutex> lock(_mutex);
  446.                 // 该队列是否不存在
  447.                 auto it = _queueMessages.find(qname);
  448.                 if (it == _queueMessages.end()) // 不存在
  449.                 {
  450.                     ELOG("从队列%s获取队首消息失败，因为该队列不存在",qname.c_str());
  451.                     return MessagePtr();
  452.                 }
  453.                 qmp=it->second;
  454.             }
  455.             return qmp->front();
  456.         }
  457.         bool ack(const std::string &qname, const std::string &id)// 确认消息
  458.         {
  459.             QueueMessage::ptr qmp;
  460.             {
  461.                 std::unique_lock<std::mutex> lock(_mutex);
  462.                 // 该队列是否不存在
  463.                 auto it = _queueMessages.find(qname);
  464.                 if (it == _queueMessages.end()) // 不存在
  465.                 {
  466.                     ELOG("向队列%s确认消息失败，因为该队列不存在",qname.c_str());
  467.                     return false;
  468.                 }
  469.                 qmp=it->second;
  470.             }
  471.             qmp->erase(id);
  472.             return true;
  473.         }
  474.         size_t pushCount(const std::string& qname)
  475.         {
  476.             QueueMessage::ptr qmp;
  477.             {
  478.                 std::unique_lock<std::mutex> lock(_mutex);
  479.                 // 该队列是否不存在
  480.                 auto it = _queueMessages.find(qname);
  481.                 if (it == _queueMessages.end()) // 不存在
  482.                 {
  483.                     ELOG("向队列%s获取待发送消息数量失败，因为该队列不存在",qname.c_str());
  484.                     return 0;
  485.                 }
  486.                 qmp=it->second;
  487.             }
  488.             return qmp->pushCount();
  489.         }
  490.         size_t durableCount(const std::string& qname)
  491.         {
  492.             QueueMessage::ptr qmp;
  493.             {
  494.                 std::unique_lock<std::mutex> lock(_mutex);
  495.                 // 该队列是否不存在
  496.                 auto it = _queueMessages.find(qname);
  497.                 if (it == _queueMessages.end()) // 不存在
  498.                 {
  499.                     ELOG("向队列%s获取持久化消息数量失败，因为该队列不存在",qname.c_str());
  500.                     return 0;
  501.                 }
  502.                 qmp=it->second;
  503.             }
  504.             return qmp->durableCount();
  505.         }
  506.         size_t waitAckCount(const std::string& qname)
  507.         {
  508.             QueueMessage::ptr qmp;
  509.             {
  510.                 std::unique_lock<std::mutex> lock(_mutex);
  511.                 // 该队列是否不存在
  512.                 auto it = _queueMessages.find(qname);
  513.                 if (it == _queueMessages.end()) // 不存在
  514.                 {
  515.                     ELOG("向队列%s获取等待确认消息数量失败，因为该队列不存在",qname.c_str());
  516.                     return 0;
  517.                 }
  518.                 qmp=it->second;
  519.             }
  520.             return qmp->waitAckCount();
  521.         }
  522.         size_t validCount(const std::string& qname)
  523.         {
  524.             QueueMessage::ptr qmp;
  525.             {
  526.                 std::unique_lock<std::mutex> lock(_mutex);
  527.                 // 该队列是否不存在
  528.                 auto it = _queueMessages.find(qname);
  529.                 if (it == _queueMessages.end()) // 不存在
  530.                 {
  531.                     ELOG("向队列%s获取有效持久化消息数量失败，因为该队列不存在",qname.c_str());
  532.                     return 0;
  533.                 }
  534.                 qmp=it->second;
  535.             }
  536.             return qmp->validCount();
  537.         }
  538.         size_t totalCount(const std::string& qname)
  539.         {
  540.             QueueMessage::ptr qmp;
  541.             {
  542.                 std::unique_lock<std::mutex> lock(_mutex);
  543.                 // 该队列是否不存在
  544.                 auto it = _queueMessages.find(qname);
  545.                 if (it == _queueMessages.end()) // 不存在
  546.                 {
  547.                     ELOG("向队列%s获取全部持久化消息数量失败，因为该队列不存在",qname.c_str());
  548.                     return 0;
  549.                 }
  550.                 qmp=it->second;
  551.             }
  552.             return qmp->totalCount();
  553.         }
  554.     private:
  555.         std::mutex _mutex;
  556.         std::unordered_map<std::string, QueueMessage::ptr> _queueMessages;
  557.         std::string _basedir;
  558.     };
  559. };
复制代码
      
                       8、假造机管理

     假造机模块是对上述三个数据管理模块的整合，并基于数据之间的关联关系进⾏团结操作。
     界说假造机类包含以下成员：
     
      假造机包含操作：
     
     假造机管理操作：可以有多台假造机，项目中并没有实现。
     
                                   登录后复制                        
  1. #pragma once
  2. #include "binding.hpp"
  3. #include "exchange.hpp"
  4. #include "message.hpp"
  5. #include "messageQueue.hpp"
  7. namespace lsxmq
  8. {
  9.     class VirtualHost//虚拟机
  10.     {
  11.     public:
  12.         using ptr=std::shared_ptr<VirtualHost>;
  13.         VirtualHost(const std::string& hname,const std::string& dbfile,const std::string& basedir)
  14.             :_hostName(hname)
  15.             ,_bmp(std::make_shared<BindingManager>(dbfile))
  16.             ,_emp(std::make_shared<ExchangeManager>(dbfile))
  17.             ,_qmmp(std::make_shared<QueueMessageManager>(basedir))
  18.             ,_mqmp(std::make_shared<MessageQueueManager>(dbfile))
  19.         {
  20.             //恢复队列消息
  21.             //1.获取所有队列
  22.             MessageQueueMap mqmap=_mqmp->getAll();
  23.             //2.用队列名称初始化队列消息
  24.             for(auto& it:mqmap)
  25.             {
  26.                 _qmmp->initQueueMessage(it.second->name);
  27.             }
  28.         }
  29.         bool declareExchange(const std::string& name,ExchangeType type
  30.                 ,bool durable,bool autoDel
  31.                 ,std::unordered_map<std::string,std::string> args)//声明交换机
  32.         {
  33.             return _emp->declareExchange(name,type,durable,autoDel,args);
  34.         }
  35.         bool deleteExchange(const std::string& ename)//删除交换机
  36.         {
  37.             bool ret=_emp->deleteExchange(ename);
  38.             if(ret==false)
  39.             {
  40.                 return false;
  41.             }
  42.             //删除交换机对应的绑定
  43.             return _bmp->deleteExchangeBinding(ename);
  44.         }
  45.         bool existsExchange(const std::string& ename)
  46.         {
  47.             return _emp->exists(ename);
  48.         }
  50.         bool declareMessageQueue(const std::string &name, bool durable, bool exclusive
  51.                 , bool autoDel, std::unordered_map<std::string, std::string> args)//声明消息队列
  52.         {
  53.             bool ret=_mqmp->declareMessageQueue(name,durable,exclusive,autoDel,args);
  54.             if(ret==false)
  55.             {
  56.                 return false;
  57.             }
  58.             _qmmp->initQueueMessage(name);
  59.             return true;
  60.         }
  61.         bool deleteMessageQueue(const std::string &qname)//删除消息队列
  62.         {
  63.             auto ret=_mqmp->deleteMessageQueue(qname);
  64.             if(ret==false)
  65.             {
  66.                 return false;
  67.             }
  68.             //删除消息队列对应的绑定
  69.             ret=_bmp->deleteMessageQueueBinding(qname);
  70.             if(ret==false)
  71.             {
  72.                 return false;
  73.             }
  74.             //删除对应的队列消息
  75.             _qmmp->destroyQueueMessage(qname);
  76.             return true;
  77.         }
  78.         bool existsMessageQueue(const std::string& qname)
  79.         {
  80.             return _mqmp->exists(qname);
  81.         }
  83.         bool bind(const std::string& ename,const std::string& qname,const std::string& binding_key)//绑定
  84.         {
  85.             //交换机是否存在
  86.             Exchange::ptr ep=_emp->selectExchange(ename);
  87.             if(ep.get()==nullptr)
  88.             {
  89.                 ELOG("%s与%s绑定失败，因为交换机不存在",ename.c_str(),qname.c_str());
  90.                 return false;
  91.             }
  92.             //队列是否存在
  93.             MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
  94.             if(mqp.get()==nullptr)
  95.             {
  96.                 ELOG("%s与%s绑定失败，因为队列不存在",ename.c_str(),qname.c_str());
  97.                 return false;
  98.             }
  99.             //根据交换机与队列的持久化方式决定绑定是否持久化
  100.             return _bmp->bind(ename,qname,binding_key,ep->durable&&mqp->durable);
  101.         }
  102.         bool unbind(const std::string& ename,const std::string& qname)//解绑
  103.         {
  104.              //交换机是否存在
  105.             Exchange::ptr ep=_emp->selectExchange(ename);
  106.             if(ep.get()==nullptr)
  107.             {
  108.                 ELOG("%s与%s解绑失败，因为交换机不存在",ename.c_str(),qname.c_str());
  109.                 return false;
  110.             }
  111.             //队列是否存在
  112.             MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
  113.             if(mqp.get()==nullptr)
  114.             {
  115.                 ELOG("%s与%s解绑失败，因为队列不存在",ename.c_str(),qname.c_str());
  116.                 return false;
  117.             }
  118.             return _bmp->unbind(ename,qname);
  119.         }
  120.         MessageQueueBindingMap exchangeBindings(const std::string& ename)//获取一台交换机的所有绑定
  121.         {
  122.             return _bmp->getExchangeBinding(ename);
  123.         }
  124.         bool existsBinding(const std::string &ename, const std::string &qname)
  125.         {
  126.             return _bmp->exists(ename,qname);
  127.         }
  129.         bool basicPublish(const std::string &qname, BasicProperties *bp
  130.                 , const std::string& body)//发布一条消息
  131.         {
  132.             MessageQueue::ptr mqp=_mqmp->selectMessageQueue(qname);
  133.             return _qmmp->insert(qname,bp,body,mqp->durable);
  134.         }
  135.         MessagePtr basicConsume(const std::string& qname)//获取一条消息
  136.         {
  137.             return _qmmp->front(qname);
  138.         }
  139.         bool basicAck(const std::string &qname, const std::string &id)//确认一条消息
  140.         {
  141.             return _qmmp->ack(qname,id);
  142.         }
  143.         
  144.         void clear()
  145.         {
  146.             _bmp->clear();
  147.             _emp->clear();
  148.             _qmmp->clear();
  149.             _mqmp->clear();
  150.         }
  151.     private:
  152.         std::string _hostName;//虚拟机名称
  153.         BindingManager::ptr _bmp;//绑定管理句柄
  154.         ExchangeManager::ptr _emp;//交换机管理句柄
  155.         QueueMessageManager::ptr _qmmp;//队列消息管理句柄
  156.         MessageQueueManager::ptr _mqmp;//消息队列管理句柄
  157.     };
  158. };
复制代码
      
                       9、交换机路由模块

     交换路由规则取决要素：
               路由交换模块：
               binding_key：
     是由数字字⺟下划线构成的, 并且使⽤ . 分成若⼲部门，并⽀持 * 和 # 通配符。
     例如： news.music.# ，这⽤于表⽰交换机绑定的当前队列是⼀个⽤于发布⾳乐消息的队列。
     
     
     
     
     routing_key：
     是由数据、字⺟和下划线构成， 并且可以使⽤ . 分别成若⼲部门。
     例如： news.music.pop ，这⽤于表⽰当前发布的消息是⼀个流⾏⾳乐的消息。
     ⽐如，在进⾏队列绑定时，某队列的binding_key约定为：news.music.#表⽰这个队列⽤于发布⾳乐消息。⽽这时候客⼾端发布了⼀条消息，其中routing_key为：news.music.pop 则可以匹配成功，⽽，如果发布消息的routing_key为：news.sport.football，这时候就会匹配失败。
     匹配算法头脑：
                                             登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/helper.hpp"
  3. #include "../mqcomm/message.pb.h"
  5. namespace lsxmq
  6. {
  7.     class Router
  8.     {
  9.     public:
  10.         static bool routingKeyIsLegal(const std::string& routingKey)//判断routing_key是否合法
  11.         {
  12.             //只需要确定是否有非法字符即可
  13.             //合法字符：a~z、A~Z、.、_
  14.             for(auto& ch:routingKey)
  15.             {
  16.                 if((ch>='a'&&ch<='z')
  17.                     ||(ch>='A'&&ch<='Z')
  18.                     ||(ch>='0'&&ch<='9')
  19.                     ||ch=='_'||ch=='.')
  20.                 {
  21.                     continue;
  22.                 }
  23.                 else
  24.                 {
  25.                     return false;//不合法
  26.                 }
  27.             }
  28.             return true;
  29.         }
  30.         static bool bindingKeyIsLegal(const std::string& bindingKey)
  31.         {
  32.             //1.确定是否有非法字符即可
  33.             //合法字符：a~z、A~Z、.、_、*、#
  34.             for(auto& ch:bindingKey)
  35.             {
  36.                 if((ch>='a'&&ch<='z')
  37.                     ||(ch>='A'&&ch<='Z')
  38.                     ||(ch>='0'&&ch<='9')
  39.                     ||ch=='_'||ch=='.'
  40.                     ||ch=='#'||ch=='*')
  41.                 {
  42.                     continue;
  43.                 }
  44.                 else
  45.                 {
  46.                     return false;//不合法
  47.                 }
  48.             }
  49.             //2.#和*必须独立存在(aa.a#.b这种就是非法)
  50.             std::vector<std::string> subStr;
  51.             StrHelper::split(bindingKey,".",subStr);
  52.             for(auto& word:subStr)
  53.             {
  54.                 if(word.size()>1
  55.                     &&word.find('#')!=std::string::npos
  56.                     &&word.find('*')!=std::string::npos)
  57.                 {
  58.                     return false;
  59.                 }
  60.             }
  61.             //3.#和*不连续出现，#的前面没有#和*，*的前面没有#
  62.             for(int i=1;i<subStr.size();++i)
  63.             {
  64.                 if(subStr[i]=="#"&&subStr[i-1]=="#")
  65.                 {
  66.                     return false;
  67.                 }
  68.                 if(subStr[i]=="#"&&subStr[i-1]=="*")
  69.                 {
  70.                     return false;
  71.                 }
  72.                 if(subStr[i]=="*"&&subStr[i-1]=="#")
  73.                 {
  74.                     return false;
  75.                 }
  76.             }
  77.             return true;
  78.         }
  79.         static bool route(const std::string& routingKey,const std::string& bindingKey,ExchangeType routeType)
  80.         {
  81.             if(routeType==DIRECT)//直接路由
  82.             {
  83.                 return routingKey==bindingKey;   
  84.             }
  85.             if(routeType==FANOUT)//广播路由
  86.             {
  87.                 return true;
  88.             }
  89.             if(routeType==TOPIC)//主题路由
  90.             {
  91.                 return topicRoute(routingKey,bindingKey);
  92.             }
  93.             ELOG("位置交换机类型：%d",routeType);
  94.             return false;
  95.         }
  96.     private:
  97.         static bool topicRoute(const std::string& routingKey,const std::string& bindingKey)
  98.         {
  99.             //1.先分割字符串
  100.             std::vector<std::string> subBinding;
  101.             StrHelper::split(bindingKey,".",subBinding);
  102.             std::vector<std::string> subRouting;
  103.             StrHelper::split(routingKey,".",subRouting);
  104.             //2.构造dp表
  105.             int row=subBinding.size(),col=subRouting.size();//行列
  106.             std::vector<std::vector<int>> dp(row+1,std::vector<int>(col+1,0));
  107.             dp[0][0]=1;
  108.             if(subBinding[0]=="#") dp[1][0]=1;//当binding_key以#起始时，需要将#对应行的第0列置为1
  109.             //3.开始匹配
  110.             for(int i=1;i<row+1;++i)
  111.             {
  112.                 for(int j=1;j<col+1;++j)
  113.                 {
  114.                     if(subBinding[i-1]=="#")//遇到#通配符匹配成功时,可以从左上方、左方、上方继承
  115.                     {
  116.                         dp[i][j]=dp[i-1][j-1]||dp[i-1][j]||dp[i][j-1];
  117.                     }
  118.                     else if(subBinding[i-1]=="*"||(subBinding[i-1]==subRouting[j-1]))
  119.                     {
  120.                         //当两个单词匹配成功时,从左上方继承结果
  121.                         dp[i][j]=dp[i-1][j-1];
  122.                     }
  123.                 }
  124.             }
  125.             return dp[row][col];
  126.         }
  127.     };
  128. };
复制代码
      

  • 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轮转
  • 互斥锁：保证线程安全
  • 队列名称
     对消费者进行同一管理
     

  • 初始化/删除队列的消费者信息结构(创建/删除队列的时候初始化)
  • 向指定队列新增消费者(客户端订阅指定队列消息的时候)新增完成时返回消费者对象，让信道能管理
  • 从指定队列移除消费者(客户端取消订阅的时候)
  • 移除指定队列的全部消费者(队列被删除时销毁)删除消费者的队列管理单元对象
  • 从指定队列获取一个消费者(轮询获取-消费者轮询消费起到负载平衡作用)
  • 判定队列消费者是否为空
  • 判定队列中指定消费者是否存在
  • 清理消费者
                                   登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/helper.hpp"
  3. #include "../mqcomm/logger.hpp"
  4. #include "../mqcomm/message.pb.h"
  6. namespace lsxmq
  7. {
  8.     using ConsumerCallBack=std::function<void(const std::string&,BasicProperties*,const std::string&)>;
  9.                                                       //消费者标识     //基本属性             //消息内容
  10.     struct Consumer
  11.     {
  12.         using ptr=std::shared_ptr<Consumer>;
  13.         Consumer(){}
  14.         Consumer(const std::string& ctag,const std::string& cqname,bool cautoAck,const ConsumerCallBack& ccallback)
  15.             :tag(ctag),qname(cqname),autoAck(cautoAck),callback(ccallback)
  16.         {}
  17.         std::string tag;//消费者标识
  18.         std::string qname;//队列名称
  19.         bool autoAck;//自动确认标志
  20.         ConsumerCallBack callback;//消费回调函数
  21.     };
  23.     //以队列为单元管理消费者
  24.     class QueueConsumer
  25.     {
  26.     public:
  27.         using ptr=std::shared_ptr<QueueConsumer>;
  28.         QueueConsumer(const std::string& qname)
  29.             :_qname(qname),_rrSeq(0)
  30.         {}
  31.         Consumer::ptr create(const std::string& tag,bool autoAck,const ConsumerCallBack& callback)//创建一个消费者
  32.         {
  33.             //1.先加锁
  34.             std::unique_lock<std::mutex> lock(_mutex);
  35.             //2.判断该消费者是否已存在
  36.             for(auto& con:_consumers)
  37.             {
  38.                 if(con->tag==tag)
  39.                 {
  40.                     return con;
  41.                 }
  42.             }
  43.             //3.构造消费者
  44.             Consumer::ptr cp=std::make_shared<Consumer>(tag,_qname,autoAck,callback);
  45.             //4.添加管理并返回消费者
  46.             _consumers.push_back(cp);
  47.             return cp;
  48.         }
  49.         void remove(const std::string& tag)//删除一个消费者
  50.         {
  51.             //1.先加锁
  52.             std::unique_lock<std::mutex> lock(_mutex);
  53.             //2.查找消费者并删除
  54.             for(auto it=_consumers.begin();it!=_consumers.end();++it)
  55.             {
  56.                 if((*it)->tag==tag)
  57.                 {
  58.                     _consumers.erase(it);
  59.                     return ;
  60.                 }
  61.             }
  62.         }
  63.         Consumer::ptr choose()//选择一个消费者
  64.         {
  65.             //1.先加锁
  66.             std::unique_lock<std::mutex> lock(_mutex);
  67.             if(_consumers.size()==0)//没有消费者
  68.             {
  69.                 return Consumer::ptr();
  70.             }
  71.             //2.获取选择的消费者序号
  72.             size_t id=_rrSeq%_consumers.size();
  73.             ++_rrSeq;
  74.             //3.返回消费者
  75.             return _consumers[id];
  76.         }
  77.         bool empty()//判断队列中是否没有消费者
  78.         {
  79.             //1.先加锁
  80.             std::unique_lock<std::mutex> lock(_mutex);
  81.             return _consumers.empty();
  82.         }
  83.         bool exists(const std::string& tag)//判断消费者是否存在
  84.         {
  85.             //1.先加锁
  86.             std::unique_lock<std::mutex> lock(_mutex);
  87.             //2.判断消费者是否存在
  88.             for(auto& con:_consumers)
  89.             {
  90.                 if(con->tag==tag)
  91.                 {
  92.                     return true;
  93.                 }
  94.             }
  95.             return false;
  96.         }
  97.         void clear()//清空
  98.         {
  99.             //1.先加锁
  100.             std::unique_lock<std::mutex> lock(_mutex);
  101.             _consumers.clear();
  102.         }
  103.     private:
  104.         std::vector<Consumer::ptr> _consumers;
  105.         size_t _rrSeq;
  106.         std::mutex _mutex;
  107.         std::string _qname;
  108.     };
  110.     class QueueConsumerManager
  111.     {
  112.     public:
  113.         using ptr=std::shared_ptr<QueueConsumerManager>;
  114.         QueueConsumerManager(){}
  115.         void initQueueConsumer(const std::string& qname)//初始化队列
  116.         {
  117.             //1.加锁
  118.             std::unique_lock<std::mutex> lock(_mutex);
  119.             //2.是否已存在
  120.             auto it=_queueConsumers.find(qname);
  121.             if(it!=_queueConsumers.end())
  122.             {
  123.                 return ;
  124.             }
  125.             //2.构造队列
  126.             QueueConsumer::ptr qcp=std::make_shared<QueueConsumer>(qname);
  127.             //3.添加管理
  128.             _queueConsumers.insert(std::make_pair(qname,qcp));
  129.         }
  130.         void destroyQueueConsumer(const std::string& qname)//销毁队列
  131.         {
  132.             //1.加锁
  133.             std::unique_lock<std::mutex> lock(_mutex);
  134.             //2.是否不存在
  135.             auto it=_queueConsumers.find(qname);
  136.             if(it==_queueConsumers.end())
  137.             {
  138.                 return ;
  139.             }
  140.             //3.删除队列消费者
  141.             //it->second->clear();//不需要，智能指针指向的对象会自动释放
  142.             _queueConsumers.erase(qname);
  143.         }
  144.         Consumer::ptr create(const std::string& qname,const std::string& tag
  145.                 ,bool autoAck,const ConsumerCallBack& callback)//往某个队列创建消费者
  146.         {
  147.             QueueConsumer::ptr qcp;
  148.             {
  149.                  //1.加锁
  150.                 std::unique_lock<std::mutex> lock(_mutex);
  151.                 //2.判断队列是否不存在
  152.                 auto it=_queueConsumers.find(qname);
  153.                 if(it==_queueConsumers.end())
  154.                 {
  155.                     ELOG("向队列%s添加消费者%s失败，因为该队列不存在",qname.c_str(),tag.c_str());
  156.                     return Consumer::ptr();
  157.                 }
  158.                 qcp=it->second;
  159.             }
  160.             //3.通过队列消费者句柄创建消费者
  161.             return qcp->create(tag,autoAck,callback);
  162.         }
  163.         void remove(const std::string& qname,const std::string& tag)//从某个队列中删除一个消费者
  164.         {
  165.             QueueConsumer::ptr qcp;
  166.             {
  167.                  //1.加锁
  168.                 std::unique_lock<std::mutex> lock(_mutex);
  169.                 //2.判断队列是否不存在
  170.                 auto it=_queueConsumers.find(qname);
  171.                 if(it==_queueConsumers.end())
  172.                 {
  173.                     ELOG("从队列%s删除消费者%s失败，因为该队列不存在",qname.c_str(),tag.c_str());
  174.                     return ;
  175.                 }
  176.                 qcp=it->second;
  177.             }
  178.             //3.通过队列消费者句柄创建消费者
  179.             qcp->remove(tag);
  180.         }
  181.         Consumer::ptr choose(const std::string& qname)//选择一个消费者
  182.         {
  183.             QueueConsumer::ptr qcp;
  184.             {
  185.                  //1.加锁
  186.                 std::unique_lock<std::mutex> lock(_mutex);
  187.                 //2.判断队列是否不存在
  188.                 auto it=_queueConsumers.find(qname);
  189.                 if(it==_queueConsumers.end())
  190.                 {
  191.                     ELOG("从队列%s获取消费者失败，因为该队列不存在",qname.c_str());
  192.                     return Consumer::ptr();
  193.                 }
  194.                 qcp=it->second;
  195.             }
  196.             //3.通过队列消费者句柄创建消费者
  197.             return qcp->choose();
  198.         }
  199.         bool empty(const std::string& qname)
  200.         {
  201.             QueueConsumer::ptr qcp;
  202.             {
  203.                  //1.加锁
  204.                 std::unique_lock<std::mutex> lock(_mutex);
  205.                 //2.判断队列是否不存在
  206.                 auto it=_queueConsumers.find(qname);
  207.                 if(it==_queueConsumers.end())
  208.                 {
  209.                     ELOG("从队列%s获取是否为空失败，因为该队列不存在",qname.c_str());
  210.                     return false;
  211.                 }
  212.                 qcp=it->second;
  213.             }
  214.             //3.通过队列消费者句柄创建消费者
  215.             return qcp->empty();
  216.             
  217.         }
  218.         bool exists(const std::string& qname,const std::string& tag)
  219.         {
  220.             QueueConsumer::ptr qcp;
  221.             {
  222.                  //1.加锁
  223.                 std::unique_lock<std::mutex> lock(_mutex);
  224.                 //2.判断队列是否不存在
  225.                 auto it=_queueConsumers.find(qname);
  226.                 if(it==_queueConsumers.end())
  227.                 {
  228.                     ELOG("从队列%s获取消费者失败，因为该队列不存在",qname.c_str());
  229.                     return false;
  230.                 }
  231.                 qcp=it->second;
  232.             }
  233.             //3.通过队列消费者句柄创建消费者
  234.             return qcp->exists(tag);
  235.         }
  236.         void clear()
  237.         {
  238.             //1.加锁
  239.             std::unique_lock<std::mutex> lock(_mutex);
  240.             _queueConsumers.clear();
  241.         }
  242.     private:
  243.         std::mutex _mutex;
  244.         std::unordered_map<std::string,QueueConsumer::ptr> _queueConsumers;
  245.     };
  246. };
复制代码
      

  • 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协议的处理句柄：网络通讯前的协议处理
  • 消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
  • 假造机句柄：交换机/队列/绑定/消息数据管理
  • 工作线程池句柄：一条消息被发布到队列后，必要将消息推送给订阅了对应队列的消费者，过程由线程池完成。
     管理操作：
     

  • 提供声明&删除交换机操作(删除交换机的同时删除交换构造联的绑定信息)
  • 提供声明&删除队列操作(删除队列的同时，删除队列关联的绑定信息，消息，消费者信息)
  • 提供绑定&解绑队列操作
  • 提供订阅&取消订阅队列消息操作
  • 提供发布&确认消息操作
     信道管理：
     

  • 信道的增删查
     应用层协议设计：
                                   登录后复制                        
  1. syntax= "proto3";
  2. package lsxmq;
  3. import "message.proto";
  5. //信道的打开与关闭
  6. message OpenChannelRequest  //打开信道
  7. {
  8.     string rid = 1;  //请求id,对应不同的请求类型
  9.     string cid = 2;  //信道id
  10. };
  11. message CloseChannelRequest  //关闭信道
  12. {
  13.     string rid = 1;
  14.     string cid = 2;
  15. };
  16. //交换机的声明与删除
  17. message DeclareExchangeRequest  //声明交换机
  18. {
  19.     string rid = 1;
  20.     string cid = 2;
  21.     string exchange_name = 3;  //交换机名称
  22.     ExchangeType exchange_type = 4;  //交换机类型
  23.     bool durable = 5;  //持久化
  24.     bool auto_delete = 6;  //自动删除
  25.     map<string, string> args = 7;  //其它参数
  26. };
  27. message DeleteExchangeRequest  //删除交换机
  28. {
  29.     string rid = 1;
  30.     string cid = 2;
  31.     string exchange_name = 3;
  32. };
  33. //消息队列的声明与删除
  34. message DeclareMessageQueueRequest  //声明消息队列
  35. {
  36.     string rid = 1;
  37.     string cid = 2;
  38.     string queue_name = 3;  //队列名称
  39.     bool exclusive = 4;  //是否独占
  40.     bool durable = 5;  
  41.     bool auto_delete = 6;
  42.     map<string, string> args = 7;
  43. };
  44. message DeleteMessageQueueRequest  //删除消息队列
  45. {
  46.     string rid = 1;
  47.     string cid = 2;
  48.     string queue_name = 3;
  49. };
  50. //绑定与解绑
  51. message BindRequest  //绑定
  52. {
  53.     string rid = 1;
  54.     string cid = 2;
  55.     string exchange_name = 3;
  56.     string queue_name = 4;
  57.     string binding_key = 5;
  58. };
  59. message UnbindRequest  //解绑
  60. {
  61.     string rid = 1;
  62.     string cid = 2;
  63.     string exchange_name = 3;
  64.     string queue_name = 4;
  65. };
  66. //订阅与取消订阅
  67. message BasicConsumeRequest  //订阅
  68. {
  69.     string rid = 1;
  70.     string cid = 2;
  71.     string consumer_tag  =3;  //消费者标识
  72.     string queue_name = 4;
  73.     bool auto_ack = 5;  //是否自动确认
  74. };
  75. message BasicCancelRequest  //取消订阅
  76. {
  77.     string rid = 1;
  78.     string cid = 2;
  79.     string consumer_tag = 3;
  80.     string queue_name = 4;
  81. };
  82. //发布与确认消息
  83. message BasicPublishRequest //发布消息
  84. {
  85.     string rid = 1;
  86.     string cid = 2;
  87.     string exchange_name = 3;
  88.     string body = 4;  //消息内容
  89.     BasicProperties properties = 5;  //消息基本属性
  90. };
  91. message BasicAckRequest   //确认消息
  92. {
  93.     string rid = 1;
  94.     string cid = 2;
  95.     string queue_name = 3;
  96.     string mid = 4;
  97. };
  98. //消息的推送
  99. message basicConsumeResponse {
  100.     string cid = 1;
  101.     string consumer_tag = 2;
  102.     string body = 3;
  103.     BasicProperties properties = 4;
  104. };
  105. //通用响应
  106. message BasicCommonResponse
  107. {
  108.     string rid = 1;
  109.     string cid = 2;
  110.     bool ok = 3;
  111. }
复制代码
      

  • 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信道的操作
     连接管理：
     

  • 连接的增删查
                                   登录后复制                        
  1. #pragma once
  2. #include "channel.hpp"
  4. namespace lsxmq
  5. {
  6.     class Connection
  7.     {
  8.     public:
  9.         using ptr=std::shared_ptr<Connection>;
  10.         Connection(const muduo::net::TcpConnectionPtr &conn
  11.             , const ProtobufCodecPtr &codec
  12.             , const QueueConsumerManager::ptr &qcmp
  13.             , const VirtualHost::ptr &vhp
  14.             , const ThreadPool::ptr &pool)
  15.             : _conn(conn)
  16.             , _codec(codec)
  17.             , _qcmp(qcmp)
  18.             , _vhp(vhp)
  19.             , _pool(pool)
  20.             , _chmp(std::make_shared<ChannelManager>())
  21.         {
  22.             DLOG("new Connection: %p",this);
  23.         }
  24.         ~Connection()
  25.         {
  26.             DLOG("delete Connection: %p",this);
  27.         }
  28.         void openChannel(const OpenChannelRequestPtr &req)
  29.         {
  30.             // 1.判断信道是否已打开
  31.             bool ret = _chmp->openChannel(req->cid(),_conn,_codec,_qcmp,_vhp,_pool);
  32.             if(ret==false)
  33.             {
  34.                 basicResponse(req->rid(),req->cid(),false);
  35.                 return;
  36.             }
  37.             //2.响应
  38.             basicResponse(req->rid(),req->cid(),true);
  39.         }
  40.         void closeChannel(const CloseChannelRequestPtr &req)
  41.         {
  42.             _chmp->closeChannel(req->cid());
  43.             basicResponse(req->rid(),req->cid(),true);
  44.         }
  45.         Channel::ptr getChannel(const std::string& cid)
  46.         {
  47.             return _chmp->getChannel(cid);
  48.         }
  49.     private:
  50.         void basicResponse(const std::string &rid, const std::string &cid, bool ok)
  51.         {
  52.             BasicCommonResponse rsp;
  53.             rsp.set_rid(rid);
  54.             rsp.set_cid(cid);
  55.             rsp.set_ok(ok);
  56.             _codec->send(_conn, rsp);
  57.         }
  59.     private:
  60.         muduo::net::TcpConnectionPtr _conn; // 关联的连接
  61.         ProtobufCodecPtr _codec;            // protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
  62.         QueueConsumerManager::ptr _qcmp;    // 消费者管理句柄
  63.         VirtualHost::ptr _vhp;              // 虚拟机管理句柄
  64.         ThreadPool::ptr _pool;              // 线程池管理句柄
  65.         ChannelManager::ptr _chmp;          // 信道管理句柄
  66.     };
  68.     class ConnectionManager
  69.     {
  70.     public:
  71.         using ptr=std::shared_ptr<ConnectionManager>;
  72.         ConnectionManager()
  73.         {
  74.             DLOG("new ConnectionManager: %p",this);
  75.         }
  76.         ~ConnectionManager()
  77.         {
  78.             DLOG("delete ConnectionManager: %p",this);
  79.         }
  80.         void newConnection(const muduo::net::TcpConnectionPtr &conn
  81.             , const ProtobufCodecPtr &codec
  82.             , const QueueConsumerManager::ptr &qcmp
  83.             , const VirtualHost::ptr &vhp
  84.             , const ThreadPool::ptr &pool)
  85.         {
  86.             std::unique_lock<std::mutex> lock(_mutex);
  87.             //连接是否已存在
  88.             auto it=_conns.find(conn);
  89.             if(it!=_conns.end())
  90.             {
  91.                 return;
  92.             }
  93.             //构建对象
  94.             Connection::ptr cop=std::make_shared<Connection>(conn,codec,qcmp,vhp,pool);
  95.             //添加管理
  96.             _conns[conn]=cop;
  97.         }
  98.         
  99.         void deleteConnection(const muduo::net::TcpConnectionPtr &conn)
  100.         {
  101.             std::unique_lock<std::mutex> lock(_mutex);
  102.             //连接是否不存在
  103.             auto it=_conns.find(conn);
  104.             if(it==_conns.end())
  105.             {
  106.                 return;
  107.             }
  108.             //删除
  109.             _conns.erase(it);
  110.         }
  111.         Connection::ptr getConnection(const muduo::net::TcpConnectionPtr &conn)
  112.         {
  113.             std::unique_lock<std::mutex> lock(_mutex);
  114.             //连接是否不存在
  115.             auto it=_conns.find(conn);
  116.             if(it==_conns.end())
  117.             {
  118.                 return Connection::ptr();
  119.             }
  120.             return it->second;
  121.         }
  122.     private:
  123.         std::mutex _mutex;
  124.         std::unordered_map<muduo::net::TcpConnectionPtr,Connection::ptr> _conns;
  125.     };
  126. };
复制代码
      

  • 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：服务器持有的假造主机。 队列、交换机 、绑定、消息等数据都是通过假造主机管理。
                                   登录后复制                        
  1. #pragma once
  2. #include "../mqthird/include/muduo/proto/codec.h"
  3. #include "../mqthird/include/muduo/proto/dispatcher.h"
  5. #include "../mqthird/include/muduo/base/Logging.h"
  6. #include "../mqthird/include/muduo/base/Mutex.h"
  7. #include "../mqthird/include/muduo/net/EventLoop.h"
  8. #include "../mqthird/include/muduo/net/TcpServer.h"
  10. #include "connection.hpp"
  11. namespace lsxmq
  12. {
  13.     //typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
  14.     #define DBFILE "/meta.db"
  15.     #define HOSTNAME "lsxhost"
  16.     class Server
  17.     {
  18.     public:
  19.         Server(uint16_t port,const std::string& basedir)
  20.             : _server(&_loop, muduo::net::InetAddress(port)
  21.             , "ProtobufServer", muduo::net::TcpServer::kReusePort)
  22.             , _dispatcher(std::bind(&Server::onUnknownMessage
  23.                     , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
  24.             , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher
  25.                     , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
  26.             ,_consumerManager(std::make_shared<QueueConsumerManager>())
  27.             ,_virtualHost(std::make_shared<VirtualHost>(HOSTNAME,basedir+DBFILE,basedir))
  28.             ,_threadPool(std::make_shared<ThreadPool>())
  29.             ,_connectionManager(std::make_shared<ConnectionManager>())
  30.         {
  31.             //需要初始化队列的消费者结构,因为消息队列是持久化的,构造时会自动加载,但消费者的队列结构不会
  32.             //1.获取所有队列
  33.             auto mqmap=_virtualHost->allQueue();
  34.             for(auto& mqp:mqmap)
  35.             {
  36.                 _consumerManager->initQueueConsumer(mqp.second->name);
  37.             }
  38.             // 注册业务处理请求函数
  39.             _dispatcher.registerMessageCallback<OpenChannelRequest>(std::bind(&Server::onOpenChannel
  40.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  41.             _dispatcher.registerMessageCallback<CloseChannelRequest>(std::bind(&Server::onCloseChannel
  42.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  43.             _dispatcher.registerMessageCallback<DeclareExchangeRequest>(std::bind(&Server::onDeclareExchange
  44.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  45.             _dispatcher.registerMessageCallback<DeleteExchangeRequest>(std::bind(&Server::onDeleteExchange
  46.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  47.             _dispatcher.registerMessageCallback<DeclareMessageQueueRequest>(std::bind(&Server::onDeclareMessageQueue
  48.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  49.             _dispatcher.registerMessageCallback<DeleteMessageQueueRequest>(std::bind(&Server::onDeleteMessageQueue
  50.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  51.             _dispatcher.registerMessageCallback<BindRequest>(std::bind(&Server::onBind
  52.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  53.             _dispatcher.registerMessageCallback<UnbindRequest>(std::bind(&Server::onUnbind
  54.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  55.             _dispatcher.registerMessageCallback<BasicConsumeRequest>(std::bind(&Server::onBasicConsume
  56.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  57.             _dispatcher.registerMessageCallback<BasicCancelRequest>(std::bind(&Server::onBasicCancel
  58.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  59.             _dispatcher.registerMessageCallback<BasicPublishRequest>(std::bind(&Server::onBasicPublish
  60.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  61.             _dispatcher.registerMessageCallback<BasicAckRequest>(std::bind(&Server::onBasicAckRequest
  62.                 , this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  63.             //注册服务器的回调函数
  64.             _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
  65.             _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
  66.                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  67.         }
  68.         void start()
  69.         {
  70.             _server.start();
  71.             _loop.loop();
  72.         }
  74.     private:
  75.         //打开信道
  76.         void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const OpenChannelRequestPtr &message, muduo::Timestamp)
  77.         {
  78.             //获取连接
  79.             auto selfConnP=_connectionManager->getConnection(conn);
  80.             if(selfConnP.get()==nullptr)
  81.             {
  82.                 ELOG("打开信道失败，因为连接不存在");
  83.                 return ;
  84.             }
  85.             selfConnP->openChannel(message);
  86.         }
  87.         //关闭信道
  88.         void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const CloseChannelRequestPtr &message, muduo::Timestamp)
  89.         {
  90.             //获取连接
  91.             auto selfConnP=_connectionManager->getConnection(conn);
  92.             if(selfConnP.get()==nullptr)
  93.             {
  94.                 ELOG("关闭信道失败，因为连接不存在");
  95.                 return ;
  96.             }
  97.             selfConnP->closeChannel(message);
  98.         }
  99.         //声明交换机
  100.         void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const DeclareExchangeRequestPtr &message, muduo::Timestamp)
  101.         {
  102.             //1.获取连接
  103.             auto selfConnP=_connectionManager->getConnection(conn);
  104.             if(selfConnP.get()==nullptr)
  105.             {
  106.                 ELOG("声明交换机失败，因为连接不存在");
  107.                 return ;
  108.             }
  109.             //2.获取信道
  110.             auto selfChannelP=selfConnP->getChannel(message->cid());
  111.             if(selfChannelP.get()==nullptr)
  112.             {
  113.                 ELOG("声明交换机失败，因为信道不存在");
  114.                 return ;
  115.             }
  116.             //声明交换机
  117.             selfChannelP->declareExchange(message);
  118.         }
  119.         //删除交换机
  120.         void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const DeleteExchangeRequestPtr &message, muduo::Timestamp)
  121.         {
  122.             //1.获取连接
  123.             auto selfConnP=_connectionManager->getConnection(conn);
  124.             if(selfConnP.get()==nullptr)
  125.             {
  126.                 ELOG("删除交换机失败，因为连接不存在");
  127.                 return ;
  128.             }
  129.             //2.获取信道
  130.             auto selfChannelP=selfConnP->getChannel(message->cid());
  131.             if(selfChannelP.get()==nullptr)
  132.             {
  133.                 ELOG("删除交换机失败，因为信道不存在");
  134.                 return ;
  135.             }
  136.             //删除交换机
  137.             selfChannelP->deleteExchange(message);
  138.         }
  139.         //声明消息队列
  140.         void onDeclareMessageQueue(const muduo::net::TcpConnectionPtr &conn, const DeclareMessageQueueRequestPtr &message, muduo::Timestamp)
  141.         {
  142.             //1.获取连接
  143.             auto selfConnP=_connectionManager->getConnection(conn);
  144.             if(selfConnP.get()==nullptr)
  145.             {
  146.                 ELOG("声明消息队列失败，因为连接不存在");
  147.                 return ;
  148.             }
  149.             //2.获取信道
  150.             auto selfChannelP=selfConnP->getChannel(message->cid());
  151.             if(selfChannelP.get()==nullptr)
  152.             {
  153.                 ELOG("声明消息队列失败，因为信道不存在");
  154.                 return ;
  155.             }
  156.             //声明消息队列
  157.             selfChannelP->declareMessageQueue(message);
  158.         }
  159.         //删除消息队列
  160.         void onDeleteMessageQueue(const muduo::net::TcpConnectionPtr &conn, const DeleteMessageQueueRequestPtr &message, muduo::Timestamp)
  161.         {
  162.             //1.获取连接
  163.             auto selfConnP=_connectionManager->getConnection(conn);
  164.             if(selfConnP.get()==nullptr)
  165.             {
  166.                 ELOG("删除消息队列失败，因为连接不存在");
  167.                 return ;
  168.             }
  169.             //2.获取信道
  170.             auto selfChannelP=selfConnP->getChannel(message->cid());
  171.             if(selfChannelP.get()==nullptr)
  172.             {
  173.                 ELOG("删除消息队列失败，因为信道不存在");
  174.                 return ;
  175.             }
  176.             //删除消息队列
  177.             selfChannelP->deleteMessageQueue(message);
  178.         }
  179.         //绑定
  180.         void onBind(const muduo::net::TcpConnectionPtr &conn, const BindRequestPtr &message, muduo::Timestamp)
  181.         {
  182.             //1.获取连接
  183.             auto selfConnP=_connectionManager->getConnection(conn);
  184.             if(selfConnP.get()==nullptr)
  185.             {
  186.                 ELOG("绑定失败，因为连接不存在");
  187.                 return ;
  188.             }
  189.             //2.获取信道
  190.             auto selfChannelP=selfConnP->getChannel(message->cid());
  191.             if(selfChannelP.get()==nullptr)
  192.             {
  193.                 ELOG("绑定失败，因为信道不存在");
  194.                 return ;
  195.             }
  196.             //绑定
  197.             selfChannelP->bind(message);
  198.         }
  199.         //解绑
  200.         void onUnbind(const muduo::net::TcpConnectionPtr &conn, const UnbindRequestPtr &message, muduo::Timestamp)
  201.         {
  202.             //1.获取连接
  203.             auto selfConnP=_connectionManager->getConnection(conn);
  204.             if(selfConnP.get()==nullptr)
  205.             {
  206.                 ELOG("解绑失败，因为连接不存在");
  207.                 return ;
  208.             }
  209.             //2.获取信道
  210.             auto selfChannelP=selfConnP->getChannel(message->cid());
  211.             if(selfChannelP.get()==nullptr)
  212.             {
  213.                 ELOG("解绑失败，因为信道不存在");
  214.                 return ;
  215.             }
  216.             //解绑
  217.             selfChannelP->unbind(message);
  218.         }
  219.         //订阅
  220.         void onBasicConsume(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeRequestPtr &message, muduo::Timestamp)
  221.         {
  222.             //1.获取连接
  223.             auto selfConnP=_connectionManager->getConnection(conn);
  224.             if(selfConnP.get()==nullptr)
  225.             {
  226.                 ELOG("订阅失败，因为连接不存在");
  227.                 return ;
  228.             }
  229.             //2.获取信道
  230.             auto selfChannelP=selfConnP->getChannel(message->cid());
  231.             if(selfChannelP.get()==nullptr)
  232.             {
  233.                 ELOG("订阅失败，因为信道不存在");
  234.                 return ;
  235.             }
  236.             //订阅
  237.             selfChannelP->basicConsume(message);
  238.         }
  239.         //取消订阅
  240.         void onBasicCancel(const muduo::net::TcpConnectionPtr &conn, const BasicCancelRequestPtr &message, muduo::Timestamp)
  241.         {
  242.             //1.获取连接
  243.             auto selfConnP=_connectionManager->getConnection(conn);
  244.             if(selfConnP.get()==nullptr)
  245.             {
  246.                 ELOG("取消订阅失败，因为连接不存在");
  247.                 return ;
  248.             }
  249.             //2.获取信道
  250.             auto selfChannelP=selfConnP->getChannel(message->cid());
  251.             if(selfChannelP.get()==nullptr)
  252.             {
  253.                 ELOG("取消订阅失败，因为信道不存在");
  254.                 return ;
  255.             }
  256.             //取消订阅
  257.             selfChannelP->basicCancel(message);
  258.         }
  259.         //发布消息
  260.         void onBasicPublish(const muduo::net::TcpConnectionPtr &conn, const BasicPublishRequestPtr &message, muduo::Timestamp)
  261.         {
  262.             //1.获取连接
  263.             auto selfConnP=_connectionManager->getConnection(conn);
  264.             if(selfConnP.get()==nullptr)
  265.             {
  266.                 ELOG("发布消息失败，因为连接不存在");
  267.                 return ;
  268.             }
  269.             //2.获取信道
  270.             auto selfChannelP=selfConnP->getChannel(message->cid());
  271.             if(selfChannelP.get()==nullptr)
  272.             {
  273.                 ELOG("发布消息失败，因为信道不存在");
  274.                 return ;
  275.             }
  276.             //发布消息
  277.             selfChannelP->basicPublish(message);
  278.         }
  279.         //确认消息
  280.         void onBasicAckRequest(const muduo::net::TcpConnectionPtr &conn, const BasicAckRequestPtr &message, muduo::Timestamp)
  281.         {
  282.             //1.获取连接
  283.             auto selfConnP=_connectionManager->getConnection(conn);
  284.             if(selfConnP.get()==nullptr)
  285.             {
  286.                 ELOG("确认消息失败，因为连接不存在");
  287.                 return ;
  288.             }
  289.             //2.获取信道
  290.             auto selfChannelP=selfConnP->getChannel(message->cid());
  291.             if(selfChannelP.get()==nullptr)
  292.             {
  293.                 ELOG("确认消息失败，因为信道不存在");
  294.                 return ;
  295.             }
  296.             //确认消息
  297.             selfChannelP->basicAck(message);
  298.         }
  300.         void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,const std::shared_ptr<google::protobuf::Message>& message,muduo::Timestamp)
  301.         {
  302.             //收到未知消息
  303.             LOG_INFO<<message->GetTypeName();
  304.             conn->shutdown();
  305.         }
  306.         void onConnection(const muduo::net::TcpConnectionPtr &conn)
  307.         {
  308.             if (conn->connected())
  309.             {
  310.                 LOG_INFO << "连接成功";
  311.                 _connectionManager->newConnection(conn,_codec,_consumerManager,_virtualHost,_threadPool);
  312.             }
  313.             else
  314.             {
  315.                 LOG_INFO << "断开连接";
  316.                 _connectionManager->deleteConnection(conn);
  317.             }
  318.         }
  319.     private:
  320.         muduo::net::EventLoop _loop;
  321.         muduo::net::TcpServer _server;  // 服务器
  322.         ProtobufDispatcher _dispatcher; // 请求分发器对象--要向其中注册请求处理函数
  323.         ProtobufCodecPtr _codec;           // protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
  324.         QueueConsumerManager::ptr _consumerManager;     // 消费者管理句柄
  325.         VirtualHost::ptr _virtualHost;      // 虚拟机管理句柄
  326.         ThreadPool::ptr _threadPool;        // 线程池管理句柄
  327.         ConnectionManager::ptr _connectionManager;     // 连接管理句柄
  328.     };
  329. };
复制代码
      

  • 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、订阅者模块

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

  • 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库的通讯是异步的，因此在收到应答后，通过判定是否是等候的指定相应来进行同步)：发送消息时，还没有声明交换机成功怎么办？以是要同步
     信道操作：
     

  • 打开信道
  • 关闭信道
  • 声明交换机
  • 删除交换机
  • 声明队列
  • 删除队列
  • 绑定
  • 解绑
  • 订阅
  • 取消订阅
  • 发布消息
  • 确认消息
                                   登录后复制                        
  1. #pragma once
  2. #include "../mqcomm/logger.hpp"
  3. #include "../mqcomm/threadpool.hpp"
  4. #include "../mqcomm/proto.pb.h"
  5. #include "../mqcomm/message.pb.h"
  6. #include "consumer.hpp"
  7. #include "muduo/net/TcpConnection.h"
  8. #include "muduo/proto/codec.h"
  10. namespace lsxmq
  11. {
  12.     using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;
  13.     using BasicCommonResponsePtr=std::shared_ptr<BasicCommonResponse>;
  14.     using BasicConsumeResponsePtr=std::shared_ptr<basicConsumeResponse>;
  15.     class Channel
  16.     {
  17.     public:
  18.         using ptr=std::shared_ptr<Channel>;
  19.         Channel(const muduo::net::TcpConnectionPtr& conn
  20.                 ,const ProtobufCodecPtr& codec)
  21.                 :_cid(UUIDHelper::uuid()),_conn(conn),_codec(codec)
  22.         {}
  23.         ~Channel()
  24.         {
  25.             basicCancel();//析构时取消订阅
  26.         }
  27.         std::string cid()
  28.         {
  29.             return _cid;
  30.         }
  31.         //打开信道
  32.         bool openChannel()
  33.         {
  34.             //构造请求
  35.             OpenChannelRequest req;
  36.             req.set_rid(UUIDHelper::uuid());
  37.             req.set_cid(_cid);
  38.             //发送请求并等待
  39.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  40.             auto rsp=waitBasicResponse(req.rid());
  41.             if(rsp->ok()==false)
  42.             {
  43.                 ELOG("打开信道失败");
  44.             }
  45.             return rsp->ok();
  46.         }
  47.         //关闭信道
  48.         bool closeChannel()
  49.         {
  50.             //构造请求
  51.             CloseChannelRequest req;
  52.             req.set_rid(UUIDHelper::uuid());
  53.             req.set_cid(_cid);
  54.             //发送请求并等待
  55.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  56.             auto rsp=waitBasicResponse(req.rid());
  57.             if(rsp->ok()==false)
  58.             {
  59.                 ELOG("关闭信道失败");
  60.             }
  61.             return rsp->ok();
  62.         }
  63.         //声明交换机
  64.         bool declareExchange(const std::string& exchangeName
  65.                             ,ExchangeType exchangeType
  66.                             ,bool durable
  67.                             ,bool autoDelete
  68.                             ,google::protobuf::Map<std::string, std::string>& args)
  69.         {
  70.             //构造请求
  71.             DeclareExchangeRequest req;
  72.             req.set_rid(UUIDHelper::uuid());
  73.             req.set_cid(_cid);
  74.             req.set_exchange_name(exchangeName);
  75.             req.set_exchange_type(exchangeType);
  76.             req.set_durable(durable);
  77.             req.set_auto_delete(autoDelete);
  78.             req.mutable_args()->swap(args);
  79.             //发送请求并等待
  80.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  81.             auto rsp=waitBasicResponse(req.rid());
  82.             if(rsp->ok()==false)
  83.             {
  84.                 ELOG("声明交换机失败");
  85.             }
  86.             return rsp->ok();
  87.         }
  88.         //删除交换机
  89.         bool deleteExchange(const std::string& exchangeName)
  90.         {
  91.             //构造请求
  92.             DeleteExchangeRequest req;
  93.             req.set_rid(UUIDHelper::uuid());
  94.             req.set_cid(_cid);
  95.             req.set_exchange_name(exchangeName);
  96.             //发送请求并等待
  97.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  98.             auto rsp=waitBasicResponse(req.rid());
  99.             if(rsp->ok()==false)
  100.             {
  101.                 ELOG("删除交换机失败");
  102.             }
  103.             return rsp->ok();
  104.         }
  105.         //声明消息队列
  106.         bool declareMessageQueue(const std::string& queueName
  107.                                 ,bool exclusive
  108.                                 ,bool durable
  109.                                 ,bool autoDelete
  110.                                 ,google::protobuf::Map<std::string, std::string>& args)
  111.         {
  112.             //构造请求
  113.             DeclareMessageQueueRequest req;
  114.             req.set_rid(UUIDHelper::uuid());
  115.             req.set_cid(_cid);
  116.             req.set_queue_name(queueName);
  117.             req.set_exclusive(exclusive);
  118.             req.set_durable(durable);
  119.             req.set_auto_delete(autoDelete);
  120.             req.mutable_args()->swap(args);
  121.             //发送请求并等待
  122.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  123.             auto rsp=waitBasicResponse(req.rid());
  124.             if(rsp->ok()==false)
  125.             {
  126.                 ELOG("声明消息队列失败");
  127.             }
  128.             return rsp->ok();
  129.         }
  130.         //删除消息队列
  131.         bool deleteMessageQueue(const std::string& queueName)
  132.         {
  133.             //构造请求
  134.             DeleteMessageQueueRequest req;
  135.             req.set_rid(UUIDHelper::uuid());
  136.             req.set_cid(_cid);
  137.             req.set_queue_name(queueName);
  138.             //发送请求并等待
  139.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  140.             auto rsp=waitBasicResponse(req.rid());
  141.             if(rsp->ok()==false)
  142.             {
  143.                 ELOG("删除消息队列失败");
  144.             }
  145.             return rsp->ok();
  146.         }
  147.         //绑定
  148.         bool bind(const std::string& exchangeName
  149.                     ,const std::string& queueName
  150.                     ,const std::string& bindingKey)
  151.         {
  152.             //构造请求
  153.             BindRequest req;
  154.             req.set_rid(UUIDHelper::uuid());
  155.             req.set_cid(_cid);
  156.             req.set_exchange_name(exchangeName);
  157.             req.set_queue_name(queueName);
  158.             req.set_binding_key(bindingKey);
  159.             //发送请求并等待
  160.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  161.             auto rsp=waitBasicResponse(req.rid());
  162.             if(rsp->ok()==false)
  163.             {
  164.                 ELOG("绑定失败");
  165.             }
  166.             return rsp->ok();
  167.         }
  168.         //解绑
  169.         bool unbind(const std::string& exchangeName
  170.                     ,const std::string& queueName)
  171.         {
  172.             //构造请求
  173.             BindRequest req;
  174.             req.set_rid(UUIDHelper::uuid());
  175.             req.set_cid(_cid);
  176.             req.set_exchange_name(exchangeName);
  177.             req.set_queue_name(queueName);
  178.             //发送请求并等待
  179.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  180.             auto rsp=waitBasicResponse(req.rid());
  181.             if(rsp->ok()==false)
  182.             {
  183.                 ELOG("解绑失败");
  184.             }
  185.             return rsp->ok();
  186.         }
  187.         //订阅
  188.         bool basicConsume(const std::string& consumerTag
  189.                         ,const std::string& queueName
  190.                         ,bool autoAck
  191.                         ,const ConsumerCallBack& cb)
  192.         {
  193.             //构造请求
  194.             if(_consumer.get()!=nullptr)
  195.             {
  196.                 DLOG("已订阅，不可重复订阅")
  197.                 return false;
  198.             }
  199.             BasicConsumeRequest req;
  200.             req.set_rid(UUIDHelper::uuid());
  201.             req.set_cid(_cid);
  202.             req.set_consumer_tag(consumerTag);
  203.             req.set_queue_name(queueName);
  204.             req.set_auto_ack(autoAck);
  205.             //发送请求并等待
  206.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  207.             auto rsp=waitBasicResponse(req.rid());
  208.             if(rsp->ok()==false)
  209.             {
  210.                 ELOG("订阅失败");
  211.                 return false;
  212.             }
  213.             //订阅成功，要构造对应的消费者
  214.             _consumer=std::make_shared<Consumer>(consumerTag,queueName,autoAck,cb);
  215.             return rsp->ok();
  216.         }
  217.         //取消订阅
  218.         bool basicCancel()
  219.         {
  220.             if(_consumer.get()==nullptr)
  221.             {
  222.                 return true;
  223.             }
  224.             //构造请求
  225.             BasicCancelRequest req;
  226.             req.set_rid(UUIDHelper::uuid());
  227.             req.set_cid(_cid);
  228.             req.set_consumer_tag(_consumer->tag);
  229.             req.set_queue_name(_consumer->qname);
  230.             //发送请求并等待
  231.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  232.             auto rsp=waitBasicResponse(req.rid());
  233.             if(rsp->ok()==false)
  234.             {
  235.                 ELOG("取消订阅失败");
  236.                 return false;
  237.             }
  238.             //取消订阅时清理消费者对象
  239.             _consumer.reset();
  240.             return rsp->ok();
  241.         }
  242.         //发布消息
  243.         bool basicPublish(const std::string& exchangeName
  244.                             ,const std::string& body
  245.                             ,BasicProperties* bp)
  246.         {
  247.             //构造请求
  248.             BasicPublishRequest req;
  249.             req.set_rid(UUIDHelper::uuid());
  250.             req.set_cid(_cid);
  251.             req.set_exchange_name(exchangeName);
  252.             req.set_body(body);
  253.             if(bp!=nullptr)
  254.             {
  255.                 req.mutable_properties()->set_id(bp->id());
  256.                 req.mutable_properties()->set_deliverymode(bp->deliverymode());
  257.                 req.mutable_properties()->set_routing_key(bp->routing_key());
  258.             }
  259.             //发送请求并等待
  260.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  261.             auto rsp=waitBasicResponse(req.rid());
  262.             if(rsp->ok()==false)
  263.             {
  264.                 ELOG("发布消息失败");
  265.             }
  266.             return rsp->ok();
  267.         }
  268.         //确认消息
  269.         bool basicAck(const std::string& mid)
  270.         {
  271.             if(_consumer.get()==nullptr)
  272.             {
  273.                 ELOG("消息确认时，没有找到消费者")
  274.                 return false;
  275.             }
  276.             //构造请求
  277.             BasicAckRequest req;
  278.             req.set_rid(UUIDHelper::uuid());
  279.             req.set_cid(_cid);
  280.             req.set_queue_name(_consumer->qname);
  281.             req.set_mid(mid);
  282.             //发送请求并等待
  283.             _codec->send(_conn,req);//send是异步的，所以需要我们手动进行等待
  284.             auto rsp=waitBasicResponse(req.rid());
  285.             if(rsp->ok()==false)
  286.             {
  287.                 ELOG("确认消息失败");
  288.             }
  289.             return rsp->ok();
  290.         }
  291.     public:
  292.         //收到基础响应时，向hash map中添加响应
  293.         void putBasicResponse(const BasicCommonResponsePtr& rsp)
  294.         {
  295.             std::unique_lock<std::mutex> lock(_mutex);
  296.             _basicRsp.insert(std::make_pair(rsp->rid(),rsp));
  297.             _cond.notify_all();//唤醒所有等待线程，只有满足条件的才能竞争锁
  298.         }
  299.         //收到消息推送后，需要信道找到对应的消费者，通过回调函数进行处理
  300.         void consume(const BasicConsumeResponsePtr& rsp)
  301.         {
  302.             if(_consumer.get()==nullptr)
  303.             {
  304.                 ELOG("消费者不存在");
  305.                 return;
  306.             }
  307.             if(rsp->consumer_tag()!=_consumer->tag)
  308.             {
  309.                 ELOG("收到错误的响应，消费者标志匹配失败");
  310.                 return;
  311.             }
  312.             //执行回调函数
  313.             _consumer->callback(rsp->consumer_tag(),rsp->mutable_properties(),rsp->body());
  314.         }
  315.     private:
  316.         //等待响应
  317.         BasicCommonResponsePtr waitBasicResponse(const std::string& rid)
  318.         {
  319.             std::unique_lock<std::mutex> lock(_mutex);
  320.             _cond.wait(lock,[&rid,this](){
  321.                 return _basicRsp.find(rid)!=_basicRsp.end();//等到收到了对应的响应才退出
  322.             });
  323.             //拿到响应，然后再删除
  324.             BasicCommonResponsePtr rsp=_basicRsp[rid];
  325.             _basicRsp.erase(rid);
  326.             return rsp;
  327.         }
  328.     private:
  329.         std::string _cid;  //信道id
  330.         muduo::net::TcpConnectionPtr _conn;  //信道关联的网络通信连接
  331.         ProtobufCodecPtr _codec;  //protobuf协议处理器
  332.         Consumer::ptr _consumer;  //信道对应的消费者,当确定了信道角色后构造(订阅操作后)
  333.         std::mutex _mutex;
  334.         std::condition_variable _cond;
  335.         std::unordered_map<std::string,BasicCommonResponsePtr> _basicRsp;  //<请求id,响应>
  336.     };
  338.     class ChannelManager
  339.     {
  340.     public:
  341.         using ptr=std::shared_ptr<ChannelManager>;
  342.         ChannelManager()
  343.         {
  344.             DLOG("new ChannelManager: %p",this);
  345.         }
  346.         ~ChannelManager()
  347.         {
  348.             DLOG("delete ChannelManager: %p",this);
  349.         }
  350.         //打开信道
  351.         Channel::ptr createChannel(const muduo::net::TcpConnectionPtr& conn
  352.             ,const ProtobufCodecPtr& codec)
  353.         {
  354.             std::unique_lock<std::mutex> lock(_mutex);
  355.             //队列是否已存在？不需要担心，因为创建信道会自动生成id
  356.             //构造信道对象
  357.             Channel::ptr chp=std::make_shared<Channel>(conn,codec);
  358.             //添加管理
  359.             _channels[chp->cid()]=chp;
  360.             return chp;
  361.         }
  362.         void deleteChannel(const std::string& cid)//关闭信道
  363.         {
  364.             std::unique_lock<std::mutex> lock(_mutex);
  365.             //队列是否不存在
  366.             auto it=_channels.find(cid);
  367.             if(it==_channels.end())
  368.             {
  369.                 return;
  370.             }
  371.             //删除信道
  372.             _channels.erase(cid);
  373.         }
  374.         Channel::ptr getChannel(const std::string& cid)
  375.         {
  376.             std::unique_lock<std::mutex> lock(_mutex);
  377.             //队列是否不存在
  378.             auto it=_channels.find(cid);
  379.             if(it==_channels.end())
  380.             {
  381.                 return Channel::ptr();
  382.             }
  383.             return it->second;
  384.         }
  385.     private:
  386.         std::mutex _mutex;
  387.         std::unordered_map<std::string,Channel::ptr> _channels;
  388.     };
  389. };
复制代码
      

  • 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，
  • ⼀个是当收到消息后进⾏异步处理的⼯作线程池。
     这两项都不是以连接为单元进⾏创建的，⽽是创建后，可以⽤以多个连接中，因此单独进⾏封装。
                                   登录后复制                        
  1. #pragma once
  2. #include "muduo/net/EventLoopThreadPool.h"
  3. #include "../mqcomm/logger.hpp"
  4. #include "../mqcomm/threadpool.hpp"
  6. namespace lsxmq
  7. {
  8.     class AsyncWorker
  9.     {
  10.     public:
  11.         using ptr=std::shared_ptr<AsyncWorker>;
  12.         muduo::net::EventLoopThreadPool _loopthread;
  13.         ThreadPool _pool;
  14.     };
  15. };
复制代码
      

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
                       4、连接管理模块

     在客⼾端这边，RabbitMQ弱化了客⼾端的概念，因为⽤⼾所需的服务都是通过信道来提供的，因此操作头脑转换为先创建连接，通过连接创建信道，通过信道提供服务这⼀流程。
     这个模块同样是针对muduo库客⼾端连接的⼆次封装，向⽤⼾提供创建channel信道的接⼝，创建信道后，可以通过信道来获取指定服务。
                                   登录后复制                        
  1. #pragma once
  2. #include "muduo/proto/dispatcher.h"
  3. #include "muduo/proto/codec.h"
  5. #include "muduo/base/Logging.h"
  6. #include "muduo/base/Mutex.h"
  7. #include "muduo/net/EventLoop.h"
  8. #include "muduo/net/TcpClient.h"
  9. #include "muduo/base/CountDownLatch.h"
  10. #include "muduo/net/EventLoopThread.h"
  12. #include "worker.hpp"
  13. #include "channel.hpp"
  14. #include <string>
  16. namespace lsxmq
  17. {
  18.     typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
  20.     class Connection
  21.     {
  22.     public:
  23.     using ptr=std::shared_ptr<Connection>;
  24.     Connection(const std::string &serverIp, uint16_t serverPort,const AsyncWorker::ptr& worker)
  25.             : _latch(1)
  26.             ,_worker(worker)
  27.             , _client(_worker->_loopthread.startLoop()
  28.                 , muduo::net::InetAddress(serverIp, serverPort), "ProtobufClient")
  29.             , _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1
  30.                 , std::placeholders::_2, std::placeholders::_3))
  31.             , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher
  32.                 , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
  33.             ,_channels(std::make_shared<ChannelManager>())
  34.         {
  35.             // 注册处理函数
  36.             _dispatcher.registerMessageCallback<BasicCommonResponse>(std::bind(&Connection::onBasicCommonResponse, this
  37.                 , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  38.             _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::onBasicConsumeResponse, this
  39.                 , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  40.             // 客户端注册回调函数
  41.             _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
  42.             _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get()
  43.                 , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
  44.                
  45.             //连接服务器
  46.             _client.connect();
  47.             _latch.wait(); // 等待连接服务器成功
  48.         }
  49.         //打开信道
  50.         Channel::ptr openChannel()
  51.         {
  52.             //构造一个信道
  53.             auto channel=_channels->createChannel(_conn,_codec);
  54.             //打开信道
  55.             auto ret=channel->openChannel();
  56.             if(ret==false)
  57.             {
  58.                 ELOG("打开信道失败");
  59.                 _channels->deleteChannel(channel->cid());
  60.                 return Channel::ptr();
  61.             }
  62.             return channel;
  63.         }
  64.         //关闭信道
  65.         bool closeChannel(const Channel::ptr& channel)
  66.         {
  67.             auto ret=channel->closeChannel();
  68.             if(ret==false)
  69.             {
  70.                 ELOG("关闭信道失败");
  71.                 return false;
  72.             }
  73.             _channels->deleteChannel(channel->cid());
  74.             return true;
  75.         }
  76.     private:
  77.         //收到普通响应
  78.         void onBasicCommonResponse(const muduo::net::TcpConnectionPtr &conn, const BasicCommonResponsePtr &message, muduo::Timestamp)
  79.         {
  80.             //1.获取信道
  81.             auto channel=_channels->getChannel(message->cid());
  82.             if(channel.get()==nullptr)
  83.             {
  84.                 ELOG("处理普通响应失败，因为获取信道失败");
  85.                 return;
  86.             }
  87.             //2.将响应加入hash map
  88.             channel->putBasicResponse(message);
  89.         }
  90.         //收到推送消息
  91.         void onBasicConsumeResponse(const muduo::net::TcpConnectionPtr &conn,const BasicConsumeResponsePtr& message, muduo::Timestamp)
  92.         {
  93.             //1.获取信道
  94.             auto channel=_channels->getChannel(message->cid());
  95.             if(channel.get()==nullptr)
  96.             {
  97.                 ELOG("处理普通响应失败，因为获取信道失败");
  98.                 return;
  99.             }
  100.             //2.对收到的消息进行处理，由工作线程池完成
  101.             auto task=std::bind(&Channel::consume,channel.get(),message);
  102.             _worker->_pool.push(task);
  103.         }
  104.         void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn,const MessagePtr& message,muduo::Timestamp)
  105.         {
  106.             LOG_INFO<<message->GetTypeName();
  107.             //收到了未定义的消息，断开连接
  108.             conn->shutdown();
  109.         }
  110.         void onConnection(const muduo::net::TcpConnectionPtr &conn)
  111.         {
  112.             if (conn->connected())
  113.             {
  114.                 LOG_INFO << "连接成功";
  115.                 _conn = conn;
  116.                 // 解除阻塞
  117.                 _latch.countDown();
  118.             }
  119.             else
  120.             {
  121.                 LOG_INFO << "断开连接";
  122.             }
  123.         }
  125.     private:
  126.         muduo::CountDownLatch _latch;            // 用于同步等待连接服务器成功
  127.         AsyncWorker::ptr _worker;  //异步工作线程
  128.         muduo::net::TcpClient _client;           // 客户端
  129.         ProtobufDispatcher _dispatcher;          // 派发器
  130.         ProtobufCodecPtr _codec;                    // protobuf协议处理器
  131.         muduo::net::TcpConnectionPtr _conn;
  132.         ChannelManager::ptr _channels;  //信道管理句柄
  133.     };
  134. }
复制代码
      

  • 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.
                       综合测试

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

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

  • 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.
                       直接交换模式：
                                   登录后复制                        
  1. //订阅者
  2. #include "connection.hpp"
  3. void callBack(const lsxmq::Channel::ptr& channel,const std::string& tag,lsxmq::BasicProperties* bp,const std::string& body)
  4. {
  5.     std::cout<<"消费者"<<tag<<"收到一条消息："<<body<<std::endl;
  6.     //发送响应
  7.     channel->basicAck(bp->id());
  8. }
  9. int main(int argc,char* argv[])
  10. {
  11.     if(argc!=2)
  12.     {
  13.         std::cout<<"格式错误"<<std::endl;
  14.         return 0;
  15.     }
  16.     std::string qname=argv[1];
  17.     //1.构造异步工作者
  18.     lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
  19.     //2.构造连接
  20.     lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
  21.     //3.打开信道
  22.     lsxmq::Channel::ptr channel=conn->openChannel();
  23.     //4.通过信道进行操作
  24.         //4.1.声明一台交换机
  25.     google::protobuf::Map<std::string, std::string> args;
  26.     channel->declareExchange("exchange1",lsxmq::ExchangeType::DIRECT,true,false,args);
  27.         //4.2.声明队列queue1
  28.     channel->declareMessageQueue("queue1",false,true,false,args);
  29.         //4.3.声明队列queue2
  30.     channel->declareMessageQueue("queue2",false,true,false,args);
  31.         //4.4.将队列与交换机绑定
  32.     channel->bind("exchange1","queue1","queue1");
  33.     channel->bind("exchange1","queue2","queue2");
  34.         //4.5订阅一个队列
  35.     auto cb=std::bind(callBack,channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
  36.     channel->basicConsume("consumer1",qname,false,cb);
  37.     //5.等待
  38.     while(1)
  39.     {
  40.         std::this_thread::sleep_for(std::chrono::seconds(3));
  41.     }
  42.     //5.关闭信道
  43.         conn->closeChannel(channel);
  44.     return 0;
  45. }
复制代码
      

  • 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.
                                                     登录后复制                        
  1. //发布者
  2. #include "connection.hpp"
  4. int main()
  5. {
  6.     //1.构造异步工作者
  7.     lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
  8.     //2.构造连接
  9.     lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
  10.     //3.打开信道
  11.     lsxmq::Channel::ptr channel=conn->openChannel();
  12.     //4.通过信道进行操作
  13.         //4.1.声明一台交换机
  14.     google::protobuf::Map<std::string, std::string> args;
  15.     channel->declareExchange("exchange1",lsxmq::ExchangeType::DIRECT,true,false,args);
  16.         //4.2.声明队列queue1
  17.     channel->declareMessageQueue("queue1",false,true,false,args);
  18.         //4.3.声明队列queue2
  19.     channel->declareMessageQueue("queue2",false,true,false,args);
  20.         //4.4.将队列与交换机绑定
  21.     channel->bind("exchange1","queue1","queue1");
  22.     channel->bind("exchange1","queue2","queue2");
  23.         //4.5循环向交换机发布消息
  24.     for(int i=1;i<=10;++i)
  25.     {
  26.         lsxmq::BasicProperties bp;
  27.         bp.set_id(lsxmq::UUIDHelper::uuid());
  28.         bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
  29.         bp.set_routing_key("queue1");
  30.         channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
  31.     }
  32.     for(int i=11;i<=15;++i)
  33.     {
  34.         lsxmq::BasicProperties bp;
  35.         bp.set_id(lsxmq::UUIDHelper::uuid());
  36.         bp.set_deliverymode(lsxmq::DeliveryMode::UNDURABLE);
  37.         bp.set_routing_key("queue2");
  38.         channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
  39.     }
  40.     //5.关闭信道
  41.         conn->closeChannel(channel);
  42.     return 0;
  43. }
复制代码
      

  • 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.
                       主题交换模式：
                                   登录后复制                        
  1. //订阅者
  2. #include "connection.hpp"
  3. void callBack(const lsxmq::Channel::ptr& channel,const std::string& tag,lsxmq::BasicProperties* bp,const std::string& body)
  4. {
  5.     std::cout<<"消费者"<<tag<<"收到一条消息："<<body<<std::endl;
  6.     //发送响应
  7.     channel->basicAck(bp->id());
  8. }
  9. int main(int argc,char* argv[])
  10. {
  11.     if(argc!=2)
  12.     {
  13.         std::cout<<"格式错误"<<std::endl;
  14.         return 0;
  15.     }
  16.     std::string qname=argv[1];
  17.     //1.构造异步工作者
  18.     lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
  19.     //2.构造连接
  20.     lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
  21.     //3.打开信道
  22.     lsxmq::Channel::ptr channel=conn->openChannel();
  23.     //4.通过信道进行操作
  24.         //4.1.声明一台交换机
  25.     google::protobuf::Map<std::string, std::string> args;
  26.     channel->declareExchange("exchange1",lsxmq::ExchangeType::TOPIC,true,false,args);
  27.         //4.2.声明队列queue1
  28.     channel->declareMessageQueue("queue1",false,true,false,args);
  29.         //4.3.声明队列queue2
  30.     channel->declareMessageQueue("queue2",false,true,false,args);
  31.         //4.4.将队列与交换机绑定
  32.     channel->bind("exchange1","queue1","sport.#");
  33.     channel->bind("exchange1","queue2","music.#");
  34.         //4.5订阅一个队列
  35.     auto cb=std::bind(callBack,channel,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
  36.     channel->basicConsume("consumer1",qname,false,cb);
  37.     //5.等待
  38.     while(1)
  39.     {
  40.         std::this_thread::sleep_for(std::chrono::seconds(3));
  41.     }
  42.     //5.关闭信道
  43.         conn->closeChannel(channel);
  44.     return 0;
  45. }
复制代码
      

  • 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.
                                                     登录后复制                        
  1. //发布者
  2. #include "connection.hpp"
  4. int main()
  5. {
  6.     //1.构造异步工作者
  7.     lsxmq::AsyncWorker::ptr worker=std::make_shared<lsxmq::AsyncWorker>();
  8.     //2.构造连接
  9.     lsxmq::Connection::ptr conn=std::make_shared<lsxmq::Connection>("127.0.0.1",8888,worker);
  10.     //3.打开信道
  11.     lsxmq::Channel::ptr channel=conn->openChannel();
  12.     //4.通过信道进行操作
  13.         //4.1.声明一台交换机
  14.     google::protobuf::Map<std::string, std::string> args;
  15.     channel->declareExchange("exchange1",lsxmq::ExchangeType::TOPIC,true,false,args);
  16.         //4.2.声明队列queue1
  17.     channel->declareMessageQueue("queue1",false,true,false,args);
  18.         //4.3.声明队列queue2
  19.     channel->declareMessageQueue("queue2",false,true,false,args);
  20.         //4.4.将队列与交换机绑定
  21.     channel->bind("exchange1","queue1","sport.#");
  22.     channel->bind("exchange1","queue2","music.#");
  23.         //4.5循环向交换机发布消息
  24.     for(int i=1;i<=10;++i)
  25.     {
  26.         lsxmq::BasicProperties bp;
  27.         bp.set_id(lsxmq::UUIDHelper::uuid());
  28.         bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
  29.         bp.set_routing_key("music.pop");
  30.         channel->basicPublish("exchange1","hello-world-"+std::to_string(i),&bp);
  31.     }
  33.     for(int i=1;i<=5;++i)
  34.     {
  35.         lsxmq::BasicProperties bp;
  36.         bp.set_id(lsxmq::UUIDHelper::uuid());
  37.         bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
  38.         bp.set_routing_key("music.pop.ajc");
  39.         channel->basicPublish("exchange1","good-morning-"+std::to_string(i),&bp);
  40.     }
  42.     for(int i=1;i<=5;++i)
  43.     {
  44.         lsxmq::BasicProperties bp;
  45.         bp.set_id(lsxmq::UUIDHelper::uuid());
  46.         bp.set_deliverymode(lsxmq::DeliveryMode::DURABLE);
  47.         bp.set_routing_key("news.gassip");
  48.         channel->basicPublish("exchange1","have-nice-day-"+std::to_string(i),&bp);
  49.     }
  50.     //5.关闭信道
  51.         conn->closeChannel(channel);
  52.     return 0;
  53. }
复制代码
      

  • 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企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4