【RabbitMQ 项目】服务端：数据管理模块之消息队列管理 ...

一.编写思绪

   

• 定义消息队列
  
  
  • 名字
    
  • 是否持久化
    
  
  
• 定义队列持久化类(持久化到 sqlite3)
  
  
  • 构造函数(只能成功，不能失败)
    
    
    • 如果数据库(文件)不存在则创建
      
    • 打开数据库
      
    • 打开 msg_queue_table 数据库表
      
    
    
  • 插入队列
    
  • 移除队列
    
  • 将数据库中的队列恢复到内存中
    传入一个哈希表，key 为名字，value 为队列的智能指针，填充该哈希表
    
  
  
• 定义队列管理类(包罗内存管理和持久化管理)
  
  
  • 构造函数：从数据库中恢复队列
    
  • 声明队列
    
  • 移除队列
    
  • 获取队列
    
  
  

  二.代码实践

MsgQueue.hpp:

1. #pragma once
2. #include "../common/Log.hpp"
3. #include "../common/Util.hpp"
4. #include "../common/Util.hpp"
5. #include <memory>
6. #include <unordered_map>
7. #include <mutex>
8. namespace ns_data
9. {
11.     class MsgQueue;
12.     using MsgQueuePtr = std::shared_ptr<MsgQueue>;
14.     /************
15.      * 定义消息队列
16.      * ****************/
17.     struct MsgQueue
18.     {
19.         std::string _name;
20.         bool _isDurable;
21.         MsgQueue(const std::string &name, bool isDurable)
22.             : _name(name),
23.               _isDurable(isDurable)
24.         {
25.         }
26.     };
28.     /*****************
29.      * 定义消息队列持久化类
30.      * ******************/
31.     class MsgQueueMapper
32.     {
33.     private:
34.         ns_util::Sqlite3Util _sqlite;
36.     public:
37.         MsgQueueMapper(const std::string &dbName)
38.             : _sqlite(dbName)
39.         {
40.             // 确保数据库文件已经存在，不存在就创建
41.             if (!ns_util::FileUtil::createFile(dbName))
42.             {
43.                 LOG(FATAL) << "create database " << dbName << " fail" << endl;
44.                 exit(1);
45.             }
47.             if (!_sqlite.open())
48.             {
49.                 LOG(FATAL) << "open database " << dbName << " fail" << endl;
50.                 exit(1);
51.             }
52.             createTable();
53.         }
55.         /*************
56.          * 插入消息队列
57.          * *************/
58.         bool insertMsgQueue(MsgQueuePtr msgQueuePtr)
59.         {
60.             char insertSql[1024];
61.             sprintf(insertSql, "insert into msg_queue_table values('%s', '%d');",
62.                     msgQueuePtr->_name.c_str(), msgQueuePtr->_isDurable);
63.             if (!_sqlite.exec(insertSql, nullptr, nullptr))
64.             {
65.                 LOG(WARNING) << "insert MsgQueue fail, MsgQueue: " << msgQueuePtr->_name << endl;
66.                 return false;
67.             }
68.             return true;
69.         }
71.         /**********
72.          * 移除消息队列
73.          * ***************/
74.         void removeMsgQueue(const std::string &name)
75.         {
76.             char deleteSql[1024];
77.             sprintf(deleteSql, "delete from msg_queue_table where name='%s';", name.c_str());
78.             if (!_sqlite.exec(deleteSql, nullptr, nullptr))
79.             {
80.                 LOG(WARNING) << "remove MsgQueue fail, MsgQueue: " << name << endl;
81.             }
82.         }
84.         /***********
85.          * 从数据库中恢复消息队列到内存
86.          * *****************/
87.         void recoverMsgQueue(std::unordered_map<std::string, MsgQueuePtr> *mapPtr)
88.         {
89.             const std::string selectSql = "select * from msg_queue_table;";
90.             if (!_sqlite.exec(selectSql.c_str(), selectCallback, mapPtr))
91.             {
92.                 LOG(FATAL) << "recover MsgQueue from msg_queue_table fail" << endl;
93.                 exit(1);
94.             }
95.         }
97.         /**************
98.          * 删除数据库表(仅调试)
99.          * ***************/
100.         void removeTable()
101.         {
102.             const std::string dropSql = "drop table if exists msg_queue_table;";
103.             if (_sqlite.exec(dropSql.c_str(), nullptr, nullptr))
104.             {
105.                 LOG(WARNING) << "remove table msg_queue_table fail" << endl;
106.             }
107.         }
109.     private:
110.         void createTable()
111.         {
112.             const std::string createSql = "create table if not exists msg_queue_table(\
113.                 name varchar(32) primary key,\
114.                 durable int\
115.             );";
117.             if (!_sqlite.exec(createSql.c_str(), nullptr, nullptr))
118.             {
119.                 LOG(FATAL) << "create table msg_queue_table fail" << endl;
120.                 exit(1);
121.             }
122.         }
124.         static int selectCallback(void *arg, int colNum, char **line, char **fields)
125.         {
126.             auto mapPtr = static_cast<std::unordered_map<std::string, MsgQueuePtr> *>(arg);
127.             std::string name = line[0];
128.             bool isDurable = std::stoi(line[1]);
129.             auto msgQueuePtr = std::make_shared<MsgQueue>(name, isDurable);
131.             mapPtr->insert({name, msgQueuePtr});
132.             return 0;
133.         }
134.     };
136.     class MsgQueueManager
137.     {
138.     private:
139.         MsgQueueMapper _mapper;
140.         std::unordered_map<std::string, MsgQueuePtr> _msgQueues;
141.         std::mutex _mtx;
143.     public:
144.         MsgQueueManager(const std::string &dbName)
145.             : _mapper(dbName)
146.         {
147.             _mapper.recoverMsgQueue(&_msgQueues);
148.         }
150.         /***********
151.          * 声明队列
152.          * ************/
153.         bool declareMsgQueue(const std::string &name, bool isDurable)
154.         {
155.             std::unique_lock<std::mutex> lck(_mtx);
157.             if (_msgQueues.count(name))
158.             {
159.                 return true;
160.             }
162.             auto msgQueuePtr = std::make_shared<MsgQueue>(name, isDurable);
163.             _msgQueues[name] = msgQueuePtr;
165.             if (isDurable)
166.             {
167.                 return _mapper.insertMsgQueue(msgQueuePtr);
168.             }
169.             return true;
170.         }
172.         /**********
173.          * 移除队列
174.          * ***********/
175.         void removeMsgQueue(const std::string &name)
176.         {
177.             std::unique_lock<std::mutex> lck(_mtx);
178.             
179.             auto it = _msgQueues.find(name);
180.             if (it == _msgQueues.end())
181.             {
182.                 return;
183.             }
184.             if (it->second->_isDurable)
185.             {
186.                 _mapper.removeMsgQueue(name);
187.             }
189.             _msgQueues.erase(name);
190.         }
192.         /************
193.          * 获取指定队列
194.          * ***************/
195.         MsgQueuePtr getMsgQueue(const std::string &name)
196.         {
197.             std::unique_lock<std::mutex> lck(_mtx);
198.             
199.             if (_msgQueues.count(name) == 0)
200.             {
201.                 return nullptr;
202.             }
203.             return _msgQueues[name];
204.         }
206.         /*************
207.          * 清理所有队列(仅调试)
208.          * ******************/
209.         void clearMsgQueues()
210.         {
211.             std::unique_lock<std::mutex> lck(_mtx);
213.             _msgQueues.clear();
214.             _mapper.removeTable();
215.         }
216.     };
218. }

复制代码

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