完成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 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

@@ -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 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/create_json.cpp

@@ -32,12 +32,10 @@
 #include <QMutexLocker>
 #include <QMapIterator>
 #include <QStringList>
-#include <QByteArray>
-#include <QList>
 
-//全局追踪表：mp_id -> rpt_no -> remaining times
+// ★MOD: 全局追踪表：mp_id -> remaining times
 static QMutex g_trace_mutex;
-static QHash<QString, QHash<int, int> > g_trace_map;
+static QHash<QString, int> g_trace_map;
 
 ///////////////////////////////////////////////////lnk2024-10-21////////////////////////////////////////////////////////
 extern void SendJsonAPI_web(const std::string& strUrl, const char* code, const std::string& json, char** ptr);
@@ -260,77 +258,36 @@ static QString escape_json_string(const QString& s)
     return out;
 }
 
-// 打开追踪：每个报告各追踪一次
+// 打开追踪：次数 times（比如 5）
 void process_trace_command(const std::string& id, int times)
 {
-    (void)times;
+    if (times <= 0) return;
     QString qid = QString::fromStdString(id).trimmed();
     if (qid.isEmpty()) return;
 
-    QList<int> rpt_nos;
-
-    QByteArray qid_bytes = qid.toLocal8Bit();
-
-    pthread_mutex_lock(&mtx);
-    LD_info_t *ld_info = find_LD_info_only_from_mp_id(qid_bytes.data());
-    if (ld_info != NULL && ld_info->rptinfo != NULL && ld_info->rptcount > 0) {
-        for (int i = 0; i < ld_info->rptcount; ++i) {
-            if (ld_info->rptinfo[i] == NULL)
-                continue;
-
-            int rpt_no = ld_info->rptinfo[i]->rptNo;
-            if (!rpt_nos.contains(rpt_no))
-                rpt_nos.append(rpt_no);
-        }
-    }
-    pthread_mutex_unlock(&mtx);
-
-    if (rpt_nos.isEmpty()) {
-        cout << "[TRACE] mp_id=" << id << " has no report info" << endl;
-        return;
-    }
-
     QMutexLocker lk(&g_trace_mutex);
-    QHash<int, int>& rpt_map = g_trace_map[qid];
+    g_trace_map[qid] = times; // 重新打开就覆盖/重置次数
-    for (int i = 0; i < rpt_nos.size(); ++i) {
-        int rpt_no = rpt_nos.at(i);
-        rpt_map[rpt_no] = 1;
-        cout << "[TRACE] mp_id=" << id << " add rpt_no=" << rpt_no
-             << " left=" << rpt_map.value(rpt_no) << endl;
-    }
 }
 
 // 查询是否要追踪
-static bool trace_is_enabled(const QString& mp_id, int rpt_no)
+static bool trace_is_enabled(const QString& mp_id)
 {
     QMutexLocker lk(&g_trace_mutex);
-    QHash<QString, QHash<int, int> >::const_iterator mp_it = g_trace_map.constFind(mp_id);
+    auto it = g_trace_map.constFind(mp_id);
-    if (mp_it == g_trace_map.constEnd())
+    return (it != g_trace_map.constEnd() && it.value() > 0);
-        return false;
-
-    QHash<int, int>::const_iterator rpt_it = mp_it.value().constFind(rpt_no);
-    return (rpt_it != mp_it.value().constEnd() && rpt_it.value() > 0);
 }
 
 // 命中一次并扣减；扣到 0 自动删
-static void trace_hit_and_decrement(const QString& mp_id, int rpt_no)
+static void trace_hit_and_decrement(const QString& mp_id)
 {
     QMutexLocker lk(&g_trace_mutex);
-    QHash<QString, QHash<int, int> >::iterator mp_it = g_trace_map.find(mp_id);
+    auto it = g_trace_map.find(mp_id);
-    if (mp_it == g_trace_map.end()) return;
+    if (it == g_trace_map.end()) return;
 
-    QHash<int, int>::iterator rpt_it = mp_it.value().find(rpt_no);
+    int left = it.value();
-    if (rpt_it == mp_it.value().end()) return;
-
-    int left = rpt_it.value();
     left -= 1;
-    if (left <= 0)
+    if (left <= 0) g_trace_map.erase(it);
-        mp_it.value().erase(rpt_it);
+    else it.value() = left;
-    else
-        rpt_it.value() = left;
-
-    if (mp_it.value().isEmpty())
-        g_trace_map.erase(mp_it);
 }
 
 //追踪61850原始数据
