初始化添加pqdif判断，新增pqdif进程

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

@@ -156,6 +156,7 @@ 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];
@@ -376,9 +377,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 +650,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());
@@ -877,6 +861,9 @@ 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;
+	}
 	
 }
 
@@ -3405,6 +3392,31 @@ 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) {
@@ -3423,6 +3435,7 @@ 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)
@@ -4102,6 +4115,13 @@ 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);
@@ -4207,19 +4227,20 @@ 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 对象以避免内存泄漏
-        	delete terminal_dev_map->value(key);
+			if (should_bind) delete terminal_dev_map->value(key);
 
         	// 移除旧的键值对
-        	terminal_dev_map->remove(key);
+			if (should_bind) terminal_dev_map->remove(key);
 
         	// 插入新的 terminal_dev 对象
-			if(atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0){//lnk20250210匹配进程号
+			if(should_bind){//lnk20250210匹配进程号
 				//调试用
 				std::cout<< "process num match" << std::endl;
         		terminal_dev_map->insert(key, dev);
@@ -4227,7 +4248,7 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
 			}//后续修改为只有进程号匹配上index才录入当前进程
     	} else {
         	// 插入新的 terminal_dev 对象
-			if(atoi(dev->processNo) == g_front_seg_index ||  g_front_seg_index == 0){//lnk20250210匹配进程号
+			if(should_bind){//lnk20250210匹配进程号
 				//调试用
 				std::cout<< "process num match" << std::endl;
         		terminal_dev_map->insert(key, dev);
@@ -7307,6 +7328,8 @@ 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,6 +229,8 @@ 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_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);

json/create_json.cpp

@@ -32,10 +32,12 @@
 #include <QMutexLocker>
 #include <QMapIterator>
 #include <QStringList>
+#include <QByteArray>
+#include <QList>
 
-// ★MOD: 全局追踪表：mp_id -> remaining times
+//全局追踪表：mp_id -> rpt_no -> remaining times
 static QMutex g_trace_mutex;
-static QHash<QString, int> g_trace_map;
+static QHash<QString, QHash<int, int> > g_trace_map;
 
 ///////////////////////////////////////////////////lnk2024-10-21////////////////////////////////////////////////////////
 extern void SendJsonAPI_web(const std::string& strUrl, const char* code, const std::string& json, char** ptr);
@@ -258,36 +260,77 @@ static QString escape_json_string(const QString& s)
     return out;
 }
 
-// 打开追踪：次数 times（比如 5）
+// 打开追踪：每个报告各追踪一次
 void process_trace_command(const std::string& id, int times)
 {
-    if (times <= 0) return;
+    (void)times;
     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);
-    g_trace_map[qid] = times; // 重新打开就覆盖/重置次数
+    QHash<int, int>& rpt_map = g_trace_map[qid];
+    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)
+static bool trace_is_enabled(const QString& mp_id, int rpt_no)
 {
     QMutexLocker lk(&g_trace_mutex);
-    auto it = g_trace_map.constFind(mp_id);
+    QHash<QString, QHash<int, int> >::const_iterator mp_it = g_trace_map.constFind(mp_id);
-    return (it != g_trace_map.constEnd() && it.value() > 0);
+    if (mp_it == g_trace_map.constEnd())
+        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)
+static void trace_hit_and_decrement(const QString& mp_id, int rpt_no)
 {
     QMutexLocker lk(&g_trace_mutex);
-    auto it = g_trace_map.find(mp_id);
+    QHash<QString, QHash<int, int> >::iterator mp_it = g_trace_map.find(mp_id);
-    if (it == g_trace_map.end()) return;
+    if (mp_it == g_trace_map.end()) return;
 
-    int left = it.value();
+    QHash<int, int>::iterator rpt_it = mp_it.value().find(rpt_no);
+    if (rpt_it == mp_it.value().end()) return;
+
+    int left = rpt_it.value();
     left -= 1;
-    if (left <= 0) g_trace_map.erase(it);
+    if (left <= 0)
-    else it.value() = left;
+        mp_it.value().erase(rpt_it);
+    else
+        rpt_it.value() = left;
+
+    if (mp_it.value().isEmpty())
+        g_trace_map.erase(mp_it);
 }
 
 //追踪61850原始数据
@@ -316,15 +359,20 @@ static QString build_mms_json_object(const json_block_data* data)
     return json;
 }
 
-static QString build_trace_json(const json_block_data* data)
+static QString build_trace_json(const json_block_data* data, const char *v_wiring_type, const char *rpt_id, int rpt_no)
 {
     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));
@@ -335,13 +383,13 @@ static QString build_trace_json(const json_block_data* data)
     return json;
 }
 
