最多只记录一次原始数据

cfg_parse/SimpleProducer.cpp

@@ -43,10 +43,6 @@
 
 #include "../log4cplus/log4.h"//lnk添加log4
 
-//添加redis stream相关头文件
-#include "../redisstream/RedisStreamMQ.h"
-#include <QSettings>
-
 using namespace std;
 
 static std::once_flag g_consumer_once;
@@ -67,11 +63,6 @@ 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;
 
@@ -89,9 +80,10 @@ 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;
@@ -110,23 +102,11 @@ int64_t G_APP_START_MS = []() -> int64_t {
 
 int64_t G_START_SKEW_MS = 1000;
 
-bool RocketMQ_should_process_after_start(const rocketmq::MQMessageExt& msg)
+bool should_process_after_start(const rocketmq::MQMessageExt& msg)
 {
     const int64_t born_ts = static_cast<int64_t>(msg.getBornTimestamp());
     return born_ts >= (G_APP_START_MS - G_START_SKEW_MS);
 }
-
-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;
 
@@ -439,7 +419,7 @@ void RocketMQConsumer::setConsumerMessageModel(const std::string& topic)
 RocketMQConsumer* g_consumer = NULL;
 
 // 初始化消费者函数
-void RocketMQ_InitializeConsumer(
+void InitializeConsumer(
     const std::string& consumerName,
     const std::string& nameServer,
     const std::vector<Subscription>& subscriptions) // 接收多个订阅
@@ -466,7 +446,7 @@ void RocketMQ_InitializeConsumer(
 }
 
 // 关闭并销毁消费者函数
-void RocketMQ_ShutdownAndDestroyConsumer()
+void ShutdownAndDestroyConsumer()
 {
     if (g_consumer != NULL) {
         delete g_consumer;
@@ -475,7 +455,7 @@ void RocketMQ_ShutdownAndDestroyConsumer()
 }
 
 // 消费消息的接口函数
-void RocketMQ_rocketmq_consumer_receive(
+void rocketmq_consumer_receive(
     const std::string& consumerName,
     const std::string& nameServer,
     const std::vector<Subscription>& subscriptions) // 接收多个订阅
@@ -864,7 +844,7 @@ private:
 RocketMQProducer* g_producer = NULL;
 
 // 初始化生产者（在程序启动时调用一次）
-void RocketMQ_InitializeProducer()
+void InitializeProducer()
 {
     if (g_producer == NULL) {
         try {
@@ -879,7 +859,7 @@ void RocketMQ_InitializeProducer()
 }
 
 // 关闭并销毁生产者（在程序结束时调用一次）
-void RocketMQ_ShutdownAndDestroyProducer()
+void ShutdownAndDestroyProducer()
 {
     if (g_producer != NULL) {
         delete g_producer;
@@ -888,7 +868,7 @@ void RocketMQ_ShutdownAndDestroyProducer()
 }
 
 // 发送消息的接口函数
-void RocketMQ_rocketmq_producer_send(const std::string& body,
+void rocketmq_producer_send(const std::string& body,
                             const std::string& topic,
                             const std::string& tags,
                             const std::string& keys)
@@ -907,102 +887,7 @@ void RocketMQ_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,9 +376,6 @@ 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;
 
@@ -652,20 +649,6 @@ 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_rocketmq_producer_send(const char* strbody,const char* topic);
+//void rocketmq_producer_send(const char* strbody,const char* topic);
 //void rocketmq_StartSendMessage(CProducer* producer,const char* strbody,const char* topic);
-void RocketMQ_rocketmq_producer_send(const std::string& body,
+void 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 RocketMQ_InitializeProducer();
+void InitializeProducer();
-void RocketMQ_ShutdownAndDestroyProducer();
+void ShutdownAndDestroyProducer();
 //////////////////////////////////////////////////////消费者
 typedef ConsumeStatus (*MessageCallBack)(
     const MQMessageExt& msg
@@ -59,13 +59,13 @@ struct Subscription {
                  MessageCallBack cb)
         : topic(t), tag(tg), callback(cb) {}
 };
-//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, const std::string& nameServer, const char* topic, const char* tag, const std::string& key);
-void RocketMQ_InitializeConsumer(const std::string& consumerName,
+void InitializeConsumer(const std::string& consumerName,
                         const std::string& nameServer,
                         const std::vector<Subscription>& subscriptions);
-void RocketMQ_ShutdownAndDestroyConsumer();
+void ShutdownAndDestroyConsumer();
 
-void RocketMQ_rocketmq_consumer_receive(
+void rocketmq_consumer_receive(
     const std::string& consumerName,
     const std::string& nameServer,
     const std::vector<Subscription>& subscriptions);

mms/db_interface.h

@@ -42,11 +42,6 @@
 #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,13 +2,7 @@ TEMPLATE = app
 TARGET = pt61850netd_pqfe
 DEPENDPATH += .
 
-INCLUDEPATH += .    ./source/include  \
+INCLUDEPATH += .    ./source/include  ./source/include/mmslite ./source/include/pg_inst ./source/include/curl ./source/include/oss_sdk ./source/include/roketmq 
-					./source/include/mmslite \
-					./source/include/pg_inst \
-					./source/include/curl \
-					./source/include/oss_sdk \
-					./source/include/roketmq \
-					./source/redisstream
 
 
 QMAKE_ORIG_TARGET = $(TARGET)
@@ -20,8 +14,6 @@ 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")
@@ -74,7 +66,6 @@ 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
@@ -107,8 +98,6 @@ unix {
    LIBS +=  -lrdkafka++
    LIBS +=  -lhttprun	
    LIBS +=  -llog4cplus
-   LIBS += -lhiredis
-   LIBS += /FeProject/lib/libz.a
 }
 
 #install
@@ -138,8 +127,7 @@ 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 \ 
@@ -175,5 +163,4 @@ 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

@@ -1,46 +0,0 @@
-#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

@@ -1,82 +0,0 @@
-#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,44 +17,18 @@
 #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;
 
-void RocketMQ_rocketmq_producer_send(const std::string& body,
+/*添加测试函数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,
                             const std::string& topic,
                             const std::string& tags,
                             const std::string& keys);
-
 extern "C" {
 void rocketmq_test_rt();
 void rocketmq_test_ud();
@@ -62,18 +36,15 @@ 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
 );
@@ -88,24 +59,7 @@ 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);
@@ -115,8 +69,7 @@ void rocketmq_consumer_receive(
     const std::string& consumerName,
     const std::string& nameServer,
     const std::vector<Subscription>& subscriptions);
+
+//////////////////////////////////////////////////////
+#endif
 //////////////////////////////////////////////////////
-
-#endif  // __cplusplus
-
-#endif  // SIMPLEPRODUCER_H