@@ -359,20 +316,15 @@ static QString build_mms_json_object(const json_block_data* data)
     return json;
 }
 
-static QString build_trace_json(const json_block_data* data, const char *v_wiring_type, const char *rpt_id, int rpt_no)
+static QString build_trace_json(const json_block_data* data)
 {
     if (!data) return "{}";
 
     QString mms_json = build_mms_json_object(data);
-    QString wiring_type = v_wiring_type ? QString::fromLocal8Bit(v_wiring_type) : "";
-    QString rpt = rpt_id ? QString::fromLocal8Bit(rpt_id) : "";
 
     QString json;
     json += "{";
     json += QString("\"mp_id\":\"%1\",").arg(escape_json_string(data->mp_id));
-    json += QString("\"v_wiring_type\":\"%1\",").arg(escape_json_string(wiring_type));
-    json += QString("\"rpt_id\":\"%1\",").arg(escape_json_string(rpt));
-    json += QString("\"rpt_no\":%1,").arg(rpt_no);
     json += QString("\"func_type\":%1,").arg(data->func_type);
     json += QString("\"data_time\":%1,").arg(QString::number((qlonglong)data->time));
     json += QString("\"voltage_level\":\"%1\",").arg(QString::number(data->voltage_level, 'f', 6));
@@ -383,13 +335,13 @@ static QString build_trace_json(const json_block_data* data, const char *v_wirin
     return json;
 }
 
-static void send_trace_if_needed(json_block_data* pdata, const char *v_wiring_type, const char *rpt_id, int rpt_no)
+static void send_trace_if_needed(json_block_data* pdata)
 {
     const QString mp_id_q = pdata->mp_id;
-    if (trace_is_enabled(mp_id_q, rpt_no)) {
+    if (trace_is_enabled(mp_id_q)) {
 
         // 1) 组 json
-        QString jsonText = build_trace_json(pdata, v_wiring_type, rpt_id, rpt_no);
+        QString jsonText = build_trace_json(pdata);
 
         // 2) 组 KafkaData
         Ckafka_data_t KafkaData;
@@ -403,25 +355,9 @@ static void send_trace_if_needed(json_block_data* pdata, const char *v_wiring_ty
         kafka_data_list_mutex.unlock();
 
         // 3) 次数 -1
-        trace_hit_and_decrement(mp_id_q, rpt_no);
+        trace_hit_and_decrement(mp_id_q);
     }
 }
-
-int trace_json_block_data(char v_wiring_type[], json_block_data *data, const char *rpt_id, int rpt_no)
-{
-    send_trace_if_needed(data, v_wiring_type, rpt_id, rpt_no);
-    return 1;
-}
-
-int trace_json_is_enabled(const char *mp_id, int rpt_no)
-{
-    QString mp_id_q = mp_id ? QString::fromLocal8Bit(mp_id).trimmed() : "";
-    if (mp_id_q.isEmpty())
-        return 0;
-
-    return trace_is_enabled(mp_id_q, rpt_no) ? 1 : 0;
-}
-
 ////////////////////////////////////////////////////////////////////////////////////////////lnk20250710添加频率值存储
 struct mp_freq_save {
     double G_FREQ;
@@ -1226,12 +1162,9 @@ void printCTopicList(const std::list<CTopic*>& ctopic_list) {
                                    val->fValue);
                         }
                     }
-					break; // 只打印第一个 Item 的 SequenceList 和 DataValueList，避免输出过多
                 }
-				break; // 只打印第一个 Monitor 的 ItemList，避免输出过多
             }
 
