完成redisstream前置生产和消费功能

cfg_parse/SimpleProducer.cpp

@@ -43,6 +43,10 @@
 
 #include "../log4cplus/log4.h"//lnk添加log4
 
+//添加redis stream相关头文件
+#include "../redisstream/RedisStreamMQ.h"
+#include <QSettings>
+
 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<int64_t>(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<Subscription>& 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<Subscription>& 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<Subscription>& 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<Subscription>& 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测试用

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());

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<Subscription>& 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<Subscription>& subscriptions);

json/create_json.cpp

@@ -1562,7 +1562,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;//不发送数据
+						}
 						//lnk20260127
 						if (typeofdata == false || data_have_static == false) {//不合并则处理完闪变就不处理其他数据,如果有统计数据或者数据类型区分了就继续处理其他数据
 							if (longjumpflag == true || shortjumpflag == true) {
@@ -2471,6 +2473,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;

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 */

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")
@@ -66,6 +74,7 @@ unix {
 	include(fe_common.pri)
 	SOURCES += source/mms/event2.c   
 	INCLUDEPATH += ./source/include/apr-linux
+	INCLUDEPATH += /FeProject/include
 	LIBS		+= -L/FeProject/lib
 	LIBS		+= -L/FeProject/lib/pgodbc
 	DEFINES += DEBUG_SISCO
@@ -98,6 +107,8 @@ unix {
    LIBS +=  -lrdkafka++
    LIBS +=  -lhttprun	
    LIBS +=  -llog4cplus
+   LIBS += -lhiredis
+   LIBS += /FeProject/lib/libz.a
 }
 
 #install
@@ -127,7 +138,8 @@ 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/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

redisstream/RedisStreamMQ.h

@@ -0,0 +1,46 @@
+#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 <string>
+#include <vector>
+
+// lnk20260626 修改：RedisStreamMQ.h 只需要知道 MQMessageExt 是一个类型即可
+// 这里用前向声明，避免 RedisStreamMQ.h 直接依赖 RocketMQ 头文件
+namespace rocketmq {
+class MQMessageExt;
+}
+
+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<Subscription>& subscriptions);
+
+void RedisStream_ShutdownAndDestroyConsumer();
+
+void RedisStream_consumer_receive(const std::string& consumerName,
+                                  const std::string& nameServer,
+                                  const std::vector<Subscription>& subscriptions);
+
+bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg);
+
+#endif // REDIS_STREAM_MQ_H

redisstream/mq_compat.h

@@ -0,0 +1,82 @@
+#ifndef MQ_COMPAT_H
+#define MQ_COMPAT_H
+
+#include <string>
+#include <stdint.h>
+
+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

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 <iostream>
 #include <string>
 
+/*
+ * 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<Subscription>& subscriptions);
+
+void RocketMQ_ShutdownAndDestroyConsumer();
+
+void RocketMQ_rocketmq_consumer_receive(
+    const std::string& consumerName,
+    const std::string& nameServer,
+    const std::vector<Subscription>& 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<Subscription>& subscriptions);
@@ -69,7 +115,8 @@ void rocketmq_consumer_receive(
     const std::string& consumerName,
     const std::string& nameServer,
     const std::vector<Subscription>& subscriptions);
+//////////////////////////////////////////////////////
 
-//////////////////////////////////////////////////////
-#endif
-//////////////////////////////////////////////////////
+#endif  // __cplusplus
+
+#endif  // SIMPLEPRODUCER_H