最多只记录一次原始数据

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 int G_MQ_BACKEND;
-extern std::string G_MQCONSUMER_ACCESSKEY;
+
-extern std::string G_MQCONSUMER_SECRETKEY;
+////////////////////////////////////////////////////////////////////////////////////////////////////////
-extern std::string G_MQCONSUMER_CHANNEL;
 
 // 前向声明 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

@@ -156,7 +156,6 @@ public:
     char dev_type[64];
     char dev_key[255];
     char dev_series[255];
-    char com_type[32];
     char addr_str[64];
     char port[64];
     char timestamp[64];
@@ -377,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;
 
@@ -650,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());
@@ -861,9 +877,6 @@ void init_config() {
 	else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
 		TEST_PORT = TEST_PORT + SOE_COMTRADE_BASE_NODE_ID +  g_front_seg_index;
 	}	
-	else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
-		TEST_PORT = TEST_PORT + PQDIF_DATA_BASE_NODE_ID +  g_front_seg_index;
-	}
 	
 }
 
@@ -3392,31 +3405,6 @@ void SendJsonAPI_web(const std::string& strUrl, const char* code, const std::str
 }
 
 // 打印 terminal_dev_map 中所有内容的函数
-static bool is_cfg_pqdif_process()
-{
-	return std::strstr(subdir, "cfg_pqdif_data") != NULL;
-}
-
-static bool is_pqdif_com_type(const char* com_type)
-{
-	QString type = com_type ? QString::fromLocal8Bit(com_type).trimmed() : QString();
-	return type.compare("pqdif", Qt::CaseInsensitive) == 0;
-}
-
-static bool terminal_should_bind_to_current_process(const terminal_dev* dev)
-{
-	if (dev == NULL)
-		return false;
-
-	if (is_pqdif_com_type(dev->com_type))
-		return is_cfg_pqdif_process();
-
-	if (is_cfg_pqdif_process())
-		return false;
-
-	return (atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0);
-}
-
 void printTerminalDevMap(const QMap<QString, terminal_dev*>& terminal_dev_map) {
 	QMap<QString, terminal_dev*>::const_iterator it;
     for (it = terminal_dev_map.constBegin(); it != terminal_dev_map.constEnd(); ++it) {
@@ -3435,7 +3423,6 @@ void printTerminalDevMap(const QMap<QString, terminal_dev*>& terminal_dev_map) {
                      << ", Device Type:" << QString(dev->dev_type)
                      << ", Device Key:" << QString(dev->dev_key)
                      << ", Device Series:" << QString(dev->dev_series)
-					 << ", Device comType:" << QString(dev->com_type)
 					 << ", Device processNo:" << QString(dev->processNo)
 					 << ", Device maxProcessNum:" << QString(dev->maxProcessNum)
                      << ", Address:" << QString(dev->addr_str)
@@ -4115,13 +4102,6 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
         else strncpy(dev->dev_series, "N/A", sizeof(dev->dev_series) - 1);
 		dev->dev_series[sizeof(dev->dev_series) - 1] = '\0';
 
-		cJSON* comType = cJSON_GetObjectItem(item, "comType");
-		if (comType == NULL)
-			comType = cJSON_GetObjectItem(item, "com_type");
-		if (comType && comType->type == cJSON_String) strncpy(dev->com_type, comType->valuestring, sizeof(dev->com_type) - 1);
-		else strncpy(dev->com_type, "mms", sizeof(dev->com_type) - 1);
-		dev->com_type[sizeof(dev->com_type) - 1] = '\0';
-
 		//lnk20250210台账进程号
 		cJSON* processNo = cJSON_GetObjectItem(item, "processNo"); // processNo转为字符串
         if (processNo && processNo->type == cJSON_Number) snprintf(dev->processNo, sizeof(dev->processNo), "%d", processNo->valueint);
@@ -4227,20 +4207,19 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
 
 		//lnk20260312防止内存泄漏
 		bool inserted = false;
-		bool should_bind = terminal_should_bind_to_current_process(dev);
 
     	// 检查是否存在重复键
     	if (terminal_dev_map->contains(key)) {
         	std::cerr << "Duplicate terminal_code found: " << key.toStdString() << std::endl;
 
         	// 删除旧的 terminal_dev 对象以避免内存泄漏
-			if (should_bind) delete terminal_dev_map->value(key);
+        	delete terminal_dev_map->value(key);
 
         	// 移除旧的键值对
-			if (should_bind) terminal_dev_map->remove(key);
+        	terminal_dev_map->remove(key);
 
         	// 插入新的 terminal_dev 对象
-			if(should_bind){//lnk20250210匹配进程号
+			if(atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0){//lnk20250210匹配进程号
 				//调试用
 				std::cout<< "process num match" << std::endl;
         		terminal_dev_map->insert(key, dev);
@@ -4248,7 +4227,7 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
 			}//后续修改为只有进程号匹配上index才录入当前进程
     	} else {
         	// 插入新的 terminal_dev 对象
-			if(should_bind){//lnk20250210匹配进程号
+			if(atoi(dev->processNo) == g_front_seg_index ||  g_front_seg_index == 0){//lnk20250210匹配进程号
 				//调试用
 				std::cout<< "process num match" << std::endl;
         		terminal_dev_map->insert(key, dev);
@@ -7328,8 +7307,6 @@ const char* get_front_msg_from_subdir() {
         return "实时数据进程";
     else if (std::strstr(subdir, "cfg_soe_comtrade") != NULL)
         return "暂态和告警进程";
-    else if (std::strstr(subdir, "cfg_pqdif_data") != NULL)
-        return "pqdif处理进程";
     else if (std::strstr(subdir, "cfg_recallhis_data") != NULL)
         return "稳态补招进程";
     else if (std::strstr(subdir, "cfg_stat_data") != NULL)

cfg_parse/log4.cpp

@@ -229,8 +229,6 @@ std::string get_front_type_from_subdir() {
         return "realTime";
     else if (std::strstr(subdir, "cfg_soe_comtrade") != NULL)
         return "comtrade";
-    else if (std::strstr(subdir, "cfg_pqdif_data") != NULL)
-        return "pqdif";
     else if (std::strstr(subdir, "cfg_recallhis_data") != NULL)
         return "recall";
     else if (std::strstr(subdir, "cfg_stat_data") != NULL)

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 RocketMQ_InitializeProducer();
-void ShutdownAndDestroyProducer();
+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 RocketMQ_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 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/mms_json_inter.h

@@ -26,7 +26,6 @@
 #define THREE_SECS_DATA_BASE_NODE_ID	200  // 三秒数据、实时数据
 #define SOE_COMTRADE_BASE_NODE_ID		300  // soe 录波
 #define HIS_DATA_BASE_NODE_ID			400  //历史、补招
-#define PQDIF_DATA_BASE_NODE_ID			800  // pqdif数据
 
 class json_block_data //json拼接参数类
 {

json/save2json.cpp

@@ -271,8 +271,6 @@ void add_comm_log(char* log_str)
 		com_log_fn += "comm_300_comtrade.txt";
 	else if (g_node_id == HIS_DATA_BASE_NODE_ID)
 		com_log_fn += "comm_400_his.txt";
-	else if (g_node_id == PQDIF_DATA_BASE_NODE_ID)
-		com_log_fn += "comm_800_pqdif.txt";
 	else if (g_node_id == NEW_HIS_DATA_BASE_NODE_ID) {
 		com_log_fn.append(QString("comm_400_his_%1.txt").arg(g_front_seg_index));
 	}

mms/db_interface.h

@@ -31,7 +31,6 @@
 #define NEW_HIS_DATA_BASE_NODE_ID		500
 #define RECALL_HIS_DATA_BASE_NODE_ID	600
 #define RECALL_ALL_DATA_BASE_NODE_ID	700
-#define PQDIF_DATA_BASE_NODE_ID			800
 
 #define REPORT_TYPE_STAT				1
 #define REPORT_TYPE_REAL				2
@@ -43,6 +42,11 @@
 #define SHOULD_UNREGISTER				2
 ///////////////////////////////////////////////////////////////////////////////
 
+#define MQ_BACKEND_ROCKETMQ      0
+#define MQ_BACKEND_REDIS_STREAM  1
+
+///////////////////////////////////////////////////////////////////////////////
+
 #ifdef __cplusplus
 extern "C" {
 #endif /* __cplusplus */

mms/main.c

@@ -81,8 +81,6 @@ void init_global_function_enable()
 		three_secs_enabled = 1;
 	}else if (strcmp(subdir,"cfg_soe_comtrade")==0) {	//告警和录波和暂态
 		g_node_id = SOE_COMTRADE_BASE_NODE_ID;
-	}else if (strcmp(subdir,"cfg_pqdif_data")==0) {
-		g_node_id = PQDIF_DATA_BASE_NODE_ID;
 	}else if (strcmp(subdir,"cfg_his_data")==0)  {		//不使用
 		g_node_id = RECALL_ALL_DATA_BASE_NODE_ID;
 		g_node_id = HIS_DATA_BASE_NODE_ID;	

mms/mms_process.c

@@ -1634,18 +1634,12 @@ void CheckAllConnectedChannel()
 			if(DEBUGOPEN)printf("[FILEDIR] enter HandleFileDirReqForChannel");
 			HandleFileDirReqForChannel(chnl_usr);//文件目录请求
 		}
-		if(g_node_id == PQDIF_DATA_BASE_NODE_ID){
-			//补招PQDIF文件
-		}
-		else{
-			ChannelCheckIECReports(chnl_usr);//报告
-		}
 
+		ChannelCheckIECReports(chnl_usr);//报告
 		if ( (g_node_id == SOE_COMTRADE_BASE_NODE_ID) || (g_node_id == HIS_DATA_BASE_NODE_ID) || (g_node_id == NEW_HIS_DATA_BASE_NODE_ID) || (g_node_id == RECALL_HIS_DATA_BASE_NODE_ID) || (g_node_id == RECALL_ALL_DATA_BASE_NODE_ID))
 			ChannelCheckWaveFiles(chnl_usr);//录波文件
 		if(g_node_id == HIS_DATA_BASE_NODE_ID || g_node_id == NEW_HIS_DATA_BASE_NODE_ID || g_node_id == RECALL_HIS_DATA_BASE_NODE_ID || (g_node_id == RECALL_ALL_DATA_BASE_NODE_ID))
 			ChannelCheckIECLogs(chnl_usr);//补招文件
-
 		if ( (sGetMsTime() - chnl_usr->m_LastPosRespTime) > 15*1000 ) //wait 15 secs，隔15秒获取一次响应，两次没响应后关闭
 		{	
 			char** varnames ;

mms/rdb_client.c

@@ -311,9 +311,6 @@ apr_status_t run_protocol()
 			else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
 				ServerPort = SOCKET_PORT + SOE_COMTRADE_BASE_NODE_ID +  g_front_seg_index;
 			}
-			else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
-				ServerPort = SOCKET_PORT + PQDIF_DATA_BASE_NODE_ID +  g_front_seg_index;
-			}
 
 			struct sockaddr_in server_sockaddr;
 			memset(&server_sockaddr, 0, sizeof(server_sockaddr));
@@ -354,9 +351,6 @@ apr_status_t run_protocol()
 			else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
 				HTTP_PORT = HTTP_PORT + SOE_COMTRADE_BASE_NODE_ID +  g_front_seg_index;
 			}
-			else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
-				HTTP_PORT = HTTP_PORT + PQDIF_DATA_BASE_NODE_ID +  g_front_seg_index;
-			}
 			printf("try_start_web_http_thread \n");
 			try_start_web_http_thread();
 			printf("try_start_http_thread \n");

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 RocketMQ_rocketmq_producer_send(const std::string& body,
-//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,
                             const std::string& topic,
                             const std::string& tags,
                             const std::string& keys);
+
 extern "C" {
 void rocketmq_test_rt();
 void rocketmq_test_ud();
@@ -38,13 +64,16 @@ void rocketmq_test_set();
 void rocketmq_test_only();
 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  // __cplusplus
-#endif
+
-//////////////////////////////////////////////////////
+#endif  // SIMPLEPRODUCER_H

set_process.sh

@@ -329,7 +329,6 @@ handle_reset() {
 			sed -i '/cfg_recallhis_data/d' /FeProject/etc/runtime.cf
 			sed -i '/cfg_3s_data/d' /FeProject/etc/runtime.cf
 			sed -i '/cfg_soe_comtrade/d' /FeProject/etc/runtime.cf
-			sed -i '/cfg_pqdif_data/d' /FeProject/etc/runtime.cf
 			sed -i '/fe_watchdog/d' /FeProject/etc/runtime.cf
 			
 			# 根据进程号添加对应进程配置
@@ -341,7 +340,6 @@ handle_reset() {
 				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_recallhis_data -s 1_1^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
 				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_3s_data^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
 				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_soe_comtrade^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
-				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_pqdif_data^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
 				
 			else
 				#看门狗固定放在第一个，防止stop时会把要杀死的进程重启
@@ -355,7 +353,6 @@ handle_reset() {
 				#以下部分没有多进程
 				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_3s_data^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
 				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_soe_comtrade^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
-				sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_pqdif_data^ ^ ^ 1 ^                ^\n')" /FeProject/etc/runtime.cf
 				
 			fi