当前位置: 首页 > news >正文

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

ConcurrentQueue开源库介绍

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

ConcurrentQueue使用

0x01 使用场景说明

我的数据平台在接收到四种不同的业务数据时,需要按数据分类写进RocketMQ。

0x02 自定义类用来存放和区分数据流

  • 设计BusinessFlowMsg
  1. 该类有定义消息的类型
  2. 该类中设计ST_BusinessSign结构体消息头,用来区分消息和获取消息体的长度
  3. BusinessFlowMsg类可以存放的数据长度为512KB
#ifndef BUSINESSFLOWMSG_HPP
#define BUSINESSFLOWMSG_HPP#include <string.h>#define MSG_ROCKETMQ_PNG 0x01
#define MSG_ROCKETMQ_AIS 0x02
#define MSG_ROCKETMQ_ROUTE 0x03
#define MSG_ROCKETMQ_VOYAGE 0x04#pragma pack(push)
#pragma pack(1)// 消息头
typedef struct s_BusinessSign
{int sign; // 业务标识unsigned int length; // 消息体的长度
}ST_BusinessSign;#pragma pack(pop)class BusinessFlowMsg
{
public:BusinessFlowMsg() = default;~BusinessFlowMsg() = default;char* get_data(){return _data;}int data_size(){return businessSign.length;}char* get_body(){return _data + sizeof(ST_BusinessSign);}int body_size(){return data_size() - sizeof(ST_BusinessSign);}ST_BusinessSign* header(){return &businessSign;}bool set_data(const char* data, int length, int sign){if(length > (max_body_len + sizeof(ST_BusinessSign))){return false;}businessSign.sign = sign;businessSign.length = length;memcpy(_data + sizeof(ST_BusinessSign), data, length);return true;}private:enum{max_body_len = 512 * 1024 // 512KB};ST_BusinessSign businessSign;char _data[max_body_len];
};#endif // BUSINESSFLOWMSG_HPP

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

  • PngUnit类型创建了四个线程来模拟不同的数据流。
  • PngUnit类多线程中并未使用互斥锁,因为ConcurrentQueue是一个线程安全的并发队列库,事实证明确实如此。
#ifndef PNGUNIT_H
#define PNGUNIT_H#include <thread>
#include <mutex>class PngUnit
{
public:PngUnit();~PngUnit() = default;void start();void sendPNG(int sign);void sendAIS(int sign);void sendRoute(int sign);void sendVoyage(int sign);private:std::thread m_th_png;std::thread m_th_ais;std::thread m_th_route;std::thread m_th_voyage;// std::mutex queue_mutex;  // 互斥锁
};#endif // PNGUNIT_H

#include "pngunit.h"
#include <unistd.h>
#include "rocketmqutils.h"
#include "BusinessFlowMsg.hpp"
#include "json11/json11.hpp"PngUnit::PngUnit()
{}void PngUnit::start()
{// m_th = std::thread([this](){//     sendPNG(100);// });if(m_th_png.joinable()){printf("[%s:%d] %s\n", __FILE__, __LINE__, "m_th_png is running");return;}m_th_png = std::thread(std::bind(&PngUnit::sendPNG, this, MSG_ROCKETMQ_PNG));m_th_ais= std::thread(std::bind(&PngUnit::sendAIS, this, MSG_ROCKETMQ_AIS));m_th_route= std::thread(std::bind(&PngUnit::sendRoute, this, MSG_ROCKETMQ_ROUTE));m_th_voyage= std::thread(std::bind(&PngUnit::sendVoyage, this, MSG_ROCKETMQ_VOYAGE));
}void PngUnit::sendPNG(int sign)
{while (true){BusinessFlowMsg pngMsg;const char* pngFile = "1234567890ABCDEF";int fileLen = strlen(pngFile) + 1;pngMsg.set_data(pngFile, fileLen, sign);{// std::lock_guard<std::mutex> lock(queue_mutex);if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(pngMsg)){printf("Failed to set PNG message data");}}sleep(1);}
}void PngUnit::sendAIS(int sign)
{json11::Json::object obj = {{"message", "AIS"},{"response", "success"}};std::string jsonStr = json11::Json(obj).dump();while (true){BusinessFlowMsg aisMsg;int jsonStrLen = jsonStr.size();aisMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);{// std::lock_guard<std::mutex> lock(queue_mutex);if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(aisMsg)){printf("Failed to set AIS message data");}}sleep(2);}
}void PngUnit::sendRoute(int sign)
{json11::Json::object obj = {{"message", "Route"},{"response", "success"}};std::string jsonStr = json11::Json(obj).dump();while (true){BusinessFlowMsg routeMsg;int jsonStrLen = jsonStr.size();routeMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);{// std::lock_guard<std::mutex> lock(queue_mutex);if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(routeMsg)){printf("Failed to set ROUTE message data");}}sleep(3);}
}void PngUnit::sendVoyage(int sign)
{json11::Json::object obj = {{"message", "Voyage"},{"response", "success"}};std::string jsonStr = json11::Json(obj).dump();while (true){BusinessFlowMsg voyageMsg;int jsonStrLen = jsonStr.size();voyageMsg.set_data(jsonStr.c_str(), jsonStrLen, sign);{// std::lock_guard<std::mutex> lock(queue_mutex);if(!RocketMQUtils::Instance()->g_businessQueue.enqueue(voyageMsg)){printf("Failed to set VOYAGE message data");}}sleep(4);}
}

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

  • RocketMQUtils类是一个单例类
#ifndef ROCKETMQUTILS_H
#define ROCKETMQUTILS_H#include <thread>
#include <concurrentqueue/moodycamel/concurrentqueue.h>
#include "BusinessFlowMsg.hpp"class RocketMQUtils
{
public:static RocketMQUtils* Instance();private:RocketMQUtils();~RocketMQUtils()=default;RocketMQUtils(const RocketMQUtils &) = delete;RocketMQUtils& operator=(const RocketMQUtils &) = delete;RocketMQUtils(RocketMQUtils &&) = delete;RocketMQUtils& operator=(RocketMQUtils &&) = delete;public:void start();void push();void poll();bool write(char *data, int len, int sign);public:moodycamel::ConcurrentQueue<BusinessFlowMsg> g_businessQueue;private:static RocketMQUtils* _instance;std::thread _pushThread;
};#endif // ROCKETMQUTILS_H

#include "rocketmqutils.h"
#include <unistd.h>RocketMQUtils * RocketMQUtils::_instance = nullptr;RocketMQUtils *RocketMQUtils::Instance()
{if(_instance == nullptr){_instance = new RocketMQUtils();}return _instance;
}RocketMQUtils::RocketMQUtils()
{}void RocketMQUtils::start()
{if(_pushThread.joinable()){return;}_pushThread = std::thread(&RocketMQUtils::push, this);
}void RocketMQUtils::push()
{while (true){BusinessFlowMsg busiMsg;if(g_businessQueue.try_dequeue(busiMsg)){write(busiMsg.get_body(), busiMsg.header()->length, busiMsg.header()->sign);}else{printf("[%s:%d] %s\n", __FILE__, __LINE__, "g_businessQueue is empty");sleep(2);}}
}void RocketMQUtils::poll()
{}bool RocketMQUtils::write(char *data, int len, int sign)
{std::string msg(data, len);if (sign == MSG_ROCKETMQ_PNG){printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());}else if (sign == MSG_ROCKETMQ_AIS){printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());}else if (sign == MSG_ROCKETMQ_ROUTE){printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());}else if (sign == MSG_ROCKETMQ_VOYAGE){printf("[%s:%d] [%d] %s\n", __FILE__, __LINE__, sign, msg.c_str());}else{printf("[%s:%d] data sign error\n", __FILE__, __LINE__);}
}

0x05 使用演示

#include <iostream>
#include <memory>
#include "rocketmqutils.h"
#include "pngunit.h"using namespace std;int main()
{cout << "==Start==" << endl;RocketMQUtils* rocketmq = RocketMQUtils::Instance();rocketmq->start();std::shared_ptr<PngUnit> ptrPngUnit = std::make_shared<PngUnit>();ptrPngUnit->start();getchar();cout << "==Over==" << endl;return 0;
}

在这里插入图片描述

http://www.lryc.cn/news/481121.html

相关文章:

  • 在vue3的vite网络请求报错 [vite] http proxy error:
  • ElasticSearch 简单的查询。查询存在该字段的资源,更新,统计
  • FOFA使用教程之从零到精通
  • 【提高篇】3.2 GPIO(二,基本结构)
  • UE hard/soft reference| DDX DDY | Unity pcg color
  • macOS 应用公证指南:使用 fastlane 实现自动化公证流程
  • 深度学习:解密图像、音频和视频数据的“理解”之道20241105
  • uniapp 实现瀑布流
  • 计算机毕业设计 | springboot+vue智慧工地管理系统 前后端分离后台管理(附源码+文档)
  • vue中html如何转成pdf下载,pdf转base64,忽略某个元素渲染在pdf中,方法封装
  • Ubuntu下如何管理多个ssh密钥
  • [vulnhub] DarkHole: 1
  • 商淘云连锁企业管理五大功能 收银系统助力门店进销存同步
  • 统信UOS开发环境支持Perl
  • Stable Diffusion Web UI - ControlNet 姿势控制 openpose
  • java中Json字符串转换
  • springboot处理跨域请求
  • S32G-VNP-RDB2开发环境搭建
  • 分布式唯一ID生成(二): leaf
  • 【开发工具】Git
  • 【go从零单排】结构嵌套struct embedding
  • Django 详细入门介绍
  • 万字长文解读深度学习——循环神经网络RNN、LSTM、GRU、Bi-RNN
  • HDR视频技术之二:光电转换与 HDR 图像显示
  • 【Linux】Linux入门实操——vim、目录结构、远程登录、重启注销
  • Redis的缓存问题与应对策略
  • Java项目实战II基于Spring Boot的智慧生活商城系统的设计与实现(开发文档+数据库+源码)
  • 每日一题之成绩排序
  • QT Widget:使用技巧
  • 深入Zookeeper节点操作:高级功能与最佳实践