C++使用开源ConcurrentQueue库处理自定义业务数据类

ConcurrentQueue开源库介绍

ConcurrentQueue是一个高性能的、线程安全的并发队列库。它旨在提供高效、无锁的数据布局，适用于多线程情况中的数据交换。concurrentqueue 支持多个生产者和多个消耗者，而且提供了多种配置选项来优化性能和内存使用。
ConcurrentQueue使用

0x01 使用场景说明

我的数据平台在吸收到四种不同的业务数据时，需要按数据分类写进RocketMQ。
0x02 自定义类用来存放和区分数据流

• 设计BusinessFlowMsg
  

• 该类有定义消息的范例
  
• 该类中设计ST_BusinessSign布局体消息头，用来区分消息和获取消息体的长度
  
• BusinessFlowMsg类可以存放的数据长度为512KB
  

1. #ifndef BUSINESSFLOWMSG_HPP
2. #define BUSINESSFLOWMSG_HPP
4. #include <string.h>
6. #define MSG_ROCKETMQ_PNG 0x01
7. #define MSG_ROCKETMQ_AIS 0x02
8. #define MSG_ROCKETMQ_ROUTE 0x03
9. #define MSG_ROCKETMQ_VOYAGE 0x04
11. #pragma pack(push)
12. #pragma pack(1)
14. // 消息头
15. typedef struct s_BusinessSign
16. {
17.     int sign; // 业务标识
18.     unsigned int length; // 消息体的长度
19. }ST_BusinessSign;
21. #pragma pack(pop)
23. class BusinessFlowMsg
24. {
25. public:
26.     BusinessFlowMsg() = default;
27.     ~BusinessFlowMsg() = default;
29.     char* get_data()
30.     {
31.         return _data;
32.     }
34.     int data_size()
35.     {
36.         return businessSign.length;
37.     }
39.     char* get_body()
40.     {
41.         return _data + sizeof(ST_BusinessSign);
42.     }
44.     int body_size()
45.     {
46.         return data_size() - sizeof(ST_BusinessSign);
47.     }
49.     ST_BusinessSign* header()
50.     {
51.         return &businessSign;
52.     }
54.     bool set_data(const char* data, int length, int sign)
55.     {
56.         if(length > (max_body_len + sizeof(ST_BusinessSign)))
57.         {
58.             return false;
59.         }
61.         businessSign.sign = sign;
62.         businessSign.length = length;
63.         memcpy(_data + sizeof(ST_BusinessSign), data, length);
65.         return true;
66.     }
68. private:
69.     enum
70.     {
71.         max_body_len = 512 * 1024 // 512KB
72.     };
74.     ST_BusinessSign businessSign;
75.     char _data[max_body_len];
76. };
78. #endif // BUSINESSFLOWMSG_HPP

复制代码

0x03 创建PngUnit类模仿接到不同的业务数据

• PngUnit范例创建了四个线程来模仿不同的数据流。
  
• PngUnit类多线程中并未使用互斥锁，因为ConcurrentQueue是一个线程安全的并发队列库，事实证实确实如此。
  

1. #ifndef PNGUNIT_H
2. #define PNGUNIT_H
4. #include <thread>
5. #include <mutex>
7. class PngUnit
8. {
9. public:
10.     PngUnit();
11.     ~PngUnit() = default;
12.     void start();
13.     void sendPNG(int sign);
14.     void sendAIS(int sign);
15.     void sendRoute(int sign);
16.     void sendVoyage(int sign);
18. private:
19.     std::thread m_th_png;
20.     std::thread m_th_ais;
21.     std::thread m_th_route;
22.     std::thread m_th_voyage;
24.     // std::mutex queue_mutex;  // 互斥锁
25. };
27. #endif // PNGUNIT_H

复制代码

---

1. #include "pngunit.h"
2. #include <unistd.h>
3. #include "rocketmqutils.h"
4. #include "BusinessFlowMsg.hpp"
5. #include "json11/json11.hpp"
7. PngUnit::PngUnit()
8. {
10. }
12. void PngUnit::start()
13. {
14.     // m_th = std::thread([this](){
15.     //     sendPNG(100);
16.     // });
18.     if(m_th_png.joinable())
19.     {
20.         printf("[%s:%d] %s\n", __FILE__, __LINE__, "m_th_png is running");
21.         return;
22.     }
24.     m_th_png = std::thread(std::bind(&PngUnit::sendPNG, this, MSG_ROCKETMQ_PNG));
25.     m_th_ais= std::thread(std::bind(&PngUnit::sendAIS, this, MSG_ROCKETMQ_AIS));
26.     m_th_route= std::thread(std::bind(&PngUnit::sendRoute, this, MSG_ROCKETMQ_ROUTE));
27.     m_th_voyage= std::thread(std::bind(&PngUnit::sendVoyage, this, MSG_ROCKETMQ_VOYAGE));
28. }
30. void PngUnit::sendPNG(int sign)
31. {
32.     while (true)
33.     {
34.         BusinessFlowMsg pngMsg;
35.         const char* pngFile = "1234567890ABCDEF";
36.         int fileLen = strlen(pngFile) + 1;
38.         pngMsg.set_data(pngFile, fileLen, sign);
40.         {
41.             // std::lock_guard<std::mutex> lock(queue_mutex);
42.             if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(pngMsg))
43.             {
44.                 printf("Failed to set PNG message data");
45.             }
46.         }
48.         sleep(1);
49.     }
50. }
52. void PngUnit::sendAIS(int sign)
53. {
54.     json11::Json::object obj = {
55.         {"message", "AIS"},
56.         {"response", "success"}
57.     };
58.     std::string jsonStr = json11::Json(obj).dump();
60.     while (true)
61.     {
62.         BusinessFlowMsg aisMsg;
63.         int jsonStrLen = jsonStr.size();
64.         aisMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);
66.         {
67.             // std::lock_guard<std::mutex> lock(queue_mutex);
68.             if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(aisMsg))
69.             {
70.                 printf("Failed to set AIS message data");
71.             }
72.         }
74.         sleep(2);
75.     }
76. }
78. void PngUnit::sendRoute(int sign)
79. {
80.     json11::Json::object obj = {
81.         {"message", "Route"},
82.         {"response", "success"}
83.     };
84.     std::string jsonStr = json11::Json(obj).dump();
86.     while (true)
87.     {
88.         BusinessFlowMsg routeMsg;
89.         int jsonStrLen = jsonStr.size();
90.         routeMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);
92.         {
93.             // std::lock_guard<std::mutex> lock(queue_mutex);
94.             if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(routeMsg))
95.             {
96.                 printf("Failed to set ROUTE message data");
97.             }
98.         }
100.         sleep(3);
101.     }
102. }
104. void PngUnit::sendVoyage(int sign)
105. {
106.     json11::Json::object obj = {
107.         {"message", "Voyage"},
108.         {"response", "success"}
109.     };
110.     std::string jsonStr = json11::Json(obj).dump();
112.     while (true)
113.     {
114.         BusinessFlowMsg voyageMsg;
115.         int jsonStrLen = jsonStr.size();
116.         voyageMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);
118.         {
119.             // std::lock_guard<std::mutex> lock(queue_mutex);
120.             if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(voyageMsg))
121.             {
122.                 printf("Failed to set VOYAGE message data");
123.             }
124.         }
126.         sleep(4);
127.     }
128. }