-
             // 如果需要打印 SOEList，可加以下：
             /*
             int soeIndex = 0;
@@ -1284,6 +1217,9 @@ int transfer_json_block_data(char v_wiring_type[], json_block_data *data) //json
         print_mms_str_map(data);
     }
 
+	//数据追踪上送
+	send_trace_if_needed(data);
+
 	list<CTopic*> ctopic_list;
 
 	////lnk2024-8-15 区分星型,角型接线
@@ -3827,3 +3763,4 @@ void Set_xml_nodeinfo_one(char* dev_type)
 	}
 		
 }
+

json/mms_json_inter.h

@@ -74,8 +74,6 @@ public:
 };
 
 int  transfer_json_block_data(char v_wiring_type[], json_block_data* data);//lnk2024-8-16添加参数
-int  trace_json_is_enabled(const char *mp_id, int rpt_no);
-int  trace_json_block_data(char v_wiring_type[], json_block_data* data, const char *rpt_id, int rpt_no);
 
 #endif /* __cplusplus */
 
@@ -97,4 +95,4 @@ void connectlog_pgsql(char* id,char* datetime,int status);
 #endif
 
 
-#endif //MMS_JSON_INTER_92327hyhy0923r_H
+#endif //MMS_JSON_INTER_92327hyhy0923r_H

json/save2json.cpp

@@ -198,7 +198,6 @@ static QMap<int, QMap<int, QList<long long>>> real_data_report_map;	//多个监
 static QMap<QString, json_block_data*> json_data_map;//CZY 2023-08-17 ww 2023年3月13日17:23:17扩展Map，用于保存各条线路的数据
 static QMap<QString, json_block_data*> json_flicker_data_map;//CZY 2023-09-11 展Map，用于保存各条线路的闪变数据
 static QMap<QString, json_block_data*> json_pst_data_map;//CZY 2023-09-11 展Map，用于保存各条线路的闪变数据
-static QMap<QString, json_block_data*> json_trace_data_map;
 
 //////////////////////////////////////////////////////////////////////////////lnk20260310文件控制
 pthread_mutex_t g_file_req_mutex = PTHREAD_MUTEX_INITIALIZER;
