From ae911b46e5dc0ca6d40cbd5a2373676b51af3512 Mon Sep 17 00:00:00 2001 From: lnk Date: Wed, 24 Jun 2026 16:37:32 +0800 Subject: [PATCH] =?UTF-8?q?redis=E5=BC=95=E5=85=A5=E6=9C=AA=E5=AE=8C?= =?UTF-8?q?=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cfg_parse/SimpleProducer.cpp | 137 ++++- cfg_parse/cfg_parser.cpp | 17 + include/rocketmq/SimpleProducer.h | 16 +- json/create_json.cpp | 3 + mms/db_interface.h | 5 + pt61850netd_pqfe.pro | 19 +- redisstream/RedisStreamMQ.cpp | 948 ++++++++++++++++++++++++++++++ redisstream/RedisStreamMQ.h | 40 ++ redisstream/mq_compat.h | 82 +++ rocketmq/SimpleProducer.h | 87 ++- 10 files changed, 1312 insertions(+), 42 deletions(-) create mode 100644 redisstream/RedisStreamMQ.cpp create mode 100644 redisstream/RedisStreamMQ.h create mode 100644 redisstream/mq_compat.h diff --git a/cfg_parse/SimpleProducer.cpp b/cfg_parse/SimpleProducer.cpp index 524f4a0..5daadfc 100644 --- a/cfg_parse/SimpleProducer.cpp +++ b/cfg_parse/SimpleProducer.cpp @@ -43,6 +43,10 @@ #include "../log4cplus/log4.h"//lnk添加log4 +//添加redis stream相关头文件 +#include "../redisstream/RedisStreamMQ.h" +#include + using namespace std; static std::once_flag g_consumer_once; @@ -63,6 +67,11 @@ extern std::string G_MQCONSUMER_TOPIC_RT; extern std::string G_MQCONSUMER_TOPIC_TEST; +//消费者连接秘钥 +extern std::string G_MQCONSUMER_ACCESSKEY; +extern std::string G_MQCONSUMER_SECRETKEY; +extern std::string G_MQCONSUMER_CHANNEL; + extern std::string FRONT_INST; extern bool DEBUGOPEN; @@ -80,10 +89,9 @@ 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; +extern int G_MQ_BACKEND; + +//////////////////////////////////////////////////////////////////////////////////////////////////////// // 前向声明 RocketMQConsumer 类 class RocketMQConsumer; @@ -102,11 +110,23 @@ int64_t G_APP_START_MS = []() -> int64_t { int64_t G_START_SKEW_MS = 1000; -bool should_process_after_start(const rocketmq::MQMessageExt& msg) +bool RocketMQ_should_process_after_start(const rocketmq::MQMessageExt& msg) { const int64_t born_ts = static_cast(msg.getBornTimestamp()); return born_ts >= (G_APP_START_MS - G_START_SKEW_MS); } + +bool should_process_after_start(const rocketmq::MQMessageExt& msg) +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + return RedisStream_should_process_after_start(msg); + } + else + { + return RocketMQ_should_process_after_start(msg); + } +} /////////////////////////////// class InternalListener; @@ -419,7 +439,7 @@ void RocketMQConsumer::setConsumerMessageModel(const std::string& topic) RocketMQConsumer* g_consumer = NULL; // 初始化消费者函数 -void InitializeConsumer( +void RocketMQ_InitializeConsumer( const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions) // 接收多个订阅 @@ -446,7 +466,7 @@ void InitializeConsumer( } // 关闭并销毁消费者函数 -void ShutdownAndDestroyConsumer() +void RocketMQ_ShutdownAndDestroyConsumer() { if (g_consumer != NULL) { delete g_consumer; @@ -455,7 +475,7 @@ void ShutdownAndDestroyConsumer() } // 消费消息的接口函数 -void rocketmq_consumer_receive( +void RocketMQ_rocketmq_consumer_receive( const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions) // 接收多个订阅 @@ -844,7 +864,7 @@ private: RocketMQProducer* g_producer = NULL; // 初始化生产者（在程序启动时调用一次） -void InitializeProducer() +void RocketMQ_InitializeProducer() { if (g_producer == NULL) { try { @@ -859,7 +879,7 @@ void InitializeProducer() } // 关闭并销毁生产者（在程序结束时调用一次） -void ShutdownAndDestroyProducer() +void RocketMQ_ShutdownAndDestroyProducer() { if (g_producer != NULL) { delete g_producer; @@ -868,7 +888,7 @@ void ShutdownAndDestroyProducer() } // 发送消息的接口函数 -void rocketmq_producer_send(const std::string& body, +void RocketMQ_rocketmq_producer_send(const std::string& body, const std::string& topic, const std::string& tags, const std::string& keys) @@ -887,7 +907,102 @@ void rocketmq_producer_send(const std::string& body, } } #endif +/////////////////////////////////////////////////////////////////////////////////////////////////////////生产者消费者切换 +void InitializeProducer() +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_InitializeProducer(); + return; + } + RocketMQ_InitializeProducer(); +} +void ShutdownAndDestroyProducer() +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_ShutdownAndDestroyProducer(); + return; + } + + RocketMQ_ShutdownAndDestroyProducer(); +} +void rocketmq_producer_send(const std::string& body, + const std::string& topic, + const std::string& tags, + const std::string& keys) +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_producer_send( + body, + topic, + tags, + keys + ); + return; + } + + RocketMQ_rocketmq_producer_send( + body, + topic, + tags, + keys + ); +} +void InitializeConsumer( + const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions) +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_InitializeConsumer( + consumerName, + nameServer, + subscriptions + ); + return; + } + + RocketMQ_InitializeConsumer( + consumerName, + nameServer, + subscriptions + ); +} +void ShutdownAndDestroyConsumer() +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_ShutdownAndDestroyConsumer(); + return; + } + + RocketMQ_ShutdownAndDestroyConsumer(); +} +void rocketmq_consumer_receive( + const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions) +{ + if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM) + { + RedisStream_consumer_receive( + consumerName, + nameServer, + subscriptions + ); + return; + } + + RocketMQ_rocketmq_consumer_receive( + consumerName, + nameServer, + subscriptions + ); +} ////////////////////////////////////////////////////////////////////////////////////////////////////////// /* // producer_send0测试用 diff --git a/cfg_parse/cfg_parser.cpp b/cfg_parse/cfg_parser.cpp index 1de0d45..3baf24f 100644 --- a/cfg_parse/cfg_parser.cpp +++ b/cfg_parse/cfg_parser.cpp @@ -376,6 +376,9 @@ int G_LOG_RATE_KEEP_BURST_MS = 1000; // 间隔 >= 1000ms，按二级策略 int G_LOG_RATE_KEEP_BURST_COUNT = 60; // 二级保留前60条 int G_LOG_RATE_KEEP_HIGHFREQ_COUNT = 10; // 高频保留前10条 +//添加底层队列开关 +int G_MQ_BACKEND = 0; // 0-默认RocketMQ，1-Redis Stream + //lnk20250115添加台账锁 extern pthread_mutex_t mtx; @@ -649,6 +652,20 @@ void init_config() { qDebug() << "Read UDS_DOWNLOAD_URL:" << UDS_DOWNLOAD_URL << endl; /*添加rocketmq的解析10-9 *///////////////////////////////////////////////////////////////// + + ba = settings.value("RocketMq/Backend", "rocketmq").toString().toLatin1(); + std::string mq_backend = ba.data(); + + if (mq_backend == "redis" || mq_backend == "redisstream" || mq_backend == "1") { + G_MQ_BACKEND = MQ_BACKEND_REDIS_STREAM; + } else { + G_MQ_BACKEND = MQ_BACKEND_ROCKETMQ; + } + + std::cout << "Read RocketMq/Backend:" << mq_backend + << ", G_MQ_BACKEND:" << G_MQ_BACKEND + << std::endl; + //生产者 ba = settings.value("RocketMq/producer", "").toString().toLatin1(); G_ROCKETMQ_PRODUCER = strdup(ba.data()); diff --git a/include/rocketmq/SimpleProducer.h b/include/rocketmq/SimpleProducer.h index f76b95c..8819247 100644 --- a/include/rocketmq/SimpleProducer.h +++ b/include/rocketmq/SimpleProducer.h @@ -23,9 +23,9 @@ using namespace rocketmq; //void producer_send0(); //void StartSendMessage(CProducer* producer,const char* strbody); //void producer_send(const char* strbody); -//void rocketmq_producer_send(const char* strbody,const char* topic); +//void RocketMQ_rocketmq_producer_send(const char* strbody,const char* topic); //void rocketmq_StartSendMessage(CProducer* producer,const char* strbody,const char* topic); -void rocketmq_producer_send(const std::string& body, +void RocketMQ_rocketmq_producer_send(const std::string& body, const std::string& topic, const std::string& tags, const std::string& keys); @@ -42,8 +42,8 @@ extern void my_rocketmq_send(Ckafka_data_t& data); ///////////////////////////////////////////////////////生产者 -void InitializeProducer(); -void ShutdownAndDestroyProducer(); +void RocketMQ_InitializeProducer(); +void RocketMQ_ShutdownAndDestroyProducer(); //////////////////////////////////////////////////////消费者 typedef ConsumeStatus (*MessageCallBack)( const MQMessageExt& msg @@ -59,13 +59,13 @@ struct Subscription { MessageCallBack cb) : topic(t), tag(tg), callback(cb) {} }; -//void InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key); -void InitializeConsumer(const std::string& consumerName, +//void RocketMQ_InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key); +void RocketMQ_InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions); -void ShutdownAndDestroyConsumer(); +void RocketMQ_ShutdownAndDestroyConsumer(); -void rocketmq_consumer_receive( +void RocketMQ_rocketmq_consumer_receive( const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions); diff --git a/json/create_json.cpp b/json/create_json.cpp index ff2c130..c9bf45d 100644 --- a/json/create_json.cpp +++ b/json/create_json.cpp @@ -2471,6 +2471,9 @@ int transfer_json_block_data(char v_wiring_type[], json_block_data *data) //json kafka_data_list_mutex.unlock(); //解锁 longjumpflag = true; } + else if (countflag < num && 0 != time_sec % 7200) { + longjumpflag = true;//不发送数据 + } if (typeofdata == false || data_have_static == false) {//不合并则处理完闪变就不处理其他数据 if (longjumpflag == true || shortjumpflag == true) { return 1; diff --git a/mms/db_interface.h b/mms/db_interface.h index 32d72c8..10986c8 100644 --- a/mms/db_interface.h +++ b/mms/db_interface.h @@ -42,6 +42,11 @@ #define SHOULD_UNREGISTER 2 /////////////////////////////////////////////////////////////////////////////// +#define MQ_BACKEND_ROCKETMQ 0 +#define MQ_BACKEND_REDIS_STREAM 1 + +/////////////////////////////////////////////////////////////////////////////// + #ifdef __cplusplus extern "C" { #endif /* __cplusplus */ diff --git a/pt61850netd_pqfe.pro b/pt61850netd_pqfe.pro index 9b6a893..10ca2f7 100644 --- a/pt61850netd_pqfe.pro +++ b/pt61850netd_pqfe.pro @@ -2,7 +2,13 @@ TEMPLATE = app TARGET = pt61850netd_pqfe DEPENDPATH += . -INCLUDEPATH += . ./source/include ./source/include/mmslite ./source/include/pg_inst ./source/include/curl ./source/include/oss_sdk ./source/include/roketmq +INCLUDEPATH += . ./source/include \ + ./source/include/mmslite \ + ./source/include/pg_inst \ + ./source/include/curl \ + ./source/include/oss_sdk \ + ./source/include/roketmq \ + ./source/redisstream QMAKE_ORIG_TARGET = $(TARGET) @@ -14,6 +20,8 @@ DEFINES += _CRT_SECURE_NO_WARNINGS DEFINES += MMS_LITE LINUX=2 MOSI LEAN_T TP0_ENABLED DEFINES += CLIENT _DEBUG _REENTRANT _GNU_SOURCE _LARGEFILE64_SOURCE +DEFINES += USE_MQ_MULTI_BACKEND + # 添加 debug/release 编译选项配置 CONFIG(debug, debug|release) { message("Building debug version with debug symbols") @@ -98,6 +106,8 @@ unix { LIBS += -lrdkafka++ LIBS += -lhttprun LIBS += -llog4cplus + LIBS += -lhiredis + LIBS += -lz } #install @@ -127,7 +137,9 @@ HEADERS += source/mms/db_interface.h \ source/json/cjson.h \ source/rocketmq/SimpleProducer.h \ source/cfg_parse/custom_printf.h \ - source/log4cplus/log4.h + source/log4cplus/log4.h \ + source/redisstream/mq_compat.h \ + source/redisstream/RedisStreamMQ.h SOURCES += source/mms/main.c \ source/mms/clntobj.c \ @@ -163,4 +175,5 @@ SOURCES += source/mms/main.c \ source/cfg_parse/base64.cpp \ source/cfg_parse/uds_huaweiyun.cpp \ source/cfg_parse/SimpleProducer.cpp \ - source/cfg_parse/log4.cpp + source/cfg_parse/log4.cpp \ + source/redisstream/RedisStreamMQ.cpp diff --git a/redisstream/RedisStreamMQ.cpp b/redisstream/RedisStreamMQ.cpp new file mode 100644 index 0000000..f0df75e --- /dev/null +++ b/redisstream/RedisStreamMQ.cpp @@ -0,0 +1,948 @@ +#include "RedisStreamMQ.h" +#include "../rocketmq/SimpleProducer.h" + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "../log4cplus/log4.h" + +/* + * lnk20260622 修改：运行时切换版 Redis Streams 后端实现。 + * + * 本文件只导出 RedisStream_xxx 函数，不再导出： + * InitializeProducer / ShutdownAndDestroyProducer + * rocketmq_producer_send + * InitializeConsumer / ShutdownAndDestroyConsumer + * rocketmq_consumer_receive + * + * 这些统一入口应在 SimpleProducer.cpp 中按配置文件 [MQ] Backend 转发， + * 否则同时编译 RocketMQ 和 RedisStream 时会发生重复定义链接错误。 + */ + +extern std::string G_ROCKETMQ_PRODUCER; // 保留原配置名；Redis 模式下仅用于日志 +extern std::string G_ROCKETMQ_IPPORT; // Redis 模式下建议填 127.0.0.1:6379[/db] +extern std::string G_MQCONSUMER_IPPORT; // Redis 模式下消费者连接地址 +extern std::string G_ROCKETMQ_CONSUMER; // Redis consumer group +extern std::string FRONT_INST; + +extern int g_front_seg_index; +extern char subdir[128]; + +static const int64_t REDIS_BLOCK_MS = 1000; //XREADGROUP 阻塞等待时间 +static const int64_t REDIS_IDLE_CLAIM_MS = 60000; //消息空闲多久被其他消费者抢占 +static const int REDIS_READ_COUNT = 16; //每次 XREADGROUP 的最大读取数量 +static const size_t GZIP_THRESHOLD_BYTES = 1024; //超过这个大小的消息才压缩 +static const char* SPOOL_DIR = "/FeProject/data/redisstream_spool"; //本地缓存目录 + +static int64_t now_ms() //当前时间，单位毫秒 +{ + struct timeval tv; + gettimeofday(&tv, NULL); + return (int64_t)tv.tv_sec * 1000 + tv.tv_usec / 1000; +} + +/* + * 注意：如果 RocketMQ 原 SimpleProducer.cpp 已经定义了 G_APP_START_MS / G_START_SKEW_MS， + * 这里不要再定义全局同名变量，避免多后端同时编译时重复定义。 + */ +static int64_t redis_app_start_ms()//RedisStream 后端的程序启动时间 +{ + static int64_t v = now_ms(); + return v; +} + +static const int64_t REDIS_START_SKEW_MS = 1000; + +static std::string to_string_i64(int64_t v) +{ + char buf[64]; + snprintf(buf, sizeof(buf), "%lld", (long long)v); + return std::string(buf); +} + +static std::string shell_safe_name(const std::string& s) +{ + std::string r; + for (size_t i = 0; i < s.size(); ++i) { + unsigned char c = (unsigned char)s[i]; + if ((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || + (c >= '0' && c <= '9') || c == '_' || c == '-' || c == '.') { + r.push_back((char)c); + } else { + r.push_back('_'); + } + } + return r.empty() ? "empty" : r; +} + +static std::string build_stream_name(const std::string& topic, const std::string& env)//构造stream名称，格式：env_shortTopic +{ + if (!topic.empty() && topic.find('%') != std::string::npos) { + std::string shortTopic = topic.substr(topic.find('%') + 1); + if (!env.empty()) return env + "_" + shortTopic; + return shortTopic; + } + + if (!env.empty()) { + const std::string prefix = env + "_"; + if (topic.find(prefix) == 0) return topic; + return prefix + topic; + } + return topic; +} + +static void ensure_dir(const std::string& dir)//创建目录，如果不存在 +{ + if (dir.empty()) return; + if (access(dir.c_str(), F_OK) == 0) return; + if (mkdir(dir.c_str(), 0755) != 0 && errno != EEXIST) { + std::cerr << "[REDIS_STREAM][SPOOL] mkdir failed: " << dir + << " errno=" << errno << " " << strerror(errno) << std::endl; + } +} + +static std::string base64_encode(const std::string& input)//编码为 base64 +{ + static const char* table = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + std::string out; + int val = 0; + int valb = -6; + for (size_t i = 0; i < input.size(); ++i) { + unsigned char c = (unsigned char)input[i]; + val = (val << 8) + c; + valb += 8; + while (valb >= 0) { + out.push_back(table[(val >> valb) & 0x3F]); + valb -= 6; + } + } + if (valb > -6) out.push_back(table[((val << 8) >> (valb + 8)) & 0x3F]); + while (out.size() % 4) out.push_back('='); + return out; +} + +static std::string gzip_compress(const std::string& input)//压缩字符串，返回 gzip 格式 +{ + if (input.empty()) return std::string(); + + z_stream zs; + memset(&zs, 0, sizeof(zs)); + + int ret = deflateInit2(&zs, Z_BEST_SPEED, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY); + if (ret != Z_OK) { + throw std::runtime_error("deflateInit2 failed"); + } + + zs.next_in = (Bytef*)input.data(); + zs.avail_in = (uInt)input.size(); + + char outbuffer[32768]; + std::string out; + + do { + zs.next_out = reinterpret_cast(outbuffer); + zs.avail_out = sizeof(outbuffer); + + ret = deflate(&zs, Z_FINISH); + + if (out.size() < zs.total_out) { + out.append(outbuffer, zs.total_out - out.size()); + } + } while (ret == Z_OK); + + deflateEnd(&zs); + + if (ret != Z_STREAM_END) { + throw std::runtime_error("gzip deflate failed"); + } + + return out; +} + +static bool gzip_decompress(const std::string& input, std::string& output) //解压缩 gzip 格式字符串 +{ + output.clear(); + if (input.empty()) return true; + + z_stream zs; + memset(&zs, 0, sizeof(zs)); + + int ret = inflateInit2(&zs, 15 + 32); + if (ret != Z_OK) return false; + + zs.next_in = (Bytef*)input.data(); + zs.avail_in = (uInt)input.size(); + + char outbuffer[32768]; + + do { + zs.next_out = reinterpret_cast(outbuffer); + zs.avail_out = sizeof(outbuffer); + + ret = inflate(&zs, 0); + + if (output.size() < zs.total_out) { + output.append(outbuffer, zs.total_out - output.size()); + } + + if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) { + inflateEnd(&zs); + return false; + } + } while (ret != Z_STREAM_END); + + inflateEnd(&zs); + return true; +} + +static int base64_value(char c) +{ + if (c >= 'A' && c <= 'Z') return c - 'A'; + if (c >= 'a' && c <= 'z') return c - 'a' + 26; + if (c >= '0' && c <= '9') return c - '0' + 52; + if (c == '+') return 62; + if (c == '/') return 63; + return -1; +} + +static bool base64_decode(const std::string& input, std::string& output) //解码 base64 +{ + output.clear(); + int val = 0; + int valb = -8; + for (size_t i = 0; i < input.size(); ++i) { + char c = input[i]; + if (c == '=') break; + int d = base64_value(c); + if (d == -1) { + if (c == '\r' || c == '\n' || c == ' ' || c == '\t') continue; + return false; + } + val = (val << 6) + d; + valb += 6; + if (valb >= 0) { + output.push_back(char((val >> valb) & 0xFF)); + valb -= 8; + } + } + return true; +} + +static bool decode_body(const std::string& enc, //解压缩 + const std::string& body, + std::string& out) +{ + if (enc == "gzip") { + std::string gz; + if (!base64_decode(body, gz)) return false; + return gzip_decompress(gz, out); + } + + out = body; + return true; +} + +static void encode_body(const std::string& json, + std::string& enc, + std::string& body) +{ + if (json.size() >= GZIP_THRESHOLD_BYTES) { + enc = "gzip"; + body = base64_encode(gzip_compress(json)); //大的压缩后编码 + } else { + enc = "plain"; + body = json; //小的保持json + } +} + +static std::string parse_host(const std::string& addr)//解析主机地址，去掉协议、端口和数据库部分 +{ + std::string s = addr; + if (s.find("redis://") == 0) s = s.substr(8); + size_t slash = s.find('/'); + if (slash != std::string::npos) s = s.substr(0, slash); + size_t colon = s.rfind(':'); + if (colon == std::string::npos) return s; + return s.substr(0, colon); +} + +static int parse_port(const std::string& addr)//提取端口 +{ + std::string s = addr; + if (s.find("redis://") == 0) s = s.substr(8); + size_t slash = s.find('/'); + if (slash != std::string::npos) s = s.substr(0, slash); + size_t colon = s.rfind(':'); + if (colon == std::string::npos) return 6379; + return atoi(s.substr(colon + 1).c_str()); +} + +static int parse_db(const std::string& addr) //提取数据库 +{ + size_t slash = addr.rfind('/'); + if (slash == std::string::npos) return -1; + std::string db = addr.substr(slash + 1); + if (db.empty()) return -1; + for (size_t i = 0; i < db.size(); ++i) { + if (!isdigit((unsigned char)db[i])) return -1; + } + return atoi(db.c_str()); +} + +class RedisConnection { //Redis 连接封装 +public: + RedisConnection() : ctx_(NULL), db_(-1) {} + ~RedisConnection() { close(); } + + void configure(const std::string& addr) + { + addr_ = addr; + host_ = parse_host(addr); + port_ = parse_port(addr); + db_ = parse_db(addr); + if (host_.empty()) host_ = "127.0.0.1"; + } + + redisContext* get() + { + if (ctx_ && ctx_->err == 0) return ctx_; + reconnect(); + return ctx_; + } + + redisReply* command(const char* fmt, ...) + { + redisContext* c = get(); + if (!c) return NULL; + + va_list ap; + va_start(ap, fmt); + void* r = redisvCommand(c, fmt, ap); + va_end(ap); + + if (!r) { + close(); + return NULL; + } + return (redisReply*)r; + } + + bool ping() + { + redisReply* r = command("PING"); + if (!r) return false; + bool ok = (r->type == REDIS_REPLY_STATUS && r->str && std::string(r->str) == "PONG"); + freeReplyObject(r); + return ok; + } + + void close() + { + if (ctx_) { + redisFree(ctx_); + ctx_ = NULL; + } + } + +private: + void reconnect() + { + close(); + + struct timeval tv; + tv.tv_sec = 1; + tv.tv_usec = 500000; + + ctx_ = redisConnectWithTimeout(host_.c_str(), port_, tv); + if (!ctx_ || ctx_->err) { + if (ctx_) { + std::cerr << "[REDIS_STREAM] connect failed: " << ctx_->errstr << std::endl; + close(); + } else { + std::cerr << "[REDIS_STREAM] connect failed: null context" << std::endl; + } + return; + } + + redisSetTimeout(ctx_, tv); + + if (db_ >= 0) { + redisReply* r = (redisReply*)redisCommand(ctx_, "SELECT %d", db_); + if (!r || (r->type == REDIS_REPLY_ERROR)) { + std::cerr << "[REDIS_STREAM] select db failed: " << db_ << std::endl; + if (r) freeReplyObject(r); + close(); + return; + } + freeReplyObject(r); + } + } + +private: + redisContext* ctx_; + std::string addr_; + std::string host_; + int port_; + int db_; +}; + +class RedisStreamProducer { //RedisStream 生产者封装 +public: + explicit RedisStreamProducer(const std::string& addr) + { + conn_.configure(addr); + ensure_dir(SPOOL_DIR); + if (!conn_.ping()) { + std::cerr << "[REDIS_STREAM][PRODUCER] Redis not ready, spool enabled" << std::endl; + } + } + + void sendMessage(const std::string& json, + const std::string& topic, + const std::string& tag, + const std::string& key) + { + flushSpool(); + + std::string enc; + std::string body; //序列化的数据json + encode_body(json, enc, body); //压缩 + + const std::string stream = build_stream_name(topic, tag); //构造stream名称，格式：env_shortTopic + const std::string ts = to_string_i64(now_ms()); + + if (!xadd(stream, key, tag, enc, body, ts)) { // + spool(stream, key, tag, enc, body, ts); + throw std::runtime_error("Redis XADD failed, message spooled"); + } + } + + void shutdown() {} + +private: + bool xadd(const std::string& stream, + const std::string& key, + const std::string& tag, + const std::string& enc, + const std::string& body, + const std::string& ts) + { + redisReply* r = conn_.command( + "XADD %b * key %b tag %b enc %b body %b ts %b", + stream.data(), stream.size(), + key.data(), key.size(), + tag.data(), tag.size(), + enc.data(), enc.size(), + body.data(), body.size(), + ts.data(), ts.size()); + + if (!r) return false; + + bool ok = (r->type == REDIS_REPLY_STRING); + if (!ok && r->type == REDIS_REPLY_ERROR) { + std::cerr << "[REDIS_STREAM][XADD_ERROR] " << (r->str ? r->str : "") << std::endl; + } + freeReplyObject(r); + return ok; + } + + void spool(const std::string& stream, + const std::string& key, + const std::string& tag, + const std::string& enc, + const std::string& body, + const std::string& ts) + { + ensure_dir(SPOOL_DIR); + std::string fn = std::string(SPOOL_DIR) + "/" + to_string_i64(now_ms()) + "_" + + shell_safe_name(stream) + "_" + shell_safe_name(key) + ".msg.tmp"; + std::string done = fn.substr(0, fn.size() - 4); + + std::ofstream out(fn.c_str(), std::ios::out | std::ios::binary); + if (!out) { + std::cerr << "[REDIS_STREAM][SPOOL] open failed: " << fn << std::endl; + return; + } + + out << "stream=" << stream << "\n"; + out << "key=" << key << "\n"; + out << "tag=" << tag << "\n"; + out << "enc=" << enc << "\n"; + out << "ts=" << ts << "\n"; + out << "body=" << body << "\n"; + out.close(); + + if (rename(fn.c_str(), done.c_str()) != 0) { + std::cerr << "[REDIS_STREAM][SPOOL] rename failed errno=" << errno << std::endl; + } + } + + static bool parse_spool_file(const std::string& path, + std::string& stream, + std::string& key, + std::string& tag, + std::string& enc, + std::string& body, + std::string& ts) + { + std::ifstream in(path.c_str(), std::ios::in | std::ios::binary); + if (!in) return false; + + std::string line; + while (std::getline(in, line)) { + if (line.find("stream=") == 0) stream = line.substr(7); + else if (line.find("key=") == 0) key = line.substr(4); + else if (line.find("tag=") == 0) tag = line.substr(4); + else if (line.find("enc=") == 0) enc = line.substr(4); + else if (line.find("ts=") == 0) ts = line.substr(3); + else if (line.find("body=") == 0) { + body = line.substr(5); + std::string rest; + while (std::getline(in, rest)) { + body += "\n"; + body += rest; + } + break; + } + } + return !stream.empty() && !enc.empty() && !ts.empty(); + } + + void flushSpool() + { + DIR* dir = opendir(SPOOL_DIR); + if (!dir) return; + + std::vector files; + struct dirent* ent = NULL; + while ((ent = readdir(dir)) != NULL) { + std::string name = ent->d_name; + if (name == "." || name == "..") continue; + if (name.size() >= 4 && name.substr(name.size() - 4) == ".tmp") continue; + files.push_back(std::string(SPOOL_DIR) + "/" + name); + } + closedir(dir); + + std::sort(files.begin(), files.end()); + + for (size_t i = 0; i < files.size(); ++i) { + std::string stream, key, tag, enc, body, ts; + if (!parse_spool_file(files[i], stream, key, tag, enc, body, ts)) { + continue; + } + if (xadd(stream, key, tag, enc, body, ts)) { + unlink(files[i].c_str()); + std::cout << "[REDIS_STREAM][SPOOL_FLUSHED] " << files[i] << std::endl; + } else { + break; + } + } + } + +private: + RedisConnection conn_; +}; + +class RedisStreamConsumer { //RedisStream 消费者封装 +public: + RedisStreamConsumer(const std::string& group, const std::string& addr) + : group_(group), stop_(false) + { + conn_.configure(addr); + if (group_.empty()) group_ = "front_default"; + + char buf[64]; + snprintf(buf, sizeof(buf), "_%s_%d", subdir, g_front_seg_index); + + group_ += buf; + + consumer_ = group_ + "_" + to_string_i64((int64_t)getpid()) + "_" + to_string_i64(now_ms()); + } + + void subscribe(const std::string& topic, + const std::string& tag, + MessageCallBack cb) + { + Sub s; + s.topic = topic; + s.tag = tag; + s.stream = build_stream_name(topic, tag); + s.cb = cb; + subs_.push_back(s); + + callbacks_[std::make_pair(topic, tag)] = cb; + callbacks_[std::make_pair(s.stream, tag)] = cb; + + size_t pos = topic.find('%'); + if (pos != std::string::npos) { + std::string shortTopic = topic.substr(pos + 1); + callbacks_[std::make_pair(shortTopic, tag)] = cb; + } + + createGroup(s.stream); + + std::cout << "[REDIS_STREAM][SUB] stream=" << s.stream + << " topic=" << topic << " tag=" << tag + << " group=" << group_ << " consumer=" << consumer_ << std::endl; + } + + void start() + { + stop_ = false; + while (!stop_) { + claimOnce(); + readOnce(); + } + } + + void shutdown() + { + stop_ = true; + conn_.close(); + } + +private: + struct Sub { + std::string topic; + std::string tag; + std::string stream; + MessageCallBack cb; + }; + + void createGroup(const std::string& stream) + { + redisReply* r = conn_.command("XGROUP CREATE %b %b $ MKSTREAM", + stream.data(), stream.size(), + group_.data(), group_.size()); + if (!r) { + std::cerr << "[REDIS_STREAM][GROUP] create failed: " << stream << std::endl; + return; + } + + if (r->type == REDIS_REPLY_ERROR) { + std::string err = r->str ? r->str : ""; + if (err.find("BUSYGROUP") == std::string::npos) { + std::cerr << "[REDIS_STREAM][GROUP_ERROR] " << stream << " " << err << std::endl; + } + } + freeReplyObject(r); + } + + bool parseFields(redisReply* arr, + std::map& fields) + { + fields.clear(); + if (!arr || arr->type != REDIS_REPLY_ARRAY) return false; + for (size_t i = 0; i + 1 < arr->elements; i += 2) { + redisReply* k = arr->element[i]; + redisReply* v = arr->element[i + 1]; + if (!k || !v || k->type != REDIS_REPLY_STRING || v->type != REDIS_REPLY_STRING) continue; + fields[std::string(k->str, k->len)] = std::string(v->str, v->len); + } + return true; + } + + static std::string getField(const std::map& m, + const std::string& k) + { + std::map::const_iterator it = m.find(k); + if (it == m.end()) return ""; + return it->second; + } + + rocketmq::MQMessageExt makeMsg(const Sub& sub, + const std::string& id, + const std::map& fields) + { + std::string enc = getField(fields, "enc"); + std::string bodyRaw = getField(fields, "body"); + std::string body; + if (!decode_body(enc, bodyRaw, body)) { + throw std::runtime_error("decode body failed"); + } + + std::string key = getField(fields, "key"); + std::string tag = getField(fields, "tag"); + std::string ts = getField(fields, "ts"); + + if (tag.empty()) tag = sub.tag; + + int64_t born = ts.empty() ? now_ms() : atoll(ts.c_str()); + + return rocketmq::MQMessageExt(sub.topic, tag, key, body, born, id); + } + + MessageCallBack findCallback(const std::string& topic, + const std::string& stream, + const std::string& tag) + { + std::map, MessageCallBack>::iterator it; + it = callbacks_.find(std::make_pair(topic, tag)); + if (it != callbacks_.end()) return it->second; + + it = callbacks_.find(std::make_pair(stream, tag)); + if (it != callbacks_.end()) return it->second; + + return NULL; + } + + void ack(const std::string& stream, const std::string& id) + { + redisReply* r = conn_.command("XACK %b %b %b", + stream.data(), stream.size(), + group_.data(), group_.size(), + id.data(), id.size()); + if (r) freeReplyObject(r); + } + + void handleOne(const Sub& sub, const std::string& id, redisReply* fieldsReply) + { + try { + std::map fields; + if (!parseFields(fieldsReply, fields)) { + std::cerr << "[REDIS_STREAM][BAD_MSG] no fields stream=" << sub.stream << " id=" << id << std::endl; + ack(sub.stream, id); + return; + } + + rocketmq::MQMessageExt msg = makeMsg(sub, id, fields); + + MessageCallBack cb = findCallback(sub.topic, sub.stream, msg.getTags()); + if (!cb) { + std::cout << "[REDIS_STREAM][NO_CALLBACK] stream=" << sub.stream + << " tag=" << msg.getTags() << std::endl; + ack(sub.stream, id); + return; + } + + rocketmq::ConsumeStatus ret = cb(msg); + if (ret == rocketmq::CONSUME_SUCCESS) { + ack(sub.stream, id); + } else { + std::cout << "[REDIS_STREAM][RETRY_LATER] stream=" << sub.stream + << " id=" << id << std::endl; + } + } catch (const std::exception& e) { + std::cerr << "[REDIS_STREAM][HANDLE_EXCEPTION] " << e.what() + << " stream=" << sub.stream << " id=" << id << std::endl; + } catch (...) { + std::cerr << "[REDIS_STREAM][HANDLE_UNKNOWN_EXCEPTION] stream=" + << sub.stream << " id=" << id << std::endl; + } + } + + void readOnce() + { + for (size_t i = 0; i < subs_.size(); ++i) { + Sub& sub = subs_[i]; + + redisReply* r = conn_.command( + "XREADGROUP GROUP %b %b COUNT %d BLOCK %lld STREAMS %b >", + group_.data(), group_.size(), + consumer_.data(), consumer_.size(), + REDIS_READ_COUNT, + (long long)REDIS_BLOCK_MS, + sub.stream.data(), sub.stream.size()); + + if (!r) { + usleep(500000); + continue; + } + + if (r->type == REDIS_REPLY_NIL) { + freeReplyObject(r); + continue; + } + + if (r->type == REDIS_REPLY_ERROR) { + std::string err = r->str ? r->str : ""; + std::cerr << "[REDIS_STREAM][XREADGROUP_ERROR] " << err << std::endl; + if (err.find("NOGROUP") != std::string::npos) { + createGroup(sub.stream); + } + freeReplyObject(r); + continue; + } + + if (r->type == REDIS_REPLY_ARRAY) { + for (size_t si = 0; si < r->elements; ++si) { + redisReply* streamPair = r->element[si]; + if (!streamPair || streamPair->type != REDIS_REPLY_ARRAY || streamPair->elements < 2) continue; + + redisReply* messages = streamPair->element[1]; + if (!messages || messages->type != REDIS_REPLY_ARRAY) continue; + + for (size_t mi = 0; mi < messages->elements; ++mi) { + redisReply* msgPair = messages->element[mi]; + if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue; + redisReply* idReply = msgPair->element[0]; + redisReply* fieldsReply = msgPair->element[1]; + if (!idReply || idReply->type != REDIS_REPLY_STRING) continue; + + std::string id(idReply->str, idReply->len); + handleOne(sub, id, fieldsReply); + } + } + } + + freeReplyObject(r); + } + } + + void claimOnce() + { + for (size_t i = 0; i < subs_.size(); ++i) { + Sub& sub = subs_[i]; + + redisReply* r = conn_.command( + "XAUTOCLAIM %b %b %b %lld 0-0 COUNT %d", + sub.stream.data(), sub.stream.size(), + group_.data(), group_.size(), + consumer_.data(), consumer_.size(), + (long long)REDIS_IDLE_CLAIM_MS, + REDIS_READ_COUNT); + + if (!r) continue; + + if (r->type == REDIS_REPLY_ERROR) { + std::string err = r->str ? r->str : ""; + if (err.find("NOGROUP") != std::string::npos) createGroup(sub.stream); + freeReplyObject(r); + continue; + } + + if (r->type == REDIS_REPLY_ARRAY && r->elements >= 2) { + redisReply* messages = r->element[1]; + if (messages && messages->type == REDIS_REPLY_ARRAY) { + for (size_t mi = 0; mi < messages->elements; ++mi) { + redisReply* msgPair = messages->element[mi]; + if (!msgPair || msgPair->type != REDIS_REPLY_ARRAY || msgPair->elements < 2) continue; + redisReply* idReply = msgPair->element[0]; + redisReply* fieldsReply = msgPair->element[1]; + if (!idReply || idReply->type != REDIS_REPLY_STRING) continue; + + std::string id(idReply->str, idReply->len); + std::cout << "[REDIS_STREAM][CLAIM] stream=" << sub.stream + << " id=" << id << std::endl; + handleOne(sub, id, fieldsReply); + } + } + } + + freeReplyObject(r); + } + } + +private: + RedisConnection conn_; + std::string group_; + std::string consumer_; + bool stop_; + std::vector subs_; + std::map, MessageCallBack> callbacks_; +}; + +static RedisStreamProducer* g_redis_producer = NULL; //全局生产者实例 +static RedisStreamConsumer* g_redis_consumer = NULL; //全局消费者实例 + +bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg) //判断消息是否在程序启动后产生 +{ + const int64_t born_ts = static_cast(msg.getBornTimestamp()); + return born_ts >= (redis_app_start_ms() - REDIS_START_SKEW_MS); +} + +void RedisStream_InitializeProducer() //初始化生产者 +{ + if (g_redis_producer == NULL) { + std::string addr = G_ROCKETMQ_IPPORT.empty() ? G_MQCONSUMER_IPPORT : G_ROCKETMQ_IPPORT; + g_redis_producer = new RedisStreamProducer(addr); + std::cout << "[REDIS_STREAM] producer initialized addr=" << addr << std::endl; + } +} + +void RedisStream_ShutdownAndDestroyProducer()//关闭生产者 +{ + if (g_redis_producer != NULL) { + g_redis_producer->shutdown(); + delete g_redis_producer; + g_redis_producer = NULL; + } +} + +void RedisStream_producer_send(const std::string& body, //发送消息 + const std::string& topic, + const std::string& tags, + const std::string& keys) +{ + if (g_redis_producer == NULL) RedisStream_InitializeProducer(); + + try { + g_redis_producer->sendMessage(body, topic, tags, keys); + } catch (const std::exception& e) { + std::cerr << "[REDIS_STREAM][SEND_FAIL] " << e.what() << std::endl; + DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, + "【ERROR】前置的%s%d号进程 redis stream发送失败,已尝试本地落盘", + get_front_msg_from_subdir(), g_front_seg_index); + } catch (...) { + std::cerr << "[REDIS_STREAM][SEND_FAIL] unknown exception" << std::endl; + DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, + "【ERROR】前置的%s%d号进程 redis stream发送未知异常", + get_front_msg_from_subdir(), g_front_seg_index); + } +} + +void RedisStream_InitializeConsumer(const std::string& consumerName, //初始化消费者 + const std::string& nameServer, + const std::vector& subscriptions) +{ + if (g_redis_consumer == NULL) { + g_redis_consumer = new RedisStreamConsumer(consumerName, nameServer); + for (size_t i = 0; i < subscriptions.size(); ++i) { + g_redis_consumer->subscribe(subscriptions[i].topic, + subscriptions[i].tag, + subscriptions[i].callback); + } + } +} + +void RedisStream_ShutdownAndDestroyConsumer()//关闭消费者 +{ + if (g_redis_consumer != NULL) { + g_redis_consumer->shutdown(); + delete g_redis_consumer; + g_redis_consumer = NULL; + } +} + +void RedisStream_consumer_receive(const std::string& consumerName, //启动消费者接收消息 + const std::string& nameServer, + const std::vector& subscriptions) +{ + RedisStream_InitializeConsumer(consumerName, nameServer, subscriptions); + if (g_redis_consumer) { + g_redis_consumer->start(); + } +} diff --git a/redisstream/RedisStreamMQ.h b/redisstream/RedisStreamMQ.h new file mode 100644 index 0000000..bf2db06 --- /dev/null +++ b/redisstream/RedisStreamMQ.h @@ -0,0 +1,40 @@ +#ifndef REDIS_STREAM_MQ_H +#define REDIS_STREAM_MQ_H + +/* + * lnk20260622 新增：Redis Streams 后端声明（运行时切换版）。 + * + * 注意： + * 1. 本文件不定义 InitializeProducer / rocketmq_producer_send 等统一入口。 + * 2. 只定义 RedisStream_xxx 后端函数，统一入口放在 SimpleProducer.cpp 中按配置转发。 + * 3. 这样 RocketMQ 原源码和 RedisStream 源码可以同时编译进程序。 + */ + +//#include "mq_compat.h" + +#include +#include + +struct Subscription; + +void RedisStream_InitializeProducer(); +void RedisStream_ShutdownAndDestroyProducer(); + +void RedisStream_producer_send(const std::string& body, + const std::string& topic, + const std::string& tags, + const std::string& keys); + +void RedisStream_InitializeConsumer(const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions); + +void RedisStream_ShutdownAndDestroyConsumer(); + +void RedisStream_consumer_receive(const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions); + +bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg); + +#endif // REDIS_STREAM_MQ_H diff --git a/redisstream/mq_compat.h b/redisstream/mq_compat.h new file mode 100644 index 0000000..5f30dde --- /dev/null +++ b/redisstream/mq_compat.h @@ -0,0 +1,82 @@ +#ifndef MQ_COMPAT_H +#define MQ_COMPAT_H + +#include +#include + +namespace mqcompat { + +enum ConsumeStatus { + CONSUME_SUCCESS = 0, + RECONSUME_LATER = 1 +}; + +class MQMessageExt { +public: + MQMessageExt() + : bornTimestamp_(0), + queueId_(0), + queueOffset_(0) { + } + + MQMessageExt(const std::string& topic, + const std::string& tags, + const std::string& keys, + const std::string& body, + int64_t bornTs, + const std::string& msgId) + : topic_(topic), + tags_(tags), + keys_(keys), + body_(body), + bornTimestamp_(bornTs), + queueId_(0), + queueOffset_(0), + msgId_(msgId), + streamId_(msgId) { + } + + const std::string& getTopic() const { return topic_; } + const std::string& getTags() const { return tags_; } + const std::string& getKeys() const { return keys_; } + const std::string& getBody() const { return body_; } + + int64_t getBornTimestamp() const { return bornTimestamp_; } + int getQueueId() const { return queueId_; } + int64_t getQueueOffset() const { return queueOffset_; } + + const std::string& getMsgId() const { return msgId_; } + const std::string& getStreamId() const { return streamId_; } + + void setTopic(const std::string& v) { topic_ = v; } + void setTags(const std::string& v) { tags_ = v; } + void setKeys(const std::string& v) { keys_ = v; } + void setBody(const std::string& v) { body_ = v; } + void setBornTimestamp(int64_t v) { bornTimestamp_ = v; } + void setQueueId(int v) { queueId_ = v; } + void setQueueOffset(int64_t v) { queueOffset_ = v; } + + void setMsgId(const std::string& v) { msgId_ = v; } + + void setStreamId(const std::string& v) { + streamId_ = v; + if (msgId_.empty()) { + msgId_ = v; + } + } + +private: + std::string topic_; + std::string tags_; + std::string keys_; + std::string body_; + int64_t bornTimestamp_; + int queueId_; + int64_t queueOffset_; + std::string msgId_; + std::string streamId_; +}; + +} // namespace mqcompat + +#endif \ No newline at end of file diff --git a/rocketmq/SimpleProducer.h b/rocketmq/SimpleProducer.h index f76b95c..90e68f0 100644 --- a/rocketmq/SimpleProducer.h +++ b/rocketmq/SimpleProducer.h @@ -1,10 +1,10 @@ +#ifndef SIMPLEPRODUCER_H +#define SIMPLEPRODUCER_H #ifdef __cplusplus + #include "../json/mms_json_inter.h" -//#include "../rocketmq/CProducer.h" -//#include "../rocketmq/CMessage.h" -//#include "../rocketmq/CSendResult.h" -//#include "../rocketmq/CPushConsumer.h" + #include "../rocketmq/DefaultMQProducer.h" #include "../rocketmq/MQMessage.h" #include "../rocketmq/SendResult.h" @@ -17,18 +17,44 @@ #include #include +/* + * lnk20260622 修改： + * 支持 RocketMQ / Redis Streams 双后端。 + * + * 编译时：两套源码都编译进去。 + * 运行时：通过配置文件 [MQ] Backend=rocketmq 或 redisstream 选择。 + * + * 业务层仍然使用： + * rocketmq::MQMessageExt + * rocketmq::ConsumeStatus + * Subscription + * InitializeProducer() + * rocketmq_producer_send() + * InitializeConsumer() + * + * 不需要业务层判断当前走 RocketMQ 还是 RedisStream。 + */ + +/* + * 关键： + * mq_compat.h 里必须定义兼容的： + * namespace rocketmq { + * enum ConsumeStatus; + * class MQMessageExt; + * } + * + * 如果实际编译 RocketMQ，也可以让 mq_compat.h 内部 include RocketMQ 原头。 + */ +//#include "../mq/mq_compat.h" +#include "../rocketmq/SimpleProducer.h" + using namespace rocketmq; -/*添加测试函数lnk10-10*/ -//void producer_send0(); -//void StartSendMessage(CProducer* producer,const char* strbody); -//void producer_send(const char* strbody); -//void rocketmq_producer_send(const char* strbody,const char* topic); -//void rocketmq_StartSendMessage(CProducer* producer,const char* strbody,const char* topic); -void rocketmq_producer_send(const std::string& body, +void RocketMQ_rocketmq_producer_send(const std::string& body, const std::string& topic, const std::string& tags, const std::string& keys); + extern "C" { void rocketmq_test_rt(); void rocketmq_test_ud(); @@ -36,15 +62,18 @@ void rocketmq_test_rc(); void rocketmq_test_log(); void rocketmq_test_set(); void rocketmq_test_only(); -void rocketmq_test_300(int mpnum,int front_index,int type); +void rocketmq_test_300(int mpnum, int front_index, int type); } + extern void my_rocketmq_send(Ckafka_data_t& data); +/////////////////////////////////////////////////////// +// 生产者统一入口 +void RocketMQ_InitializeProducer(); +void RocketMQ_ShutdownAndDestroyProducer(); -///////////////////////////////////////////////////////生产者 -void InitializeProducer(); -void ShutdownAndDestroyProducer(); -//////////////////////////////////////////////////////消费者 +/////////////////////////////////////////////////////// +// 消费者统一入口 typedef ConsumeStatus (*MessageCallBack)( const MQMessageExt& msg ); @@ -59,7 +88,24 @@ struct Subscription { MessageCallBack cb) : topic(t), tag(tg), callback(cb) {} }; -//void InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key); + +void RocketMQ_InitializeConsumer(const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions); + +void RocketMQ_ShutdownAndDestroyConsumer(); + +void RocketMQ_rocketmq_consumer_receive( + const std::string& consumerName, + const std::string& nameServer, + const std::vector& subscriptions); + +void rocketmq_producer_send(const std::string& body, + const std::string& topic, + const std::string& tags, + const std::string& keys); +void InitializeProducer(); +void ShutdownAndDestroyProducer(); void InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions); @@ -69,7 +115,8 @@ void rocketmq_consumer_receive( const std::string& consumerName, const std::string& nameServer, const std::vector& subscriptions); +////////////////////////////////////////////////////// -////////////////////////////////////////////////////// -#endif -////////////////////////////////////////////////////// +#endif // __cplusplus + +#endif // SIMPLEPRODUCER_H \ No newline at end of file