复制代码

0x04 创建RocketMQUtils类，在ConcurrentQueue队列中获取数据写进RocketMQ

• RocketMQUtils类是一个单例类
  

1. #ifndef ROCKETMQUTILS_H
2. #define ROCKETMQUTILS_H
4. #include <thread>
5. #include <concurrentqueue/moodycamel/concurrentqueue.h>
6. #include "BusinessFlowMsg.hpp"
8. class RocketMQUtils
9. {
10. public:
11.     static RocketMQUtils* Instance();
13. private:
14.     RocketMQUtils();
15.     ~RocketMQUtils()=default;
16.     RocketMQUtils(const RocketMQUtils &) = delete;
17.     RocketMQUtils& operator=(const RocketMQUtils &) = delete;
19.     RocketMQUtils(RocketMQUtils &&) = delete;
20.     RocketMQUtils& operator=(RocketMQUtils &&) = delete;
22. public:
23.     void start();
24.     void push();
25.     void poll();
26.     bool write(char *data, int len, int sign);
28. public:
29.     moodycamel::ConcurrentQueue<BusinessFlowMsg> g_businessQueue;
31. private:
32.     static RocketMQUtils* _instance;
34.     std::thread _pushThread;
35. };
37. #endif // ROCKETMQUTILS_H

复制代码

---

1. #include "rocketmqutils.h"
2. #include <unistd.h>
4. RocketMQUtils * RocketMQUtils::_instance = nullptr;
6. RocketMQUtils *RocketMQUtils::Instance()
7. {
8.     if(_instance == nullptr)
9.     {
10.         _instance = new RocketMQUtils();
11.     }
12.     return _instance;
13. }
15. RocketMQUtils::RocketMQUtils()
16. {
18. }
20. void RocketMQUtils::start()
21. {
22.     if(_pushThread.joinable())
23.     {
24.         return;
25.     }
26.     _pushThread = std::thread(&RocketMQUtils::push, this);
27. }
29. void RocketMQUtils::push()
30. {
31.     while (true)
32.     {
33.         BusinessFlowMsg busiMsg;
35.         if(g_businessQueue.try_dequeue(busiMsg))
36.         {
37.             write(busiMsg.get_body(), busiMsg.header()->length, busiMsg.header()->sign);
38.         }
39.         else
40.         {
41.             printf("[%s:%d] %s\n", __FILE__, __LINE__, "g_businessQueue is empty");
42.             sleep(2);
43.         }
44.     }
45. }
47. void RocketMQUtils::poll()
48. {
50. }
52. bool RocketMQUtils::write(char *data, int len, int sign)
53. {
54.     std::string msg(data, len);
55.     if (sign == MSG_ROCKETMQ_PNG)
56.     {
57.         printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());
58.     }
59.     else if (sign == MSG_ROCKETMQ_AIS)
60.     {
61.         printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());
62.     }
63.     else if (sign == MSG_ROCKETMQ_ROUTE)
64.     {
65.         printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());
66.     }
67.     else if (sign == MSG_ROCKETMQ_VOYAGE)
68.     {
69.         printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());
70.     }
71.     else
72.     {
73.         printf("[%s:%d] data sign error\n", __FILE__, __LINE__);
74.     }
75. }

复制代码

0x05 使用演示

1. #include <iostream>
2. #include <memory>
3. #include "rocketmqutils.h"
4. #include "pngunit.h"
6. using namespace std;
8. int main()
9. {
10.     cout << "==Start==" << endl;
11.     RocketMQUtils* rocketmq = RocketMQUtils::Instance();
12.     rocketmq->start();
14.     std::shared_ptr<PngUnit> ptrPngUnit = std::make_shared<PngUnit>();
15.     ptrPngUnit->start();
17.     getchar();
19.     cout << "==Over==" << endl;
21.     return 0;
22. }

复制代码

---

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