@@ -1540,7 +1539,7 @@ int parse_log(const std::string& json_str) {
 		}
 		else if((level == "measurepoint") && (grade == "TRACE") && (!id.empty() && !is_blank(id))){	//数据追踪
 			//打开监测点数据追踪开关
-			process_trace_command(id,1);	//每个报告各追踪1次
+			process_trace_command(id,3);	//3表示追踪次数
 		}
 		else{
 			std::cout << "type doesnt match" <<std::endl;
@@ -3634,119 +3633,6 @@ int json_block_create_data(char monid_char[], char* mms_str, double v, int flick
 	return TRUE;
 }
 
-static QString json_trace_block_key(char monid_char[], int flicker_flag, int rpt_no)
-{
-	QString key;
-	key.append(monid_char ? monid_char : "");
-	key.append("|");
-	key.append(QString::number(flicker_flag));
-	key.append("|");
-	key.append(QString::number(rpt_no));
-	return key;
-}
-
-static json_block_data* get_json_trace_block_data(char monid_char[], int flicker_flag, int rpt_no)
-{
-	QString key = json_trace_block_key(monid_char, flicker_flag, rpt_no);
-	if (!json_trace_data_map.contains(key))
-		return NULL;
-	return json_trace_data_map.value(key);
-}
-
-static void init_json_trace_block_data(char monid_char[], char voltage_level[], int flicker_flag, int rpt_no)
-{
-	QString key = json_trace_block_key(monid_char, flicker_flag, rpt_no);
-	json_block_data* pdata = NULL;
-	if (!json_trace_data_map.contains(key)) {
-		pdata = new json_block_data();
-		json_trace_data_map.insert(key, pdata);
-	}
-
-	pdata = json_trace_data_map.value(key);
-	pdata->monitorId = -1;
-	pdata->func_type = g_node_id;
-	pdata->flag = 0;
-	pdata->time = 0;
-	pdata->voltage_level = get_voltage_level(voltage_level);
-	pdata->mp_id = monid_char ? QString::fromLocal8Bit(monid_char) : QString("not define");
-	pdata->dev_type.clear();
-	pdata->mms_str_map.clear();
-	pdata->data_have_statistic = 0;
-}
-
-int json_trace_block_create_start(char voltage_level[], char monid_char[], int flicker_flag, char temcode[], int line_id, char v_wiring_type[], char rpt_id[], int rpt_no)
-{
-	(void)v_wiring_type;
-	(void)rpt_id;
-	if (!trace_json_is_enabled(monid_char, rpt_no))
-		return 0;
-
-	try_start_kafka_thread();
-	init_json_trace_block_data(monid_char, voltage_level, flicker_flag, rpt_no);
-
-	json_block_data* pdata = get_json_trace_block_data(monid_char, flicker_flag, rpt_no);
-	if (pdata != NULL) {
-		pdata->dev_type.append(temcode ? temcode : "");
-		pdata->monitorId = line_id;
-	}
-
-	return TRUE;
-}
-
-int json_trace_block_create_time(char monid_char[], long long Time, int flicker_flag, char rpt_id[], int rpt_no)
-{
-	(void)rpt_id;
-	json_block_data* pdata = get_json_trace_block_data(monid_char, flicker_flag, rpt_no);
-	if (pdata == NULL)
-		return 0;
-
-	pdata->time = Time;
-	return TRUE;
-}
-
-int json_trace_block_create_flag(char monid_char[], int flag, int flicker_flag, char rpt_id[], int rpt_no)
-{
-	(void)rpt_id;
-	json_block_data* pdata = get_json_trace_block_data(monid_char, flicker_flag, rpt_no);
-	if (pdata == NULL)
-		return 0;
-
-	pdata->flag = flag;
-	return TRUE;
-}
-
-int json_trace_block_create_data(char monid_char[], char* mms_str, double v, int flicker_flag, char rpt_id[], int rpt_no)
-{
-	(void)rpt_id;
-	json_block_data* pdata = get_json_trace_block_data(monid_char, flicker_flag, rpt_no);
-	if (pdata == NULL)
-		return 0;
-
-	pdata->mms_str_map.insert(QString::fromAscii(mms_str), v);
-	return TRUE;
-}
-
-int json_trace_block_create_end(char v_wiring_type[], char monid_char[], int flicker_flag, char rpt_id[], int rpt_no)
-{
-	QString key = json_trace_block_key(monid_char, flicker_flag, rpt_no);
-	if (!json_trace_data_map.contains(key))
-		return 1;
-
-	json_block_data* pdata = json_trace_data_map.value(key);
-	if (pdata == NULL) {
-		json_trace_data_map.remove(key);
-		return 1;
-	}
-
-	int ret = 1;
-	if (pdata->mms_str_map.count() > 0)
-		ret = trace_json_block_data(v_wiring_type, pdata, rpt_id, rpt_no);
-
-	delete pdata;
-	json_trace_data_map.remove(key);
-	return ret;
-}
-
 
 //lnk2024-8-16添加接线参数
 int json_block_create_end(char v_wiring_type[], char monid_char[], int flicker_flag)//WW 2023年3月13日16:38:41 多ICD修改
@@ -3807,7 +3693,7 @@ int json_block_create_end(char v_wiring_type[], char monid_char[], int flicker_f
 			json_pst_data_map.remove(monid_char);
 
 		}
-		printf("---------- json_block_create_end: mp_id= %s pdata->mms_str_map.count() == 0 ----------\n", monid_char);
+		printf("---------- json_block_create_end: pdata->mms_str_map.count() == 0 ----------\n");
 		return 1;
 	}
 	//lnk2024-8-16添加接线参数
@@ -3992,4 +3878,4 @@ int sel_mvl_type_ctrl_flag(char doname[])
 		return -1;
 	}
 }
