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

🕗 发布于 2024-11-11 12:14 c++ ConcurrentQueue RocketMQ 信息与通信

ConcurrentQueue开源库介绍

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

ConcurrentQueue使用

0x01 使用场景说明

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

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

设计BusinessFlowMsg

该类有定义消息的类型
该类中设计ST_BusinessSign结构体消息头，用来区分消息和获取消息体的长度
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;
}

在这里插入图片描述

原文地址：https://blog.csdn.net/Love_XiaoQinEr/article/details/143665928

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：简易入手《SOM神经网络》的本质与原理
下一篇：【.NET 8 实战--孢子记账--从单体到微服务】--简易权限--访问权限中间件

如何构建安全可靠的 HarmonyOS 应用
本文将深入探讨 HarmonyOS App 的安全编码规范与最佳实践，帮助开发者在代码编写中避免常见的安全漏洞，如 SQL 注入、XSS攻击等。我们将提供具体的编码示例，并结合ArkUI和ArkTS实
阅读更多2024-11-16
js像循环数组那样循环一个数字，Array.from()
js像循环数组那样循环一个数字，Array.from()
阅读更多2024-11-16
【C++笔记】vector使用详解及模拟实现
vector的文档使用STL的三个境界：能用、明理、能扩展，下面学习vector，我们也按照这个境界去学习。vector是可以改变大小的数组序列容器，也就是数据结构的顺序表。构造函数声明接口说明vec
阅读更多2024-11-16
Java线程池：ThreadPoolExecutor原理解析
本文介绍了线程池的基本概念、主要参数、工作流程，以及 execute() 方法的源码分析，此外，还讨论了在实际应用中可能遇到的陷阱和问题。
阅读更多2024-11-16
1.两数之和-力扣（LeetCode）
1.两数之和-力扣（LeetCode）
阅读更多2024-11-16
Xss挑战（跨脚本攻击）
这里将script，on，src，data，href，进行了过滤，并且在尝试的时候关键字双写不能用了，那么这里直接选择不去闭合标签，直接使用伪协议，但是发现javascript也被拆开了，这里可以对伪
阅读更多2024-11-16
《Python 网络爬虫》
本文介绍了 Python 网络爬虫的基本概念、技术原理、常用工具以及实战案例，希望能够帮助读者快速掌握 Python 网络爬虫技术。在实际应用中，需要根据具体的需求选择合适的工具和方法，并注意遵守法律
阅读更多2024-11-16
UEFI学习（五）——启动框架
https://www.zhihu.com/question/36313402/answer/2398532123UEFI（统一可扩展固件接口）在启动过程主要有以下几个阶段：
阅读更多2024-11-16
【洛谷】T539820 202411A Giants
C++ LGR-207-Div.4】洛谷入门赛 #29 第一题
阅读更多2024-11-16
Python 正则表达式进阶用法：量词与范围
匹配前面的字符零次或多次，相当于“任意多次”。：匹配前面的字符一次或多次，相当于“至少一次”。?：匹配前面的字符零次或一次，相当于“可有可无”。{n}：匹配前面的字符n次。{n,}：匹配前面的字符至少
阅读更多2024-11-16