-static void send_trace_if_needed(json_block_data* pdata)
+static void send_trace_if_needed(json_block_data* pdata, const char *v_wiring_type, const char *rpt_id, int rpt_no)
 {
     const QString mp_id_q = pdata->mp_id;
-    if (trace_is_enabled(mp_id_q)) {
+    if (trace_is_enabled(mp_id_q, rpt_no)) {
 
         // 1) 组 json
-        QString jsonText = build_trace_json(pdata);
+        QString jsonText = build_trace_json(pdata, v_wiring_type, rpt_id, rpt_no);
 
         // 2) 组 KafkaData
         Ckafka_data_t KafkaData;
@@ -355,9 +403,25 @@ static void send_trace_if_needed(json_block_data* pdata)
         kafka_data_list_mutex.unlock();
 
         // 3) 次数 -1
-        trace_hit_and_decrement(mp_id_q);
+        trace_hit_and_decrement(mp_id_q, rpt_no);
     }
 }
+
+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;
@@ -1162,9 +1226,12 @@ void printCTopicList(const std::list<CTopic*>& ctopic_list) {
                                    val->fValue);
                         }
                     }
+					break; // 只打印第一个 Item 的 SequenceList 和 DataValueList，避免输出过多
                 }
+				break; // 只打印第一个 Monitor 的 ItemList，避免输出过多
             }
 
+
             // 如果需要打印 SOEList，可加以下：
             /*
             int soeIndex = 0;
@@ -1217,9 +1284,6 @@ 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 区分星型,角型接线
@@ -3763,4 +3827,3 @@ void Set_xml_nodeinfo_one(char* dev_type)
 	}
 		
 }
-

json/mms_json_inter.h

@@ -26,6 +26,7 @@
 #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拼接参数类
 {
@@ -74,6 +75,8 @@ 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 */
 

json/save2json.cpp

@@ -198,6 +198,7 @@ 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;
@@ -270,6 +271,8 @@ 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));
 	}
@@ -1539,7 +1542,7 @@ int parse_log(const std::string& json_str) {
 		}
 		else if((level == "measurepoint") && (grade == "TRACE") && (!id.empty() && !is_blank(id))){	//数据追踪
 			//打开监测点数据追踪开关
-			process_trace_command(id,3);	//3表示追踪次数
+			process_trace_command(id,1);	//每个报告各追踪1次
 		}
 		else{
 			std::cout << "type doesnt match" <<std::endl;
@@ -3633,6 +3636,119 @@ 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修改
@@ -3693,7 +3809,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: pdata->mms_str_map.count() == 0 ----------\n");
+		printf("---------- json_block_create_end: mp_id= %s pdata->mms_str_map.count() == 0 ----------\n", monid_char);
 		return 1;
 	}
 	//lnk2024-8-16添加接线参数

mms/db_interface.h

@@ -31,6 +31,7 @@
 #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
@@ -42,11 +43,6 @@
 #define SHOULD_UNREGISTER				2
 ///////////////////////////////////////////////////////////////////////////////
 
-#define MQ_BACKEND_ROCKETMQ      0
-#define MQ_BACKEND_REDIS_STREAM  1
-
-///////////////////////////////////////////////////////////////////////////////
-
 #ifdef __cplusplus
 extern "C" {
 #endif /* __cplusplus */
@@ -79,6 +75,12 @@ 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[]);

mms/main.c

@@ -81,6 +81,8 @@ 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

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

@@ -993,6 +993,7 @@ 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);
@@ -1057,6 +1058,7 @@ 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;
 						}
 					}
@@ -1076,6 +1078,7 @@ 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
@@ -1108,8 +1111,10 @@ 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			   	
 			}
 		}
@@ -1126,6 +1131,7 @@ 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 增加闪变标志
 			{
@@ -1376,6 +1382,7 @@ 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) {
@@ -1461,6 +1468,7 @@ 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;
 						}
 					}
@@ -1484,6 +1492,7 @@ 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")) {
@@ -1497,6 +1506,7 @@ 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;
 									}
 								}								
@@ -1531,8 +1541,10 @@ 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			   	
 			}
 
@@ -1550,6 +1562,7 @@ 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 增加闪变标志
 			{

mms/rdb_client.c

@@ -311,6 +311,9 @@ 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));
@@ -351,6 +354,9 @@ 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,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

set_process.sh

@@ -329,6 +329,7 @@ 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
 			
 			# 根据进程号添加对应进程配置
@@ -340,6 +341,7 @@ 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时会把要杀死的进程重启
@@ -353,6 +355,7 @@ 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