1052 lines
36 KiB
C++
1052 lines
36 KiB
C++
#include <fstream> // 用于 std::ifstream
|
||
#include <sstream> // 用于 std::stringstream
|
||
#include <unistd.h>
|
||
#include <stdlib.h>
|
||
#include <iostream>
|
||
#include <atomic>
|
||
#include <string>
|
||
#include "../mms/db_interface.h"
|
||
|
||
#include "../rocketmq/CProducer.h"
|
||
#include "../rocketmq/CMessage.h"
|
||
#include "../rocketmq/CSendResult.h"
|
||
#include "../rocketmq/SimpleProducer.h"
|
||
|
||
//测试300数据用lnk20241202
|
||
#include <ctime>
|
||
#include <QThread>
|
||
//测试300数据用
|
||
//lnk20241209添加队列选择
|
||
#include "../rocketmq/MQSelector.h"
|
||
#include "../rocketmq/MQMessageQueue.h"
|
||
//#include <atomic>
|
||
#include <vector>
|
||
#include <stdexcept>
|
||
//lnk20241209添加队列选择
|
||
#include <cstring>
|
||
|
||
//引入消费起点
|
||
#include "../rocketmq/DefaultMQPushConsumer.h"
|
||
#include "../rocketmq/ConsumeType.h"
|
||
|
||
#include "../rocketmq/MQMessageListener.h"
|
||
#include "../rocketmq/MQMessageExt.h"
|
||
#include "../rocketmq/SessionCredentials.h"
|
||
|
||
// 引入提供的消费者接口头文件
|
||
//#include "../rocketmq/CPushConsumer.h"
|
||
//#include "../rocketmq/CCommon.h"
|
||
//#include "../rocketmq/CMessageExt.h"
|
||
#include <map>
|
||
#include <pthread.h> // 用于互斥锁（在 C++98 中没有 std::mutex）
|
||
#include <utility> // for std::pair
|
||
|
||
#include "../log4cplus/log4.h"//lnk添加log4
|
||
|
||
using namespace std;
|
||
|
||
static std::once_flag g_consumer_once;
|
||
static std::once_flag g_producer_once;
|
||
|
||
extern std::string G_ROCKETMQ_PRODUCER;//rocketmq producer
|
||
extern std::string G_ROCKETMQ_IPPORT;//rocketmq ip+port
|
||
|
||
extern std::string G_ROCKETMQ_TOPIC_TEST;//topie
|
||
extern std::string G_ROCKETMQ_TAG_TEST;//tag
|
||
extern std::string G_ROCKETMQ_KEY_TEST;//key
|
||
|
||
extern std::string G_MQCONSUMER_TOPIC_LOG;
|
||
extern std::string G_MQCONSUMER_TOPIC_SET;
|
||
extern std::string G_MQCONSUMER_TOPIC_RC;
|
||
extern std::string G_MQCONSUMER_TOPIC_UD;
|
||
extern std::string G_MQCONSUMER_TOPIC_RT;
|
||
|
||
extern std::string G_MQCONSUMER_TOPIC_TEST;
|
||
|
||
extern std::string FRONT_INST;
|
||
extern bool DEBUGOPEN;
|
||
|
||
#ifdef __cplusplus
|
||
extern "C" {
|
||
#endif
|
||
|
||
#ifdef __cplusplus
|
||
}
|
||
#endif
|
||
|
||
extern int QUEUENUM;
|
||
|
||
extern int g_front_seg_index;
|
||
extern char subdir[128];
|
||
////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||
|
||
//消费者连接秘钥
|
||
extern std::string G_MQCONSUMER_ACCESSKEY;
|
||
extern std::string G_MQCONSUMER_SECRETKEY;
|
||
extern std::string G_MQCONSUMER_CHANNEL;
|
||
|
||
// 前向声明 RocketMQConsumer 类
|
||
class RocketMQConsumer;
|
||
|
||
// 全局映射：CPushConsumer* -> RocketMQConsumer*
|
||
//std::map<CPushConsumer*, RocketMQConsumer*> g_consumerMap;//
|
||
//pthread_mutex_t g_consumerMapMutex = PTHREAD_MUTEX_INITIALIZER;
|
||
|
||
////////////////////////////////
|
||
int64_t G_APP_START_MS = []() -> int64_t {
|
||
using namespace std::chrono;
|
||
return duration_cast<milliseconds>(
|
||
system_clock::now().time_since_epoch()
|
||
).count();
|
||
}();
|
||
|
||
int64_t G_START_SKEW_MS = 1000;
|
||
|
||
bool should_process_after_start(const rocketmq::MQMessageExt& msg)
|
||
{
|
||
const int64_t born_ts = static_cast<int64_t>(msg.getBornTimestamp());
|
||
return born_ts >= (G_APP_START_MS - G_START_SKEW_MS);
|
||
}
|
||
///////////////////////////////
|
||
class InternalListener;
|
||
|
||
class RocketMQConsumer {
|
||
public:
|
||
// 构造函数：初始化消费者并启动
|
||
RocketMQConsumer(const std::string& consumerName, const std::string& nameServer);
|
||
|
||
// 禁用拷贝和赋值
|
||
RocketMQConsumer(const RocketMQConsumer&) = delete;
|
||
RocketMQConsumer& operator=(const RocketMQConsumer&) = delete;
|
||
|
||
// 订阅主题和标签，并注册回调
|
||
void subscribe(const std::string& topic,
|
||
const std::string& tag,
|
||
MessageCallBack callback);
|
||
|
||
// 启动消费者
|
||
void start();
|
||
|
||
//修改消费模式
|
||
void setConsumerMessageModel(const std::string& topic);
|
||
|
||
rocketmq::ConsumeStatus handleMessage(const rocketmq::MQMessageExt& msg);
|
||
|
||
// 析构函数：关闭并销毁消费者
|
||
~RocketMQConsumer();
|
||
|
||
private:
|
||
//CPushConsumer* consumer_; // C 接口消费者指针
|
||
//MessageCallBack messageCallback_; // 函数指针用于回调
|
||
rocketmq::DefaultMQPushConsumer consumer_;
|
||
InternalListener* listener_;
|
||
|
||
std::map<std::pair<std::string, std::string>, MessageCallBack> callbacks_; // 订阅到回调的映射
|
||
|
||
/*// 静态消息处理回调
|
||
static int messageHandler(CPushConsumer* consumer, CMessageExt* msg);
|
||
// 实例消息处理函数
|
||
int handleMessage(CMessageExt* msg);*/
|
||
|
||
};
|
||
class InternalListener : public rocketmq::MessageListenerConcurrently {
|
||
public:
|
||
explicit InternalListener(RocketMQConsumer* owner)
|
||
: owner_(owner) {}
|
||
|
||
rocketmq::ConsumeStatus consumeMessage(
|
||
const std::vector<rocketmq::MQMessageExt>& msgs) override
|
||
{
|
||
for (size_t i = 0; i < msgs.size(); ++i) {
|
||
rocketmq::ConsumeStatus ret = owner_->handleMessage(msgs[i]);
|
||
if (ret != rocketmq::CONSUME_SUCCESS) {
|
||
return ret;
|
||
}
|
||
}
|
||
return rocketmq::CONSUME_SUCCESS;
|
||
}
|
||
|
||
private:
|
||
RocketMQConsumer* owner_;
|
||
};
|
||
// 构造函数实现C
|
||
/*RocketMQConsumer::RocketMQConsumer(const std::string& consumerName, const std::string& nameServer, const std::string& groupId)
|
||
: consumer_(NULL)//, messageCallback_(NULL)
|
||
{
|
||
// 创建消费者
|
||
consumer_ = CreatePushConsumer(consumerName.c_str());
|
||
if (consumer_ == NULL) {
|
||
std::cout << "error CreatePushConsumer"<< std::endl;
|
||
throw std::runtime_error("Failed to create push consumer.");
|
||
}
|
||
|
||
SetPushConsumerSessionCredentials(consumer_,G_MQCONSUMER_ACCESSKEY.c_str(),G_MQCONSUMER_SECRETKEY.c_str(),G_MQCONSUMER_CHANNEL.c_str());
|
||
|
||
// 设置 NameServer 地址
|
||
if (SetPushConsumerNameServerAddress(consumer_, nameServer.c_str()) != 0) {
|
||
DestroyPushConsumer(consumer_);
|
||
std::cout << "error setting nameServer"<< std::endl;
|
||
throw std::runtime_error("Failed to set NameServer address.");
|
||
}
|
||
|
||
// 设置消费者组ID
|
||
if (SetPushConsumerGroupID(consumer_, groupId.c_str()) != 0) {
|
||
DestroyPushConsumer(consumer_);
|
||
std::cout << "error setting groupId"<< std::endl;
|
||
throw std::runtime_error("Failed to set Consumer Group ID.");
|
||
}
|
||
|
||
//调试用
|
||
std::string consumerlog = "./mqconsumer/" + consumerName +".log";
|
||
if ( (SetPushConsumerLogPath(consumer_,consumerlog.c_str()) || SetPushConsumerLogFileNumAndSize(consumer_,10,100) || SetPushConsumerLogLevel(consumer_,E_LOG_LEVEL_DEBUG) ) != 0) {//记录消费日志
|
||
DestroyPushConsumer(consumer_);
|
||
std::cout << "error setting logpath"<< std::endl;
|
||
}
|
||
std::cout << "logpath:" << consumerlog << std::endl;
|
||
|
||
// 注册消息回调
|
||
if (RegisterMessageCallback(consumer_, RocketMQConsumer::messageHandler) != 0) {
|
||
DestroyPushConsumer(consumer_);
|
||
std::cout << "error setting Callback"<< std::endl;
|
||
throw std::runtime_error("Failed to register message callback.");
|
||
}
|
||
|
||
// 将消费者实例添加到全局映射
|
||
pthread_mutex_lock(&g_consumerMapMutex);
|
||
g_consumerMap[consumer_] = this;
|
||
pthread_mutex_unlock(&g_consumerMapMutex);
|
||
|
||
std::cout << "RocketMQ Consumer initialized and started." << std::endl;
|
||
}*/
|
||
|
||
//构造函数实现C++
|
||
RocketMQConsumer::RocketMQConsumer(const std::string& consumerGroup,
|
||
const std::string& nameServer)
|
||
: consumer_(consumerGroup),
|
||
listener_(NULL)
|
||
{
|
||
consumer_.setNamesrvAddr(nameServer);
|
||
|
||
consumer_.setSessionCredentials(
|
||
G_MQCONSUMER_ACCESSKEY,
|
||
G_MQCONSUMER_SECRETKEY,
|
||
G_MQCONSUMER_CHANNEL
|
||
);
|
||
|
||
// 限制消费线程池，防止 ConsumeTP 爆炸
|
||
consumer_.setConsumeThreadCount(4);
|
||
|
||
listener_ = new InternalListener(this);
|
||
}
|
||
|
||
// 启动消费者
|
||
void RocketMQConsumer::start()
|
||
{
|
||
static bool started = false;
|
||
if (started) {
|
||
std::cout << "Consumer already started" << std::endl;
|
||
return;
|
||
}
|
||
|
||
consumer_.registerMessageListener(listener_);
|
||
consumer_.start();
|
||
|
||
started = true;
|
||
}
|
||
|
||
void RocketMQConsumer::subscribe(const std::string& topic, const std::string& tag, MessageCallBack callback)
|
||
{
|
||
/*if (Subscribe(consumer_, topic.c_str(), tag.c_str()) != 0) {
|
||
throw std::runtime_error("Failed to subscribe to topic/tag.");
|
||
}*/
|
||
consumer_.subscribe(topic, tag);
|
||
|
||
//调试用
|
||
std::cout << "Subscribed to topic: " << topic << ", tag: " << tag << std::endl;
|
||
|
||
// 使用 std::pair 作为键
|
||
std::pair<std::string, std::string> mapKey(topic, tag);
|
||
callbacks_[mapKey] = callback;
|
||
}
|
||
/*
|
||
// 静态消息处理回调实现
|
||
int RocketMQConsumer::messageHandler(CPushConsumer* consumer, CMessageExt* msg)
|
||
{
|
||
RocketMQConsumer* instance = NULL;
|
||
|
||
//调试用
|
||
std::cout << "messagehandler" << std::endl;
|
||
|
||
// 查找对应的消费者实例
|
||
pthread_mutex_lock(&g_consumerMapMutex);
|
||
std::map<CPushConsumer*, RocketMQConsumer*>::iterator it = g_consumerMap.find(consumer);
|
||
if (it != g_consumerMap.end()) {
|
||
instance = it->second;
|
||
}
|
||
pthread_mutex_unlock(&g_consumerMapMutex);
|
||
|
||
if (instance) {
|
||
return instance->handleMessage(msg);
|
||
} else {
|
||
std::cerr << "Consumer instance not found for callback." << std::endl;
|
||
//return E_RECONSUME_LATER; // 默认返回重试状态
|
||
return rocketmq::RECONSUME_LATER;
|
||
}
|
||
}
|
||
*/
|
||
/*int RocketMQConsumer::handleMessage(CMessageExt* msg)
|
||
{
|
||
// 检查 msg 和 consumer_ 是否为 NULL
|
||
if (!msg || !consumer_) {
|
||
std::cerr << "Received null message or consumer." << std::endl;
|
||
//return E_RECONSUME_LATER;
|
||
return rocketmq::RECONSUME_LATER;
|
||
}*/
|
||
|
||
// 获取消息的主题和标签
|
||
//std::string topic = GetMessageTopic(msg); // 假设存在此函数
|
||
//std::string tag = GetMessageTags(msg); // 假设存在此函数
|
||
|
||
rocketmq::ConsumeStatus RocketMQConsumer::handleMessage( const rocketmq::MQMessageExt& msg) {
|
||
std::string tag = msg.getTags();
|
||
std::string topic = msg.getTopic();
|
||
// 打印调试信息
|
||
std::cout << "Handling message for topic: " << topic << ", tag: " << tag << std::endl;
|
||
// 使用 std::pair 作为键
|
||
std::pair<std::string, std::string> key(topic, tag);
|
||
// 查找对应的回调函数
|
||
std::map<std::pair<std::string, std::string>, MessageCallBack>::iterator it = callbacks_.find(key);
|
||
if (it != callbacks_.end())
|
||
{ // 调用对应的回调函数
|
||
//调试
|
||
std::cout << "callback Handling message " <<std::endl;
|
||
//return it->second(consumer_, msg);
|
||
return it->second(msg);
|
||
}
|
||
else {
|
||
//调试
|
||
std::cout << "there is no callback " <<std::endl;
|
||
// 如果没有找到对应的回调，执行默认处理
|
||
//const char* body = GetMessageBody(msg);
|
||
//const char* msgKey = GetMessageKeys(msg);
|
||
std::string body = msg.getBody();
|
||
std::string msgKey = msg.getKeys();
|
||
if (!body.empty()) {
|
||
std::cout << "Received message body: " << body << std::endl;
|
||
} else {
|
||
std::cout << "Received message with empty body." << std::endl;
|
||
} if (!msgKey.empty()) {
|
||
std::cout << "Message Key: " << msgKey << std::endl;
|
||
} else {
|
||
std::cout << "Message Key: N/A" << std::endl;
|
||
} //return E_CONSUME_SUCCESS;
|
||
return rocketmq::CONSUME_SUCCESS;
|
||
}
|
||
}
|
||
|
||
|
||
// 析构函数实现
|
||
RocketMQConsumer::~RocketMQConsumer()
|
||
{
|
||
/*if (consumer_) {
|
||
// 关闭消费者
|
||
ShutdownPushConsumer(consumer_);
|
||
|
||
// 从全局映射中移除
|
||
pthread_mutex_lock(&g_consumerMapMutex);
|
||
g_consumerMap.erase(consumer_);
|
||
pthread_mutex_unlock(&g_consumerMapMutex);
|
||
|
||
// 销毁消费者
|
||
DestroyPushConsumer(consumer_);
|
||
consumer_ = NULL;
|
||
|
||
std::cout << "RocketMQ Consumer shutdown and destroyed." << std::endl;
|
||
}*/
|
||
try {
|
||
consumer_.shutdown();
|
||
} catch (...) {
|
||
}
|
||
|
||
sleep(1); // 等内部线程退出
|
||
|
||
delete listener_;
|
||
listener_ = NULL;
|
||
|
||
std::cout << "RocketMQ Consumer shutdown and destroyed." << std::endl;
|
||
}
|
||
|
||
// 在 RocketMQConsumer 类中新增函数用来设置消费模式
|
||
void RocketMQConsumer::setConsumerMessageModel(const std::string& topic)
|
||
{
|
||
/*if (topic == G_MQCONSUMER_TOPIC_SET) {
|
||
// 设置为普通消费模式
|
||
if (SetPushConsumerMessageModel(consumer_, CLUSTERING) != 0) {
|
||
std::cout << "Error setting message model to CLUSTERING for topic: " << topic << std::endl;
|
||
} else {
|
||
std::cout << "Set consumer to CLUSTERING for topic: " << topic << std::endl;
|
||
}
|
||
} else*/ {
|
||
// 默认设置为广播消费模式
|
||
/*if (SetPushConsumerMessageModel(consumer_, BROADCASTING) != 0) {
|
||
std::cout << "Error setting message model to BROADCASTING for topic: " << topic << std::endl;
|
||
} else {
|
||
std::cout << "Set consumer to BROADCASTING for topic: " << topic << std::endl;
|
||
}*/
|
||
consumer_.setMessageModel(rocketmq::BROADCASTING);
|
||
std::cout << "Set consumer to BROADCASTING for topic: " << topic << std::endl;
|
||
}
|
||
}
|
||
|
||
// 全局消费者实例
|
||
RocketMQConsumer* g_consumer = NULL;
|
||
|
||
// 初始化消费者函数
|
||
void InitializeConsumer(
|
||
const std::string& consumerName,
|
||
const std::string& nameServer,
|
||
const std::vector<Subscription>& subscriptions) // 接收多个订阅
|
||
{
|
||
if (g_consumer == NULL) {
|
||
std::cout << "create new consumer!" << std::endl;
|
||
try {
|
||
//g_consumer = new RocketMQConsumer(consumerName, nameServer,consumerName);//用消费名作为消费组，不同进程（不同的消费者）同时消费topic的同一条消息
|
||
g_consumer = new RocketMQConsumer(consumerName, nameServer);
|
||
|
||
for (size_t i = 0; i < subscriptions.size(); ++i) {
|
||
g_consumer->setConsumerMessageModel(subscriptions[i].topic);//初始化时根据topic设置消费模式
|
||
g_consumer->subscribe(subscriptions[i].topic, subscriptions[i].tag,subscriptions[i].callback);
|
||
}
|
||
|
||
g_consumer->start();
|
||
|
||
}
|
||
catch (const std::exception& e) {
|
||
std::cerr << "Failed to initialize consumer: " << e.what() << std::endl;
|
||
throw; // 重新抛出异常
|
||
}
|
||
}
|
||
}
|
||
|
||
// 关闭并销毁消费者函数
|
||
void ShutdownAndDestroyConsumer()
|
||
{
|
||
if (g_consumer != NULL) {
|
||
delete g_consumer;
|
||
g_consumer = NULL;
|
||
}
|
||
}
|
||
|
||
// 消费消息的接口函数
|
||
void rocketmq_consumer_receive(
|
||
const std::string& consumerName,
|
||
const std::string& nameServer,
|
||
const std::vector<Subscription>& subscriptions) // 接收多个订阅
|
||
{
|
||
std::call_once(g_consumer_once, [&](){
|
||
try {
|
||
InitializeConsumer(consumerName, nameServer, subscriptions);
|
||
} catch (...) {
|
||
std::cerr << "Cannot consume message because consumer initialization failed." << std::endl;
|
||
}
|
||
});
|
||
|
||
}
|
||
/////////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||
//封装生产者类
|
||
#if 1
|
||
// 全局或静态变量，用于维护当前队列索引
|
||
static int currentQueueId = 0;
|
||
|
||
// 队列选择器回调函数：轮询选择队列 ID
|
||
/*int RoundRobinSelector(int queueNum, CMessage* msg, void* arg) {
|
||
if (queueNum == 0) {
|
||
throw std::runtime_error("No available queues");
|
||
}
|
||
int queueId = currentQueueId % queueNum;
|
||
currentQueueId++;
|
||
if (currentQueueId >= 1024 - queueNum) {
|
||
currentQueueId = 0;
|
||
}
|
||
return queueId;
|
||
}*/
|
||
int RoundRobinSelector(int queueNum, CMessage* msg, void* arg) {
|
||
static std::atomic<int> currentQueueId(0);
|
||
|
||
if (queueNum <= 0) {
|
||
std::cout << "[MQ][SELECTOR_FAIL] queueNum=" << queueNum << std::endl;
|
||
return -1;
|
||
}
|
||
|
||
int id = currentQueueId.fetch_add(1, std::memory_order_relaxed);
|
||
int queueId = id % queueNum;
|
||
|
||
std::cout << "[MQ][SELECTOR] queueNum=" << queueNum
|
||
<< ", current=" << id
|
||
<< ", selected=" << queueId
|
||
<< std::endl;
|
||
|
||
return queueId;
|
||
}
|
||
|
||
// 封装生产者的类
|
||
class RocketMQProducer {
|
||
public:
|
||
RocketMQProducer(const std::string& producerName, const std::string& nameServer)
|
||
: producer_(producerName)
|
||
{
|
||
// 创建生产者
|
||
/*producer_ = CreateProducer(producerName.c_str());
|
||
if (producer_ == NULL) {
|
||
throw std::runtime_error("Failed to create producer.");
|
||
}*/
|
||
|
||
// 设置日志
|
||
producer_.setLogLevel(rocketmq::eLOG_LEVEL_ERROR);
|
||
producer_.setLogFileSizeAndNum(5, 50);
|
||
|
||
// 设置 nameserver 地址
|
||
//SetProducerNameServerAddress(producer_, nameServer.c_str());
|
||
producer_.setNamesrvAddr(nameServer);
|
||
|
||
//lnk20260417设置数据上送消息体最大值，默认4M，调整为1M，避免过大消息导致发送失败
|
||
//SetProducerMaxMessageSize(producer_, 1024 * 1024); // 1MB
|
||
producer_.setMaxMessageSize(1024 * 1024);
|
||
|
||
//SetProducerSessionCredentials(producer_, G_MQCONSUMER_ACCESSKEY.c_str(),G_MQCONSUMER_SECRETKEY.c_str(), "");
|
||
producer_.setSessionCredentials(
|
||
G_MQCONSUMER_ACCESSKEY,
|
||
G_MQCONSUMER_SECRETKEY,
|
||
""
|
||
);
|
||
|
||
// 启动生产者
|
||
//StartProducer(producer_);
|
||
producer_.start();
|
||
|
||
std::cout << "rocketmq_Producer initialized and started." << std::endl;
|
||
}
|
||
|
||
// 禁用拷贝和赋值
|
||
RocketMQProducer(const RocketMQProducer&) = delete;
|
||
RocketMQProducer& operator=(const RocketMQProducer&) = delete;
|
||
|
||
/*void printSendResult(const CSendResult& result) {
|
||
std::cout << "SendResult:" << std::endl;
|
||
std::cout << " Status: ";
|
||
switch (result.sendStatus) {
|
||
case E_SEND_OK:
|
||
std::cout << "E_SEND_OK";
|
||
break;
|
||
case E_SEND_FLUSH_DISK_TIMEOUT:
|
||
std::cout << "E_SEND_FLUSH_DISK_TIMEOUT";
|
||
break;
|
||
case E_SEND_FLUSH_SLAVE_TIMEOUT:
|
||
std::cout << "E_SEND_FLUSH_SLAVE_TIMEOUT";
|
||
break;
|
||
case E_SEND_SLAVE_NOT_AVAILABLE:
|
||
std::cout << "E_SEND_SLAVE_NOT_AVAILABLE";
|
||
break;
|
||
default:
|
||
std::cout << "UNKNOWN(" << result.sendStatus << ")";
|
||
break;
|
||
}
|
||
std::cout << std::endl;
|
||
|
||
std::cout << " MsgID : " << result.msgId << std::endl;
|
||
std::cout << " Offset: " << result.offset << std::endl;
|
||
}*/
|
||
|
||
// 发送消息
|
||
/* void sendMessage(const char* strbody, const char* topic, const std::string& tags, const std::string& keys) {
|
||
|
||
if (DEBUGOPEN) {
|
||
std::cout << "sendMessage called with topic: " << (topic ? topic : "NULL")
|
||
<< ", tags: " << tags
|
||
<< ", keys: " << keys
|
||
<< std::endl;
|
||
|
||
if (strbody) {
|
||
// ===== 1️⃣ 真实长度 vs strlen =====
|
||
std::string body_str(strbody);
|
||
|
||
std::cout << "[MQ][LEN_CHECK]"
|
||
<< " strlen=" << strlen(strbody)
|
||
<< ", std::string.size=" << body_str.size()
|
||
<< std::endl;
|
||
|
||
// ===== 2️⃣ 检测是否包含 \0 =====
|
||
bool has_null = false;
|
||
for (size_t i = 0; i < body_str.size(); i++) {
|
||
if (body_str[i] == '\0') {
|
||
has_null = true;
|
||
std::cout << "[MQ][FOUND_NULL] index=" << i << std::endl;
|
||
break;
|
||
}
|
||
}
|
||
std::cout << "[MQ][HAS_NULL] " << (has_null ? "YES" : "NO") << std::endl;
|
||
|
||
// ===== 3️⃣ 打印头部（可读）=====
|
||
size_t len = strlen(strbody);
|
||
size_t n = std::min((size_t)200, len);
|
||
|
||
std::cout << "[MQ][BODY_HEAD] "
|
||
<< std::string(strbody, n)
|
||
<< std::endl;
|
||
|
||
std::cout << "[MQ][BODY_TAIL] "
|
||
<< std::string(strbody + (len - n), n)
|
||
<< std::endl;
|
||
|
||
// ===== 4️⃣ 十六进制打印前100字节 =====
|
||
std::cout << "[MQ][HEX_HEAD] ";
|
||
for (size_t i = 0; i < std::min((size_t)100, body_str.size()); i++) {
|
||
printf("%02X ", (unsigned char)body_str[i]);
|
||
}
|
||
printf("\n");
|
||
|
||
// ===== 5️⃣ 十六进制打印尾部100字节 =====
|
||
std::cout << "[MQ][HEX_TAIL] ";
|
||
size_t start = (body_str.size() > 100) ? body_str.size() - 100 : 0;
|
||
for (size_t i = start; i < body_str.size(); i++) {
|
||
printf("%02X ", (unsigned char)body_str[i]);
|
||
}
|
||
printf("\n");
|
||
} else {
|
||
std::cout << "[MQ][ERROR] strbody is NULL" << std::endl;
|
||
}
|
||
}
|
||
|
||
CSendResult result;
|
||
CMessage* msg = NULL;
|
||
|
||
try {
|
||
// 创建消息并设置属性
|
||
msg = CreateMessage(topic);
|
||
if (msg == NULL) {
|
||
throw std::runtime_error("Failed to create message.");
|
||
}
|
||
|
||
SetMessageTags(msg, tags.c_str());
|
||
SetMessageKeys(msg, keys.c_str());
|
||
SetMessageBody(msg, strbody);
|
||
|
||
// 假设队列数量和 Broker 名称是固定的
|
||
int queueNum = QUEUENUM; // 配置的队列数量，例如5
|
||
|
||
// 发送消息
|
||
int sendResult = SendMessageOnewayOrderly(
|
||
producer_,
|
||
msg,
|
||
RoundRobinSelector, // 队列选择器回调函数
|
||
&queueNum // 传递给选择器的额外参数（队列数量）
|
||
);
|
||
/////////////////////////////////替换接口，性能较低但不影响
|
||
/*CSendResult result;
|
||
memset(&result, 0, sizeof(result));
|
||
|
||
int sendResult = SendMessageOrderly(
|
||
producer_,
|
||
msg,
|
||
RoundRobinSelector,
|
||
&queueNum,
|
||
0, // autoRetryTimes
|
||
&result
|
||
);
|
||
|
||
std::cout << "[MQ][ORDERLY_RESULT]"
|
||
<< " ret=" << sendResult
|
||
<< ", sendStatus=" << (int)result.sendStatus
|
||
<< ", msgId=" << result.msgId
|
||
<< ", offset=" << result.offset
|
||
<< ", topic=" << (topic ? topic : "")
|
||
<< ", body_len=" << (strbody ? strlen(strbody) : 0)
|
||
<< std::endl;*/
|
||
/////////////////////////////////替换接口，性能较低但不影响
|
||
// 发送消息：临时改成同步发送，绕过 orderly / selector，便于定位问题
|
||
/*CSendResult result;
|
||
memset(&result, 0, sizeof(result));
|
||
|
||
int sendResult = SendMessageSync(
|
||
producer_,
|
||
msg,
|
||
&result
|
||
);
|
||
|
||
std::cout << "[MQ][SYNC_RESULT]"
|
||
<< " ret=" << sendResult
|
||
<< ", sendStatus=" << (int)result.sendStatus
|
||
<< ", msgId=" << result.msgId
|
||
<< ", offset=" << result.offset
|
||
<< ", topic=" << (topic ? topic : "")
|
||
<< ", body_len=" << (strbody ? strlen(strbody) : 0)
|
||
<< std::endl;*/
|
||
// 发送消息：临时改成同步发送，绕过 orderly / selector，便于定位问题
|
||
|
||
/*if (sendResult == 0) { // 假设返回 0 表示成功
|
||
std::cout << "[MQ][SEND_OK]"
|
||
<< " topic=" << (topic ? topic : "")
|
||
<< ", tags=" << tags
|
||
<< ", keys=" << keys
|
||
<< ", body_len=" << (strbody ? strlen(strbody) : 0)
|
||
<< std::endl;
|
||
} else {
|
||
|
||
std::cout << "[MQ][SEND_FAIL]"
|
||
<< " ret=" << sendResult
|
||
<< ", topic=" << (topic ? topic : "")
|
||
<< ", tags=" << tags
|
||
<< ", keys=" << keys
|
||
<< ", body_len=" << (strbody ? strlen(strbody) : 0)
|
||
<< std::endl;
|
||
std::cout << "[MQ][BODY_HEAD] " << std::string(strbody, std::min((size_t)200, strlen(strbody))) << std::endl;
|
||
std::cout << "[MQ][BODY_TAIL] " << std::string(strbody + std::max((size_t)0, strlen(strbody) - std::min((size_t)200, strlen(strbody)))) << std::endl;
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,"【ERROR】前置的%s%d号进程 mq发送失败,请检查mq配置", get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
|
||
// 销毁消息
|
||
DestroyMessage(msg);
|
||
msg = NULL; // 防止重复销毁
|
||
}
|
||
catch (const std::runtime_error& e) {
|
||
std::cerr << "Runtime error during message sending: " << e.what() << std::endl;
|
||
// 如果消息已经创建，确保销毁它
|
||
if (msg != NULL) {
|
||
DestroyMessage(msg);
|
||
}
|
||
// 可以在这里添加更多的错误处理逻辑，比如重试、记录日志等
|
||
}
|
||
catch (const std::exception& e) {
|
||
std::cerr << "Exception during message sending: " << e.what() << std::endl;
|
||
if (msg != NULL) {
|
||
DestroyMessage(msg);
|
||
}
|
||
}
|
||
catch (...) {
|
||
std::cerr << "Unknown error during message sending." << std::endl;
|
||
if (msg != NULL) {
|
||
DestroyMessage(msg);
|
||
}
|
||
}
|
||
}*/
|
||
void sendMessage(const std::string& body,
|
||
const std::string& topic,
|
||
const std::string& tags,
|
||
const std::string& keys)
|
||
{
|
||
try {
|
||
if (DEBUGOPEN) {
|
||
std::cout << "sendMessage called with topic: " << topic
|
||
<< ", tags: " << tags
|
||
<< ", keys: " << keys
|
||
<< ", body_len=" << body.size()
|
||
<< std::endl;
|
||
|
||
size_t n = std::min((size_t)200, body.size());
|
||
|
||
std::cout << "[MQ][BODY_HEAD] "
|
||
<< body.substr(0, n)
|
||
<< std::endl;
|
||
|
||
if (body.size() > n) {
|
||
std::cout << "[MQ][BODY_TAIL] "
|
||
<< body.substr(body.size() - n, n)
|
||
<< std::endl;
|
||
}
|
||
|
||
std::cout << "[MQ][HEX_HEAD] ";
|
||
for (size_t i = 0; i < std::min((size_t)100, body.size()); ++i) {
|
||
printf("%02X ", (unsigned char)body[i]);
|
||
}
|
||
printf("\n");
|
||
}
|
||
|
||
rocketmq::MQMessage msg(topic, tags, keys, body);
|
||
|
||
rocketmq::SendResult result = producer_.send(msg);
|
||
|
||
std::cout << "[MQ][SEND_OK]"
|
||
<< " topic=" << topic
|
||
<< ", tags=" << tags
|
||
<< ", keys=" << keys
|
||
<< ", msgId=" << result.getMsgId()
|
||
<< ", status=" << result.getSendStatus()
|
||
<< ", body_len=" << body.size()
|
||
<< std::endl;
|
||
}
|
||
catch (const rocketmq::MQClientException& e) {
|
||
std::cerr << "[MQ][SEND_FAIL] MQClientException: "
|
||
<< e.what() << std::endl;
|
||
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 mq发送失败,mq客户端错误,请检查mq配置",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
catch (const std::exception& e) {
|
||
std::cerr << "[MQ][SEND_FAIL] exception: "
|
||
<< e.what() << std::endl;
|
||
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 mq发送失败,mq发送错误,发送请检查mq配置",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
catch (...) {
|
||
std::cerr << "[MQ][SEND_FAIL] unknown exception" << std::endl;
|
||
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 mq发送失败,未知错误,请检查mq配置",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
}
|
||
|
||
|
||
// 析构函数中关闭并销毁生产者
|
||
~RocketMQProducer() {
|
||
/*if (producer_) {
|
||
ShutdownProducer(producer_);
|
||
DestroyProducer(producer_);
|
||
std::cout << "rocketmq_Producer shutdown and destroyed." << std::endl;
|
||
}*/
|
||
try {
|
||
producer_.shutdown();
|
||
}
|
||
catch (...) {
|
||
}
|
||
|
||
std::cout << "rocketmq_Producer shutdown and destroyed." << std::endl;
|
||
}
|
||
|
||
private:
|
||
//CProducer* producer_;
|
||
rocketmq::DefaultMQProducer producer_;
|
||
};
|
||
|
||
// 全局生产者实例
|
||
RocketMQProducer* g_producer = NULL;
|
||
|
||
// 初始化生产者（在程序启动时调用一次）
|
||
void InitializeProducer()
|
||
{
|
||
if (g_producer == NULL) {
|
||
try {
|
||
g_producer = new RocketMQProducer(G_ROCKETMQ_PRODUCER, G_ROCKETMQ_IPPORT);//生产者名称和NameServer地址
|
||
}
|
||
catch (const std::exception& e) {
|
||
std::cerr << "Failed to initialize producer: " << e.what() << std::endl;
|
||
// 根据需求决定是否终止程序或采取其他措施
|
||
throw; // 重新抛出异常
|
||
}
|
||
}
|
||
}
|
||
|
||
// 关闭并销毁生产者（在程序结束时调用一次）
|
||
void ShutdownAndDestroyProducer()
|
||
{
|
||
if (g_producer != NULL) {
|
||
delete g_producer;
|
||
g_producer = NULL;
|
||
}
|
||
}
|
||
|
||
// 发送消息的接口函数
|
||
void rocketmq_producer_send(const std::string& body,
|
||
const std::string& topic,
|
||
const std::string& tags,
|
||
const std::string& keys)
|
||
{
|
||
std::call_once(g_producer_once, [&](){
|
||
InitializeProducer();
|
||
});
|
||
|
||
try {
|
||
g_producer->sendMessage(body, topic, tags, keys);
|
||
} catch (const std::exception& e) {
|
||
std::cerr << "Failed to send message: " << e.what() << std::endl;
|
||
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ,
|
||
"【ERROR】前置的%s%d号进程 mq发送失败,请检查mq配置",
|
||
get_front_msg_from_subdir(), g_front_seg_index);
|
||
}
|
||
}
|
||
#endif
|
||
|
||
//////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||
/*
|
||
// producer_send0测试用
|
||
void StartSendMessage(CProducer* producer)
|
||
{
|
||
CSendResult result;
|
||
|
||
// create message and set some values for it
|
||
CMessage* msg = CreateMessage(G_ROCKETMQ_TOPIC_TEST.c_str());
|
||
SetMessageTags(msg, G_ROCKETMQ_TAG_TEST.c_str());
|
||
SetMessageKeys(msg, G_ROCKETMQ_KEY_TEST.c_str());
|
||
|
||
for (int i = 0; i < 10; i++)
|
||
{
|
||
// construct different body
|
||
string strMessageBody = "this is body number";
|
||
|
||
SetMessageBody(msg, strMessageBody.c_str());
|
||
// send message
|
||
SendMessageSync(producer, msg, &result);
|
||
|
||
cout << "send message[" << i << "], result status:" << result.sendStatus << ", msgBody:" << strMessageBody << endl;
|
||
usleep(1000);
|
||
}
|
||
|
||
// destroy message
|
||
DestroyMessage(msg);
|
||
}
|
||
//producer_send 测试用
|
||
void StartSendMessage(CProducer* producer,const char* strbody)
|
||
{
|
||
CSendResult result;
|
||
|
||
// create message and set some values for it
|
||
CMessage* msg = CreateMessage(G_ROCKETMQ_TOPIC_TEST.c_str());
|
||
SetMessageTags(msg, G_ROCKETMQ_TAG_TEST.c_str());
|
||
SetMessageKeys(msg, G_ROCKETMQ_KEY_TEST.c_str());
|
||
|
||
SetMessageBody(msg, strbody);
|
||
// send message
|
||
SendMessageSync(producer, msg, &result);
|
||
|
||
cout << "result status:" << result.sendStatus << ", msgBody:" << strbody << endl;
|
||
|
||
// destroy message
|
||
DestroyMessage(msg);
|
||
}
|
||
|
||
//测试用 固定消息体
|
||
void producer_send0()
|
||
{
|
||
cout << "Producer Initializing....." << endl;
|
||
|
||
// create producer and set some values for it
|
||
CProducer* producer = CreateProducer(G_ROCKETMQ_PRODUCER.c_str());
|
||
SetProducerNameServerAddress(producer, G_ROCKETMQ_IPPORT.c_str());
|
||
// start producer
|
||
StartProducer(producer);
|
||
cout << "Producer start....." << endl;
|
||
// send message
|
||
StartSendMessage(producer);
|
||
// shutdown producer
|
||
ShutdownProducer(producer);
|
||
// destroy producer
|
||
DestroyProducer(producer);
|
||
cout << "Producer Shutdown!" << endl;
|
||
}
|
||
//测试用 可控制消息体
|
||
void producer_send(const char* strbody)
|
||
{
|
||
cout << "Producer Initializing....." << endl;
|
||
|
||
// create producer and set some values for it
|
||
CProducer* producer = CreateProducer(G_ROCKETMQ_PRODUCER.c_str());
|
||
SetProducerNameServerAddress(producer, G_ROCKETMQ_IPPORT.c_str());
|
||
// start producer
|
||
StartProducer(producer);
|
||
cout << "Producer start....." << endl;
|
||
// send message
|
||
StartSendMessage(producer, strbody);
|
||
// shutdown producer
|
||
ShutdownProducer(producer);
|
||
// destroy producer
|
||
DestroyProducer(producer);
|
||
cout << "Producer Shutdown!" << endl;
|
||
}
|
||
*/
|
||
///////////////////////////////////////////////////////////////////////////////////////////////////////////
|
||
|
||
extern "C" {
|
||
//extern std::string G_MQCONSUMER_TOPIC_RT;
|
||
void rocketmq_test_rt()
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_RT);
|
||
std::ifstream file("rt.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
//extern std::string G_MQCONSUMER_TOPIC_UD;
|
||
void rocketmq_test_ud()//用来测试台账更新
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_UD);
|
||
std::ifstream file("ud.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
|
||
void rocketmq_test_set()//用来测试进程控制脚本
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_SET);
|
||
std::ifstream file("set.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
|
||
void rocketmq_test_only()//用来测试进程控制脚本
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_TEST);
|
||
std::ifstream file("test.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
|
||
//extern std::string G_MQCONSUMER_TOPIC_RC;
|
||
void rocketmq_test_rc()
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_RC);
|
||
std::ifstream file("rc.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
|
||
//extern std::string G_MQCONSUMER_TOPIC_LOG;
|
||
void rocketmq_test_log()
|
||
{
|
||
Ckafka_data_t data;
|
||
data.monitor_id = 123123;
|
||
data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_LOG);
|
||
std::ifstream file("log_test.txt"); // 文件中存储长字符串
|
||
std::stringstream buffer;
|
||
buffer << file.rdbuf(); // 读取整个文件内容
|
||
|
||
data.strText = QString::fromStdString(buffer.str());
|
||
data.mp_id = 123123;
|
||
my_rocketmq_send(data);
|
||
}
|
||
|
||
}
|
||
|
||
|