-//ZW 2024-01-31 end
+//ZW 2024-01-31 end

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 */
@@ -74,12 +79,6 @@ int json_block_create_data(char monid_char[], char* mms_str , double v, int flic
 //lnk2024-8-16添加参数
 int json_block_create_end(char v_wiring_type[], char monid_char[], int flicker_flag); //CZY 2023-08-17 测试
 
-int json_trace_block_create_start(char voltage_level[], char monid_char[], int flicker_flag, char temcode[], int line_id, char v_wiring_type[], char rpt_id[], int rpt_no);
-int json_trace_block_create_time(char monid_char[], long long Time, int flicker_flag, char rpt_id[], int rpt_no);
-int json_trace_block_create_flag(char monid_char[], int flag, int flicker_flag, char rpt_id[], int rpt_no);
-int json_trace_block_create_data(char monid_char[], char* mms_str, double v, int flicker_flag, char rpt_id[], int rpt_no);
-int json_trace_block_create_end(char v_wiring_type[], char monid_char[], int flicker_flag, char rpt_id[], int rpt_no);
-
 //zw 2024-01-31 补招模式优化
 void add_mvl_type_ctrl(char doname[], int ctrl);
 int sel_mvl_type_ctrl_flag(char doname[]);
@@ -200,4 +199,4 @@ typedef struct file_dir_req_t
 #endif
 
 
-#endif //DB_INTERFACE_7ew2327hyhy0923r_H
+#endif //DB_INTERFACE_7ew2327hyhy0923r_H

mms/mms_process.c

@@ -387,7 +387,7 @@ void ChannelCheckIECReports(chnl_usr_t *chnl_usr)
 			rptinfo = LD_info->rptinfo[rpt_no] ;
 
 			//检查是否需要注册或取消注册报告，或不做任何处理
-			if(DEBUGOPEN)printf("[RPT][CHECK] ip=%s cpu=%d rpt_no=%d rptcount=%d LD_name=%s rptinfo=%p\n",
+			printf("[RPT][CHECK] ip=%s cpu=%d rpt_no=%d rptcount=%d LD_name=%s rptinfo=%p\n",
 			       chnl_usr->ip_str,
 			       cpuno,
 			       rpt_no,

mms/mmscli_rpt.c

@@ -993,7 +993,6 @@ ST_VOID u_iec_rpt_ind_data_by_devtype(MVL_VAR_ASSOC** info_va,
 			ied = find_ied_from_dev_code(LD_info->terminal_code);
 
 			ied_usr_t* ied_usr = (ied_usr_t*)ied->usr_ext;
-			json_trace_block_create_start(LD_info->voltage_level, LD_info->mp_id, rptinfo->flickerflag, ied_usr->dev_type, LD_info->line_id, LD_info->v_wiring_type, rcb_info->RptID, rptinfo->rptNo);
 
 			if (rptinfo->flickerflag==1)//CZY 2023-08-17 WW 2022-11-14 只有闪变和第一次进入才会初始化 json_block_create_start(LD_info->line_id);
 				json_block_create_start( LD_info->voltage_level, LD_info->mp_id, rptinfo->flickerflag, ied_usr->dev_type, LD_info->line_id);
@@ -1058,7 +1057,6 @@ ST_VOID u_iec_rpt_ind_data_by_devtype(MVL_VAR_ASSOC** info_va,
 						}
 						else {
 							json_block_create_flag(LD_info->mp_id, flag, rptinfo->flickerflag);
-							json_trace_block_create_flag(LD_info->mp_id, flag, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 							not_set_rpt_q_this = FALSE;
 						}
 					}
@@ -1078,7 +1076,6 @@ ST_VOID u_iec_rpt_ind_data_by_devtype(MVL_VAR_ASSOC** info_va,
 						}
 						else {
 							json_block_create_time(LD_info->mp_id, t / 1000, rptinfo->flickerflag);
-							json_trace_block_create_time(LD_info->mp_id, t / 1000, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 							printf("rcb_info->RptID=%s ,LineId=%d , Timestamp= %lld \n", rcb_info->RptID, LD_info->line_id, t / 1000);
 							not_set_rpt_TimeID_this = FALSE;
 							/*if (strstr(rcb_info->RptID, "LLN0$RP$urcbRealData")) {//lnk 20250624
@@ -1111,10 +1108,8 @@ ST_VOID u_iec_rpt_ind_data_by_devtype(MVL_VAR_ASSOC** info_va,
 					else if (strstr(rcb_info->RptID, "RDRE")) {//CZY 2023-08-17 WW 2022-11-14 修改判断LLN0$BR$brcbRDRE
 						processRDRE_data(LD_info, FULL_FCDA_Name, v);
 					}
-					else {
+					else
 						json_block_create_data(LD_info->mp_id, FULL_FCDA_Name, v, rptinfo->flickerflag);
-						json_trace_block_create_data(LD_info->mp_id, FULL_FCDA_Name, v, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
-					}
 				}//else			   	
 			}
 		}
@@ -1131,7 +1126,6 @@ ST_VOID u_iec_rpt_ind_data_by_devtype(MVL_VAR_ASSOC** info_va,
 		}
 		else {
 			printf("%d : %d", LD_info->rptRecvFlag, LD_info->rptRecvCheckFlag);
-			json_trace_block_create_end(LD_info->v_wiring_type, LD_info->mp_id, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 			//append_db_records(RPT_IDX);
 			if (rptinfo->flickerflag==1)//CZY 2023-08-17 WW 2022-11-14 增加闪变标志
 			{
@@ -1382,7 +1376,6 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 			ied = find_ied_from_dev_code(LD_info->terminal_code);
 
 			ied_usr_t* ied_usr = (ied_usr_t*)ied->usr_ext;
-			json_trace_block_create_start(LD_info->voltage_level, LD_info->mp_id, rptinfo->flickerflag, ied_usr->dev_type, LD_info->line_id, LD_info->v_wiring_type, rcb_info->RptID, rptinfo->rptNo);
 			if (rptinfo->flickerflag == 1)//CZY 2023-08-17 WW 2022-11-14 只有闪变和第一次进入才会初始化 json_block_create_start(LD_info->line_id);
 				json_block_create_start(LD_info->voltage_level, LD_info->mp_id, rptinfo->flickerflag, ied_usr->dev_type, LD_info->line_id);
 			else if (rptinfo->flickerflag == 0) {
@@ -1468,7 +1461,6 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 						}
 						else {
 							json_block_create_flag(LD_info->mp_id, flag, rptinfo->flickerflag);
-							json_trace_block_create_flag(LD_info->mp_id, flag, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 							not_set_rpt_q_this = FALSE;
 						}
 					}
@@ -1492,7 +1484,6 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 						}
 						else {
 							json_block_create_time(LD_info->mp_id, t / 1000, rptinfo->flickerflag);
-							json_trace_block_create_time(LD_info->mp_id, t / 1000, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 							printf("rcb_info->RptID=%s ,LineId=%d , Timestamp= %lld \n", rcb_info->RptID, LD_info->line_id, t / 1000);
 							not_set_rpt_TimeID_this = FALSE;
 							//if (strstr(rcb_info->RptID, "LLN0$RP$urcbRealData")) {
@@ -1506,7 +1497,6 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 									printf("rtdata RptID match");
 									if (urcbRealDataHasReceived(ied_usr->dev_idx,rpt_no,LD_info, t / 1000)){//判断时间重复
 										printf("this rt report Time repeats");
-										//json_trace_block_create_end(LD_info->v_wiring_type, LD_info->mp_id, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 										return;
 									}
 								}								
@@ -1541,10 +1531,8 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 						ied_usr_t* ied_usr = GET_IEDEXT_ADDR(ied);
 						processGGIO_start_data_end(LD_info->mp_id, FULL_FCDA_Name, v, time, ied_usr->dev_type, LD_info->line_id);//GGIO数据全套处理流程
 					}
-					else {
+					else
 						json_block_create_data(LD_info->mp_id, FULL_FCDA_Name, v, rptinfo->flickerflag);
-						json_trace_block_create_data(LD_info->mp_id, FULL_FCDA_Name, v, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
-					}
 				}//else			   	
 			}
 
@@ -1562,7 +1550,6 @@ ST_VOID u_iec_rpt_ind_data(MVL_VAR_ASSOC** info_va,
 		}
 		else {
 			printf("%d : %d", LD_info->rptRecvFlag, LD_info->rptRecvCheckFlag);
-			json_trace_block_create_end(LD_info->v_wiring_type, LD_info->mp_id, rptinfo->flickerflag, rcb_info->RptID, rptinfo->rptNo);
 			//append_db_records(RPT_IDX);
 			if (rptinfo->flickerflag == 1)//CZY 2023-08-17 WW 2022-11-14 增加闪变标志
 			{

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();
@@ -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  // __cplusplus
-#endif
+
-//////////////////////////////////////////////////////
+#endif  // SIMPLEPRODUCER_H