完成实时数据拆分

.vscode/settings.json

@@ -128,5 +128,6 @@
     "typeinfo": "cpp",
     "valarray": "cpp",
     "variant": "cpp"
-  }
+  },
+  "cmake.sourceDirectory": "D:/canneng/云前置移植项目/zw/Linux_Front1056/LFtid1056/lib/libuv-v1.51.0"
 }

LFtid1056/LFtid1056.vcxproj

@@ -147,7 +147,6 @@
     <ClInclude Include="cloudfront\code\log4cplus\internal\internal.h" />
     <ClInclude Include="cloudfront\code\log4cplus\internal\socket.h" />
     <ClInclude Include="cloudfront\code\log4cplus\layout.h" />
-    <ClInclude Include="cloudfront\code\log4cplus\log4.h" />
     <ClInclude Include="cloudfront\code\log4cplus\log4cplus.h" />
     <ClInclude Include="cloudfront\code\log4cplus\log4judpappender.h" />
     <ClInclude Include="cloudfront\code\log4cplus\logger.h" />

LFtid1056/LFtid1056.vcxproj.filters

@@ -169,9 +169,6 @@
     <ClInclude Include="cloudfront\code\nlohmann\thirdparty\hedley\hedley_undef.hpp">
       <Filter>cloudfront\code\nlohmann\thirdparty\hedley</Filter>
     </ClInclude>
-    <ClInclude Include="cloudfront\code\log4cplus\log4.h">
-      <Filter>cloudfront\code\log4cplus</Filter>
-    </ClInclude>
     <ClInclude Include="cloudfront\code\log4cplus\logger.h">
       <Filter>cloudfront\code\log4cplus</Filter>
     </ClInclude>

LFtid1056/PQSMsg.h

@@ -1768,9 +1768,7 @@ public:
         // 26-33: <20><>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ǰ8<C7B0><38>Ԫ<EFBFBD>أ<EFBFBD>
         for (int i = 0; i < 2; ++i) {
             for (int j = 0; j < 4; ++j) { // ȡÿ<C8A1>е<EFBFBD>ǰ4<C7B0><34>Ԫ<EFBFBD><D4AA>
-                if (i == 0 || j < 3) { // ȷ<><C8B7>ֻȡ8<C8A1><38>
-                    float_buffer.push_back(UI_Seq[i][j]);
-                }
+                float_buffer.push_back(UI_Seq[i][j]);
             }
         }
 
@@ -1888,9 +1886,7 @@ public:
         // 26-33: <20><>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ǰ8<C7B0><38>Ԫ<EFBFBD>أ<EFBFBD>
         for (int i = 0; i < 2; ++i) {
             for (int j = 0; j < 4; ++j) {
-                if (i == 0 || j < 3) {
-                    float_buffer.push_back(UI_Seq[i][j]);
-                }
+                float_buffer.push_back(UI_Seq[i][j]);
             }
         }
 
@@ -1965,7 +1961,47 @@ public:
         return floatVectorToBase64(float_buffer);
     }
 
-    // <20><><EFBFBD>ݽ<EFBFBD><EFBFBD>߷<EFBFBD>ʽѡ<EFBFBD><EFBFBD>ת<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    // <20><><EFBFBD>ͽ<EFBFBD><EFBFBD><EFBFBD>ת<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>-г<><D0B3><EFBFBD><EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>+<EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    std::string ConvertToBase64_Delta_RtHarmV() const {
+        std::vector<float> float_buffer;
+
+        //THD<48>ߵ<EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD>ʣ<EFBFBD>3-5<><35>
+        for (int i = 3; i < 6; ++i) {
+            float_buffer.push_back(THD[i]);
+        }
+
+        //<2F><>ѹг<D1B9><D0B3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʣ<EFBFBD>1-49<34>Σ<EFBFBD>
+        for (int i = 0; i < 3; ++i) {
+            for (int j = 1; j < HARMNUM; ++j) {
+                float_buffer.push_back(HARMV[i][j]);
+            }
+        }
+
+        // ת<><D7AA>ΪBase64
+        return floatVectorToBase64(float_buffer);
+    }
+
+    // <20><><EFBFBD>ͽ<EFBFBD><CDBD><EFBFBD>ת<EFBFBD><D7AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD>-г<><D0B3><EFBFBD><EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>+<2B><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    std::string ConvertToBase64_Star_RtHarmV() const {
+        std::vector<float> float_buffer;
+
+        //THD<48>ߵ<EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD>ʣ<EFBFBD>3-5<><35>
+        for (int i = 0; i < 3; ++i) {
+            float_buffer.push_back(THD[i]);
+        }
+
+        //<2F><>ѹг<D1B9><D0B3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʣ<EFBFBD>1-49<34>Σ<EFBFBD>
+        for (int i = 0; i < 3; ++i) {
+            for (int j = 1; j < HARMNUM; ++j) {
+                float_buffer.push_back(HARMV[i][j]);
+            }
+        }
+
+        // ת<><D7AA>ΪBase64
+        return floatVectorToBase64(float_buffer);
+    }
+
+    // <20><><EFBFBD>ݽ<EFBFBD><DDBD>߷<EFBFBD>ʽѡ<CABD><D1A1>ת<EFBFBD><D7AA><EFBFBD><EFBFBD><EFBFBD><EFBFBD> <20><><EFBFBD><EFBFBD>ȫ<EFBFBD><C8AB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
     std::string ConvertToBase64(int wiringType) const {
         // 1Ϊ<31><CEAA><EFBFBD>ͽ<EFBFBD><CDBD>ߣ<EFBFBD><DFA3><EFBFBD><EFBFBD><EFBFBD>Ϊ<EFBFBD><CEAA><EFBFBD>ͽ<EFBFBD><CDBD><EFBFBD>
         if (wiringType == 1) {
@@ -1975,6 +2011,47 @@ public:
             return ConvertToBase64_Star();
         }
     }
+
+    //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>-г<><D0B3><EFBFBD><EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>+<2B><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    std::string ConvertToBase64_RtHarmV(int wiringType) const {
+        // 1Ϊ<31><CEAA><EFBFBD>ͽ<EFBFBD><CDBD>ߣ<EFBFBD><DFA3><EFBFBD><EFBFBD><EFBFBD>Ϊ<EFBFBD><CEAA><EFBFBD>ͽ<EFBFBD><CDBD><EFBFBD>
+        if (wiringType == 1) {
+            return ConvertToBase64_Delta_RtHarmV();
+        }
+        else {
+            return ConvertToBase64_Star_RtHarmV();
+        }
+    }
+
+    //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>-г<><D0B3><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ֵ
+    std::string ConvertToBase64_RtHarmI() const {
+        std::vector<float> float_buffer;
+
+        //<2F><><EFBFBD><EFBFBD>г<EFBFBD><D0B3><EFBFBD><EFBFBD>ֵ<EFBFBD><D6B5>1-49<34>Σ<EFBFBD>
+        for (int i = 0; i < 3; ++i) {
+            for (int j = 1; j < HARMNUM; ++j) {
+                float_buffer.push_back(HARMI[i][j]);
+            }
+        }
+
+        // ת<><D7AA>ΪBase64
+        return floatVectorToBase64(float_buffer);
+    }
+
+    //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>-<2D><>г<EFBFBD><D0B3><EFBFBD><EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+    std::string ConvertToBase64_RtInHarmV() const {
+        std::vector<float> float_buffer;
+
+        //<2F><>ѹ<EFBFBD><D1B9>г<EFBFBD><D0B3><EFBFBD><EFBFBD>ֵ<EFBFBD><D6B5>0-49<34>Σ<EFBFBD>
+        for (int i = 0; i < 3; ++i) {
+            for (int j = 0; j < HARMNUM; ++j) {
+                float_buffer.push_back(INHARMV[i][j]);
+            }
+        }
+
+        // ת<><D7AA>ΪBase64
+        return floatVectorToBase64(float_buffer);
+    }
 };
 #pragma pack(pop)
 

LFtid1056/cloudfront/code/cfg_parser.cpp

@@ -125,6 +125,10 @@ std::string TOPIC_RTDATA = "";
 std::string G_ROCKETMQ_TAG = "";//tag
 std::string G_ROCKETMQ_KEY = "";//key
 
+//实时数据tagkey
+std::string G_RT_TAG = "";//tag
+std::string G_RT_KEY = "";//key
+
 //生产者
 std::string G_ROCKETMQ_PRODUCER = "";		//rocketmq producer
 std::string G_MQPRODUCER_IPPORT = "";		//rocketmq ip+port
@@ -272,6 +276,10 @@ void loadConfig(const std::string& filename) {
     strMap["Queue.QUEUE_TAG"]     = &G_ROCKETMQ_TAG;
     strMap["Queue.QUEUE_KEY"]     = &G_ROCKETMQ_KEY;
 
+    //添加rt的tagkey
+    strMap["Queue.RT_TAG"]        = &G_RT_TAG;
+    strMap["Queue.RT_KEY"]        = &G_RT_KEY;
+
     // [RocketMq] —— 生产者
     strMap["RocketMq.producer"]             = &G_ROCKETMQ_PRODUCER;
     strMap["RocketMq.Ipport"]               = &G_MQPRODUCER_IPPORT;
@@ -2827,6 +2835,9 @@ bool compare_qvvr_and_file(const std::string& cfg_path, const std::vector<qvvr_d
         return false;
     }
 
+    //打印提取到的时间戳
+    std::cout << "[调试] 提取到的起始时间戳: " << start_tm << ", 触发时间戳: " << trig_tm << "\n";
+
     // 遍历所有暂态事件，查找与 trig_tm 匹配的
     for (const auto& data : data_list) {
         long long diff = static_cast<long long>(data.QVVR_time) - trig_tm;
@@ -2919,6 +2930,8 @@ void upload_data_test(){
     data.strTopic = TOPIC_ALARM;
     data.strText = js;
     data.mp_id = "test";
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -2963,36 +2976,48 @@ std::vector<DeviceInfo> GenerateDeviceInfoFromLedger(const std::vector<terminal_
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////录波文件通知
 
-bool assign_qvvr_file_list(const std::string& id, ushort nCpuNo, const std::vector<std::string>& file_list_raw) {
+bool assign_qvvr_file_list(const std::string& id,
+                           ushort nCpuNo,
+                           const std::vector<std::string>& file_list_raw) {
+    // ★新增：台账加锁（若上层已有锁，可移除本行）
+    std::lock_guard<std::mutex> lk(ledgermtx);
+
     std::vector<std::string> file_names;
 
     // 1. 提取文件名部分
-    for (const auto& full_path : file_list_raw) {
+    for (const auto& full_path_raw : file_list_raw) {
+        std::string full_path = sanitize(full_path_raw);                    // ★修改：清洗入参路径
         size_t pos = full_path.find_last_of("/\\");
-        if (pos != std::string::npos && pos + 1 < full_path.size()) {
-            file_names.push_back(full_path.substr(pos + 1));
-        } else {
-            file_names.push_back(full_path);
-        }
+        std::string name = (pos != std::string::npos && pos + 1 < full_path.size())
+                           ? full_path.substr(pos + 1)
+                           : full_path;
+        name = sanitize(name);                                              // ★修改：清洗提取的文件名
+        file_names.push_back(name);
     }
 
+    // ★可选：去重（如果 file_list_raw 里可能有重复）
+     std::sort(file_names.begin(), file_names.end());
+     file_names.erase(std::unique(file_names.begin(), file_names.end()), file_names.end());
+
     // 2. 遍历终端
     for (auto& dev : terminal_devlist) {
-        if (dev.terminal_id == id) {    //根据终端id匹配终端
+        if (dev.terminal_id == id) {    // 根据终端id匹配终端
             for (auto& monitor : dev.line) {
                 try {
-                    ushort monitor_seq = static_cast<ushort>(std::stoi(monitor.logical_device_seq));
-                    if (monitor_seq == nCpuNo) {    //根据监测点编号匹配监测点
+                    // ★修改：清洗 logical_device_seq 再进行转换
+                    std::string seq_str = sanitize(monitor.logical_device_seq);
+                    ushort monitor_seq = static_cast<ushort>(std::stoi(seq_str));
+                    if (monitor_seq == nCpuNo) {    // 根据监测点编号匹配监测点
                         // 构造 qvvr_file
                         qvvr_file qfile;
-                        qfile.file_name.assign(file_names.begin(), file_names.end());   //终端文件列表
-                        qfile.is_download = false;
-                        qfile.is_pair = false;
+                        qfile.file_name.assign(file_names.begin(), file_names.end());   // 终端文件列表（已清洗）
+                        qfile.is_download   = false;
+                        qfile.is_pair       = false;
                         qfile.file_time_count = 0;
-                        qfile.used_status =true;
+                        qfile.used_status   = true;
 
                         // 添加到唯一的 qvvrevent
-                        monitor.qvvrevent.qvvrfile.push_back(std::move(qfile)); //记录暂态文件组
+                        monitor.qvvrevent.qvvrfile.push_back(std::move(qfile)); // 记录暂态文件组
                         return true;
                     }
                 } catch (...) {
@@ -3004,7 +3029,6 @@ bool assign_qvvr_file_list(const std::string& id, ushort nCpuNo, const std::vect
 
     return false;
 }
-
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////下载成功通知
 //提取下载路径的文件名
 std::string extract_filename1(const std::string& path) {
@@ -3065,36 +3089,96 @@ bool SendAllQvvrFiles(qvvr_file& qfile, std::string& out_wavepath) {
 }
 
 //文件下载结束接口
-bool update_qvvr_file_download(const std::string& filename_with_mac, const std::string& terminal_id) {
+
+bool update_qvvr_file_download(const std::string& filename_with_mac_in, const std::string& terminal_id) {
+
+    // ★ 先把原始入参清洗
+    std::string filename_with_mac = sanitize(filename_with_mac_in);
+
+    std::cout << "[update_qvvr_file_download] raw=" << filename_with_mac_in
+              << " | sanitized=" << filename_with_mac
+              << " | terminal_id=" << terminal_id << std::endl;
 
     //台账加锁
     std::lock_guard<std::mutex> lock(ledgermtx);
 
     // 去除 mac 路径前缀，仅保留文件名
-    std::string filename = extract_filename1(filename_with_mac);
+    std::string filename = sanitize(extract_filename1(filename_with_mac));
 
-    // 提取逻辑序号（如 PQM1 → 1）
-    size_t under_pos = filename.find('_');
-    if (under_pos == std::string::npos) return false;
+    // 提取逻辑序号（如 PQ_PQLD1 → 1）
+    size_t under_pos1 = filename.find('_');
+    if (under_pos1 == std::string::npos) {
+        std::cout << "[DEBUG] 未找到 '_'，filename=" << filename
+                  << "，under_pos=npos，返回 false\n";
+        return false;
+    }
+    size_t under_pos2 = filename.find('_', under_pos1 + 1);
+    std::string type_part = (under_pos2 == std::string::npos)
+        ? filename.substr(0, under_pos1)         // 兜底：只有一个下划线
+        : filename.substr(0, under_pos2);         // 取到第二个下划线（得到 PQ_PQLD1）
+    std::cout << "[DEBUG] type_part=" << type_part
+              << " (under_pos1=" << under_pos1
+              << ", under_pos2=" << under_pos2 << ")\n";
 
-    std::string type_part = filename.substr(0, under_pos); // PQMonitor_PQM1
     size_t num_start = type_part.find_last_not_of("0123456789");
-    if (num_start == std::string::npos || num_start + 1 >= type_part.size()) return false;
-
+    if (num_start == std::string::npos || num_start + 1 >= type_part.size()) {
+        std::cout << "[DEBUG] 数字起始位置异常：num_start=" << num_start
+                  << "，type_part.size()=" << type_part.size()
+                  << "，type_part=\"" << type_part << "\"，返回 false\n";
+        return false;
+    }
     std::string seq_str = type_part.substr(num_start + 1);
-    ushort logical_seq = static_cast<ushort>(std::stoi(seq_str)); // 逻辑序号
+    unsigned short logical_seq = static_cast<unsigned short>(std::stoul(seq_str));
+    std::cout << "[DEBUG] 解析到 logical_seq=" << logical_seq << "\n";
 
     //找终端
     for (auto& dev : terminal_devlist) {
-        if (dev.terminal_id != terminal_id) continue;
-
+        if (dev.terminal_id != terminal_id) {
+            std::cout << "[cmp-terminal-id][NOT-MATCH]"
+                      << " dev_id=" << dev.terminal_id
+                      << " target_id=" << terminal_id
+                      << std::endl;
+            continue;
+        }
         //找监测点
         for (auto& monitor : dev.line) {
             try {
                 // 将监测点台账中的 logical_device_seq 转换为数字进行匹配
                 ushort monitor_seq = static_cast<ushort>(std::stoi(monitor.logical_device_seq));
-                if (monitor_seq != logical_seq) continue;
-            } catch (...) {
+                if (monitor_seq != logical_seq) {
+                    // ★新增：不匹配时对比打印
+                    std::cout << "[cmp-monitor-seq][NOT-MATCH]"
+                          << " monitor_id=" << monitor.monitor_id
+                          // ★ 这里之前打印的是 seq_str，容易误导。改为 ledger 的原始串：
+                            << " seq_in_ledger_raw=\"" << monitor.logical_device_seq << "\""
+                          << " parsed=" << monitor_seq
+                          << " target_seq=" << logical_seq
+                          << std::endl;
+                    continue;
+                }
+                else{
+                    std::cout << "[cmp-monitor-seq][MATCH!!!]"
+                          << " monitor_id=" << monitor.monitor_id
+                          << " seq_in_ledger_raw=\"" << monitor.logical_device_seq << "\""
+                          << " parsed=" << monitor_seq
+                          << " target_seq=" << logical_seq
+                          << std::endl;
+                }
+            } catch (const std::exception& e) {
+                // ★新增：解析失败详细原因
+                std::cout << "[cmp-monitor-seq][PARSE-FAIL]"
+                      << " monitor_id=" << monitor.monitor_id
+                      << " seq_in_ledger=\"" << monitor.logical_device_seq << "\""
+                      << " err=" << e.what()
+                      << std::endl;
+                continue; // logical_device_seq 非法，跳过
+            }catch (...) {
+                // ★新增：未知异常
+                std::cout << "[cmp-monitor-seq][PARSE-FAIL]"
+                      << " monitor_id=" << monitor.monitor_id
+                      << " seq_in_ledger=\"" << monitor.logical_device_seq << "\""
+                      << " err=<unknown>"
+                      << std::endl;
                 continue; // logical_device_seq 非法，跳过
             }
 
@@ -3108,6 +3192,7 @@ bool update_qvvr_file_download(const std::string& filename_with_mac, const std::
                 if (it != qfile.file_name.end()) {
                     // 添加到 file_download（记录完整路径，避免重复）
                     if (std::find(qfile.file_download.begin(), qfile.file_download.end(), filename_with_mac) == qfile.file_download.end()) {
+                        std::cout << "[update_qvvr_file_download] Adding downloaded file: " << filename_with_mac << std::endl;
                         qfile.file_download.push_back(filename_with_mac);
                     }
 
@@ -3117,11 +3202,19 @@ bool update_qvvr_file_download(const std::string& filename_with_mac, const std::
                     std::set<std::string> s_name(qfile.file_name.begin(), qfile.file_name.end());
                     std::set<std::string> s_down;
                     for (const auto& path : qfile.file_download) {
-                        s_down.insert(extract_filename1(path));  // 提取每个路径中的文件名
+                        s_down.insert(sanitize(extract_filename1(path)));  // 提取每个路径中的文件名
                     }
 
+                    //打印s_name和s_down内容
+                    std::cout << "[update_qvvr_file_download] Expected files (file_name): ";
+                    for (const auto& fn : s_name) std::cout << fn << " ";
+                    std::cout << std::endl;
+                    std::cout << "[update_qvvr_file_download] Downloaded files (file_download): ";
+                    for (const auto& fn : s_down) std::cout << fn << " ";
+
                     // 检查 file_download 是否与 file_name 完全一致（集合相同）
                     if (s_name == s_down) {
+                        std::cout << "[update_qvvr_file_download] All files downloaded for qfile in logical_seq=" << logical_seq << std::endl;
                         qfile.is_download = true;   // 全部下载完成
 
                         // 找到其中的 .cfg 文件进行匹配
@@ -3169,7 +3262,13 @@ bool update_qvvr_file_download(const std::string& filename_with_mac, const std::
                                         if (it != monitor.qvvrevent.qvvrdata.end()) {
                                             monitor.qvvrevent.qvvrdata.erase(it);
                                         }
-                                    }                                   
+                                    } 
+                                    else {
+                                        std::cerr << "[update_qvvr_file_download] Failed to send qvvr files for logical_seq=" << logical_seq << std::endl;
+                                    }                                  
+                                }
+                                else {
+                                    std::cout << "[update_qvvr_file_download] No matching qvvr_data found for cfg file: " << fpath << std::endl;
                                 }
                                 break; // 只处理第一个 cfg 文件
                             }
@@ -3265,6 +3364,8 @@ bool send_file_list(const std::string &dev_id, const std::vector<tag_dir_info> &
     queue_data_t connect_info;
     connect_info.strTopic = Topic_Reply_Topic;
     connect_info.strText  = j.dump();  // 序列化为字符串
+    connect_info.tag = Topic_Reply_Tag;
+    connect_info.key = Topic_Reply_Key;
     {
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
         queue_data_list.push_back(std::move(connect_info));
@@ -3658,6 +3759,8 @@ bool send_set_value_reply(const std::string &dev_id, unsigned char mp_index, con
     queue_data_t connect_info;
     connect_info.strTopic = Topic_Reply_Topic;
     connect_info.strText  = j.dump(); // 序列化为字符串
+    connect_info.tag = Topic_Reply_Tag;
+    connect_info.key = Topic_Reply_Key;
 
     {
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
@@ -3854,6 +3957,8 @@ bool send_internal_value_reply(const std::string &dev_id, const std::vector<DZ_k
     queue_data_t connect_info;
     connect_info.strTopic = Topic_Reply_Topic;
     connect_info.strText  = j.dump(); // 序列化为字符串
+    connect_info.tag = Topic_Reply_Tag;
+    connect_info.key = Topic_Reply_Key;
 
     {
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
@@ -3919,6 +4024,8 @@ void send_reply_to_kafka_recall(const std::string& guid, const std::string& step
     queue_data_t connect_info;
     connect_info.strTopic = Topic_Reply_Topic;
     connect_info.strText = jsonString;
+    connect_info.tag = Topic_Reply_Tag;
+    connect_info.key = Topic_Reply_Key;
 
     // 加入发送队列（带互斥锁保护）
     queue_data_list_mutex.lock();
@@ -4561,11 +4668,6 @@ void on_device_response_minimal(int response_code,
         } 
         else {
 
-            // 准备日志用的 key
-            std::ostringstream mon_key_c, mon_key_d;
-            mon_key_c << "monitor." << matched_monitor->terminal_id << "." << matched_monitor->logical_device_seq << ".COM";
-            mon_key_d << "monitor." << matched_monitor->terminal_id << "." << matched_monitor->logical_device_seq << ".DATA";
-
             // 2) 仅更新该监测点 recall_list 的首条，且要求处于 RUNNING
             if (!matched_monitor->recall_list.empty()) {
                 RecallMonitor& front = matched_monitor->recall_list.front();    //取出首条
@@ -4584,7 +4686,7 @@ void on_device_response_minimal(int response_code,
                                   << " rc=" << response_code << std::endl;  //错误响应码
 
                         //记录日志
-                        DIY_ERRORLOG_CODE(mon_key_d.str().c_str(),static_cast<int>(LogCode::LOG_CODE_RECALL),"【ERROR】监测点:%s 补招数据失败 - 失败时间点:%lld 至 %lld",mon_key_d.str().c_str(),front.StartTime,front.EndTime);
+                        DIY_ERRORLOG_CODE(matched_monitor->monitor_id.c_str(),2,static_cast<int>(LogCode::LOG_CODE_RECALL),"【ERROR】监测点:%s 补招数据失败 - 失败时间点:%lld 至 %lld",matched_monitor->monitor_id.c_str(),front.StartTime,front.EndTime);
                     }
                     updated = true;
                 } else {    //首条不是 RUNNING 状态，不应该收到这条响应
@@ -4835,6 +4937,157 @@ void on_device_response_minimal(int response_code,
     }
 }
 
+///////////////////////////////////////////////////////////////////////////////////////////////////////////////////记录暂态事件到本地
+// 将一条暂态数据更新/写入到指定终端ID与逻辑序号的监测点
+// 返回 true 表示已写入（更新或追加），false 表示未找到对应终端/监测点。
+
+bool append_qvvr_event(const std::string& terminal_id,
+                              int              logical_seq,      // 监测点序号（如 1）
+                              int              nType,            // 事件类型
+                              double           fPersisstime_sec, // 持续时间（秒）
+                              double           fMagnitude_pu,    // 幅值（pu）
+                              uint64_t         triggerTimeMs,    // 触发时间（毫秒）
+                              int              phase)            // 相别
+{
+    std::cout << "[append_qvvr_event] enter"
+              << " tid=" << std::this_thread::get_id()
+              << " terminal_id=" << terminal_id
+              << " logical_seq=" << logical_seq
+              << " type=" << nType
+              << " per_s=" << fPersisstime_sec
+              << " mag_pu=" << fMagnitude_pu
+              << " time_ms=" << static_cast<unsigned long long>(triggerTimeMs)
+              << " phase=" << phase
+              << std::endl;
+
+    std::lock_guard<std::mutex> lk(ledgermtx);
+    std::cout << "[append_qvvr_event] lock acquired. terminal_devlist.size="
+              << terminal_devlist.size() << std::endl;
+
+    // 1) 找终端
+    auto dev_it = std::find_if(terminal_devlist.begin(), terminal_devlist.end(),
+        [&](const terminal_dev& d){ return d.terminal_id == terminal_id; });
+
+    if (dev_it == terminal_devlist.end()) {
+        std::cout << "[append_qvvr_event][MISS] terminal not found: "
+                  << terminal_id << std::endl;
+        return false;
+    }
+    std::cout << "[append_qvvr_event][HIT] terminal_id=" << terminal_id
+              << " monitors(line).size=" << dev_it->line.size()
+              << std::endl;
+
+    // 2) 找监测点（按逻辑序号匹配：字符串等于 或 数值等于）
+    ledger_monitor* pMon = nullptr;
+    for (size_t i = 0; i < dev_it->line.size(); ++i) {
+        auto& m = dev_it->line[i];
+        bool eq_str = (m.logical_device_seq == std::to_string(logical_seq));
+        bool eq_num = false;
+        try {
+            if (!m.logical_device_seq.empty())
+                eq_num = (std::stoi(m.logical_device_seq) == logical_seq);
+        } catch (...) {
+            // 仅调试提示，不改变原逻辑
+            std::cout << "[append_qvvr_event][monitor #" << i
+                      << "] stoi fail, logical_device_seq=\""
+                      << m.logical_device_seq << "\"" << std::endl;
+        }
+
+        std::cout << "[append_qvvr_event][probe monitor #" << i
+                  << "] monitor_id=" << m.monitor_id
+                  << " logical_device_seq=\"" << m.logical_device_seq << "\""
+                  << " eq_str=" << eq_str << " eq_num=" << eq_num << std::endl;
+
+        if (eq_str || eq_num) { pMon = &m; break; }
+    }
+
+    if (!pMon) {
+        std::cout << "[append_qvvr_event][MISS] monitor not found by seq="
+                  << logical_seq << " in terminal_id=" << terminal_id << std::endl;
+        return false;
+    }
+    std::cout << "[append_qvvr_event][HIT] monitor_id=" << pMon->monitor_id
+              << " logical_device_seq=" << pMon->logical_device_seq
+              << " qvvrdata.size=" << pMon->qvvrevent.qvvrdata.size()
+              << std::endl;
+
+    qvvr_event& qe = pMon->qvvrevent;
+
+    // 3) 先尝试“就地更新”（同类型 + 同时间戳 视为同一事件）
+    for (size_t i = 0; i < qe.qvvrdata.size(); ++i) {
+        auto& q = qe.qvvrdata[i];
+        if (q.QVVR_type == nType && q.QVVR_time == triggerTimeMs) {
+            std::cout << "[append_qvvr_event][UPDATE match idx=" << i << "]"
+                      << " old{used=" << q.used_status
+                      << ", per=" << q.QVVR_PerTime
+                      << ", mag=" << q.QVVR_Amg
+                      << ", phase=" << q.phase
+                      << "} -> new{used=true"
+                      << ", per=" << fPersisstime_sec
+                      << ", mag=" << fMagnitude_pu
+                      << ", phase=" << phase
+                      << "}" << std::endl;
+
+            q.used_status   = true;
+            q.QVVR_PerTime  = fPersisstime_sec;
+            q.QVVR_Amg      = fMagnitude_pu;
+            q.phase         = phase;
+
+            std::cout << "[append_qvvr_event] done(update)."
+                      << std::endl;
+            return true;
+        }
+    }
+
+    // 4) 复用空槽（used_status=false）
+    for (size_t i = 0; i < qe.qvvrdata.size(); ++i) {
+        auto& q = qe.qvvrdata[i];
+        if (!q.used_status) {
+            std::cout << "[append_qvvr_event][REUSE idx=" << i << "]"
+                      << " set{type=" << nType
+                      << ", time_ms=" << static_cast<unsigned long long>(triggerTimeMs)
+                      << ", per=" << fPersisstime_sec
+                      << ", mag=" << fMagnitude_pu
+                      << ", phase=" << phase
+                      << "}" << std::endl;
+
+            q.used_status   = true;
+            q.QVVR_type     = nType;
+            q.QVVR_time     = triggerTimeMs;
+            q.QVVR_PerTime  = fPersisstime_sec;
+            q.QVVR_Amg      = fMagnitude_pu;
+            q.phase         = phase;
+
+            std::cout << "[append_qvvr_event] done(reuse)."
+                      << std::endl;
+            return true;
+        }
+    }
+
+    // 5) 直接追加
+    qvvr_data q{};
+    q.used_status   = true;
+    q.QVVR_type     = nType;
+    q.QVVR_time     = triggerTimeMs;     // ms
+    q.QVVR_PerTime  = fPersisstime_sec;  // s
+    q.QVVR_Amg      = fMagnitude_pu;
+    q.phase         = phase;
+    qe.qvvrdata.push_back(q);
+
+    std::cout << "[append_qvvr_event][PUSH_BACK]"
+              << " new_size=" << qe.qvvrdata.size()
+              << " last_idx=" << (qe.qvvrdata.empty() ? -1 : (int)qe.qvvrdata.size()-1)
+              << " {type=" << nType
+              << ", time_ms=" << static_cast<unsigned long long>(triggerTimeMs)
+              << ", per=" << fPersisstime_sec
+              << ", mag=" << fMagnitude_pu
+              << ", phase=" << phase
+              << "}" << std::endl;
+
+    std::cout << "[append_qvvr_event] done(push_back)."
+              << std::endl;
+    return true;
+}
 
 
 

LFtid1056/cloudfront/code/interface.cpp

@@ -25,7 +25,7 @@
 #include "interface.h"
 #include "rocketmq.h"
 #include "nlohmann/json.hpp"
-#include "log4cplus/log4.h"
+#include "log4.h"
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
@@ -200,9 +200,28 @@ void handleUploadResponse(const std::string& response, std::string& wavepath) {
 
 //上传文件
 void SendFileWeb(const std::string& strUrl, const std::string& localpath, const std::string& cloudpath, std::string& wavepath) {
+
+    // 基本存在性检查
+    if (access(localpath.c_str(), F_OK) != 0) {
+        std::cerr << "Local file does not exist: " << localpath << std::endl;
+        return;
+    }
+    // ★新增：stat 打印大小，便于快速确认读源
+    struct stat st {};
+    if (stat(localpath.c_str(), &st) != 0) {
+        perror("stat");
+    } else {
+        std::cout << "[debug] upload file: " << localpath
+                  << ", size=" << static_cast<long long>(st.st_size) << " bytes\n";
+    }
+
     // 初始化 curl
     CURL* curl = curl_easy_init();
     if (curl) {
+        // ★新增：错误缓冲，便于看到更具体的错误
+        char errbuf[CURL_ERROR_SIZE] = {0};
+        curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, errbuf); // ★新增
+
         // 设置请求 URL 和 POST 请求
         curl_easy_setopt(curl, CURLOPT_URL, strUrl.c_str());
         curl_easy_setopt(curl, CURLOPT_POST, 1);
@@ -211,21 +230,16 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
         curl_httppost* formpost = nullptr;
         curl_httppost* lastptr = nullptr;
 
-        if (access(localpath.c_str(), F_OK) != 0) {
-            std::cerr << "Local file does not exist: " << localpath << std::endl;
-            return;
-        }
-
         // 添加文件字段，直接从本地路径读取文件内容
         curl_formadd(&formpost, &lastptr,
                      CURLFORM_COPYNAME, "file",
-                     CURLFORM_FILE, localpath,
+                     CURLFORM_FILE, localpath.c_str(),
                      CURLFORM_END);
 
         // 添加 `path` 字段
         curl_formadd(&formpost, &lastptr,
                      CURLFORM_COPYNAME, "path",
-                     CURLFORM_COPYCONTENTS, cloudpath,
+                     CURLFORM_COPYCONTENTS, cloudpath.c_str(),
                      CURLFORM_END);
 
         // 添加 `isReserveName` 字段
@@ -238,13 +252,14 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
         curl_easy_setopt(curl, CURLOPT_HTTPPOST, formpost);
 
         // 设置头信息
-        struct curl_slist* header_list = nullptr;
-        header_list = curl_slist_append(header_list, "Content-Type: multipart/form-data");
-        curl_easy_setopt(curl, CURLOPT_HTTPHEADER, header_list);
+        //struct curl_slist* header_list = nullptr;
+        //header_list = curl_slist_append(header_list, "Content-Type: multipart/form-data");
+        //curl_easy_setopt(curl, CURLOPT_HTTPHEADER, header_list);
 
         // 设置超时时间
         curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 10);
         curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10);
+        curl_easy_setopt(curl, CURLOPT_NOSIGNAL,      1L); // ★新增：避免多线程环境下超时触发信号
 
         // 设置写入响应数据的函数
         std::string resPost0;
@@ -254,7 +269,8 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
         // 执行请求
         CURLcode res = curl_easy_perform(curl);
         if (res != CURLE_OK) {
-            std::cerr << "http web failed: " << curl_easy_strerror(res) << std::endl;
+            const char* em = errbuf[0] ? errbuf : curl_easy_strerror(res);
+            std::cerr << "http web failed: " << em << std::endl;
 			DIY_ERRORLOG("process","【ERROR】前置上传暂态录波文件 %s 失败，请检查文件上传接口配置",localpath);
         } else {
             std::cout << "http web success, response: " << resPost0 << std::endl;
@@ -263,7 +279,7 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
 
         // 清理
         curl_formfree(formpost);  // 释放表单数据
-        curl_slist_free_all(header_list);  // 释放头部列表
+        //curl_slist_free_all(header_list);  // 释放头部列表
         curl_easy_cleanup(curl);
     } else {
         std::cerr << ">>> curl init failed" << std::endl;
@@ -1176,22 +1192,41 @@ static void scanAndResendOfflineFiles(const std::string& dirPath)
     }
 }
 
+// 0->"A", 1->"B", 2->"C", 3->"AB", 4->"BC", 5->"CA", 6/其它->"ABC"
+inline std::string phase_to_text(int phase) {
+    switch (phase) {
+        case 0: return "A";
+        case 1: return "B";
+        case 2: return "C";
+        case 3: return "AB";
+        case 4: return "BC";
+        case 5: return "CA";
+        default: return "ABC"; // 包含 6 和其他任何值
+    }
+}
+
 int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
                             double mag, double dur, long long start_tm, int dis_kind,int phase,
                             const std::string& wavepath) {
-    
-    // 监测点日志的 key, lnk20250526
-    std::string full_key_m_c = "monitor." + dev_id + "." + std::to_string(monitor_id) + ".COM";
-    std::string full_key_m_d = "monitor." + dev_id + "." + std::to_string(monitor_id) + ".DATA";
-    // 监测点日志的 key, lnk20250526
 
     if (dev_id.empty()) {
-        std::cout << "dev_id is null" << std::endl;
-        return 0;
+        std::cout << "qvvr send error ,dev_id is null" << std::endl;
+        return 1;
     }
 
     // 构造 JSON 对象
     json root;
+
+    //找监测点id
+    std::string mpid;
+
+    get_monitor_id_by_dev_and_seq(dev_id, monitor_id, mpid);
+    if(mpid.empty()) {
+        std::cout << "qvvr send error ,monitorId is null" << std::endl;
+        return 1;
+    }
+
+    root["monitorId"] = mpid;
     root["devId"] = dev_id;
     root["CpuNo"] = monitor_id;
     root["amplitude"] = mag;
@@ -1208,8 +1243,8 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
     std::string start_time_str = start_time_stream.str();
 
     root["startTime"] = start_time_str;
-    root["wavePath"] = wavepath;   
-    root["phase"] = phase;  
+    root["wavePath"] = wavepath;
+    root["phase"] = phase_to_text(phase);
 
 
     std::string json_string = root.dump(4);
@@ -1226,7 +1261,7 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
             // 有效响应，略过
         } catch (...) {
             // 响应异常，保存 json
-            DIY_ERRORLOG(full_key_m_d.c_str(), "【ERROR】暂态接口响应异常,无法上送装置%s监测点%s的暂态事件",dev_id, monitor_id);
+            DIY_ERRORLOG(mpid.c_str(), "【ERROR】暂态接口响应异常,无法上送装置%s监测点%s的暂态事件",dev_id, monitor_id);
 
             std::cout << "qvvr send fail ,store in local" << std::endl;
             std::string qvvrDir = FRONT_PATH + "/dat/qvvr/";
@@ -1236,7 +1271,7 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
         }
     } else {
         // 无响应，保存 json
-        DIY_ERRORLOG(full_key_m_d.c_str(), "【ERROR】暂态接口无响应,无法上送装置%s监测点%s的暂态事件",dev_id, monitor_id);
+        DIY_ERRORLOG(mpid.c_str(), "【ERROR】暂态接口无响应,无法上送装置%s监测点%s的暂态事件",dev_id, monitor_id);
 
         std::cout << "qvvr send fail ,store in local" << std::endl;
         std::string qvvrDir = FRONT_PATH + "/dat/qvvr/";
@@ -1301,3 +1336,36 @@ void handleCommentResponse(const std::string& response) {
     }
 }
 
+/////////////////////////////////////////////////////////////////////////////////////////////////////找监测点id
+bool get_monitor_id_by_dev_and_seq(const std::string& terminal_id,
+                                   unsigned short logical_seq,
+                                   std::string& out_monitor_id)
+{
+    std::lock_guard<std::mutex> lk(ledgermtx);   // 若你的工程里有全局 ledgermtx/terminal_devlist
+
+    for (const auto& dev : terminal_devlist) {
+        if (dev.terminal_id != terminal_id) continue;
+
+        // 命中该装置后，仅遍历它的监测点
+        for (const auto& mon : dev.line) {
+            try {
+                // logical_device_seq 存在台账里是字符串，转成数值再比对
+                unsigned long v = std::stoul(mon.logical_device_seq);
+                unsigned short seq = static_cast<unsigned short>(v);
+                if (seq == logical_seq) {
+                    out_monitor_id = mon.monitor_id;
+                    return true;
+                }
+            } catch (...) {
+
+                // 非法数值（空/非数字等）直接跳过
+                continue;
+            }
+        }
+        break; // 已找到对应装置，没必要再看其它装置
+    }
+    return false;
+}
+
+
+

LFtid1056/cloudfront/code/interface.h

@@ -265,6 +265,8 @@ public:
 	std::string		strTopic;		   //发送topic
 	std::string		strText;		   //发送的json字符串
 	std::string	    mp_id;             //监测点id
+    std::string     tag;              //消息tag
+    std::string     key;                // 消息key
 }; 
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -589,6 +591,13 @@ std::string generate_json(  //构造装置主动上送数据的报文
 );
 
 //暂态事件接口
+bool append_qvvr_event(const std::string& terminal_id,
+                              int              logical_seq,      // 监测点序号（如 1）
+                              int              nType,            // 事件类型
+                              double           fPersisstime_sec, // 持续时间（秒）
+                              double           fMagnitude_pu,    // 幅值（pu）
+                              uint64_t         triggerTimeMs,    // 触发时间（毫秒）
+                              int              phase);
 int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
                             double mag, double dur, long long start_tm, int dis_kind,int phase,
                             const std::string& wavepath);
@@ -650,6 +659,11 @@ void on_device_response_minimal(int response_code,
                                 unsigned char cid,
                                 int device_state_int);
 
+//找监测点id
+bool get_monitor_id_by_dev_and_seq(const std::string& terminal_id,
+                                   unsigned short logical_seq,
+                                   std::string& out_monitor_id);
+
 //处理补招的任务
 void check_recall_event();
 void check_recall_file();
@@ -675,6 +689,18 @@ inline std::string trim_cstr(const char* s, size_t n) {
     return out;
 }
 
+inline std::string sanitize(std::string s) {
+    // 截断第一个 NUL 及其后内容
+    size_t z = s.find('\0');
+    if (z != std::string::npos) s.erase(z);
+    // 去掉尾部不可打印字符（含 \r \n 等）
+    while (!s.empty()) {
+        unsigned char c = static_cast<unsigned char>(s.back());
+        if (c >= 32 && c != 127) break;
+        s.pop_back();
+    }
+    return s;
+}
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 extern int g_front_seg_index;
 extern std::string FRONT_IP;

LFtid1056/cloudfront/code/log4.cpp

@@ -24,12 +24,11 @@
 #include "log4cplus/fileappender.h"
 #include "log4cplus/layout.h"
 #include "log4cplus/ndc.h"
-#include "log4cplus/log4.h"
 #include "log4cplus/spi/loggingevent.h"
 
 #include "rocketmq.h"
 #include "interface.h"
-#include "log4cplus/log4.h"
+#include "log4.h"
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 //log4命名空间
@@ -152,8 +151,6 @@ protected:
                 << "\",\"businessId\":\"" << extract_logger_id(logger_name)
                 << "\",\"level\":\"" << level_str
                 << "\",\"grade\":\"" << get_level_str(level)
-                << "\",\"logtype\":\"" << (logtype == LOGTYPE_COM ? "com" : "data")
-                << "\",\"frontType\":\"" << "cloudfront"
                 // ★新增：输出 code 字段（整型）
                 << "\",\"code\":\"" << code
                 << "\",\"log\":\"" << escape_json(msg) << "\"}";
@@ -163,6 +160,8 @@ protected:
             queue_data_t connect_info;
             connect_info.strTopic = G_LOG_TOPIC;
             connect_info.strText = jsonString;
+            connect_info.tag = G_LOG_TAG;
+            connect_info.key = G_LOG_KEY;
 
             std::lock_guard<std::mutex> lock(queue_data_list_mutex);
             queue_data_list.push_back(connect_info);
@@ -309,12 +308,10 @@ void init_loggers_bydevid(const std::string& dev_id)
 
         std::string ip_str = term.addr_str.empty() ? "unknown" : term.addr_str;
         std::string device_dir = base_dir + "/" + ip_str;
-        std::string device_key_c = std::string("terminal.") + dev_id + ".COM";
-        std::string device_key_d = std::string("terminal.") + dev_id + ".DATA";
+        std::string device_key = std::string("terminal.") + dev_id;
 
         // 添加判断：终端日志 logger 是否已存在
-        if (logger_map.find(device_key_c) == logger_map.end() &&
-            logger_map.find(device_key_d) == logger_map.end()) {
+        if (logger_map.find(device_key) == logger_map.end()) {
 
             // 所有终端日志（com 和 data）写到同一个 device 日志文件中
             std::string file_path_t = device_dir + "/" + dev_id + ".log";
@@ -323,27 +320,23 @@ void init_loggers_bydevid(const std::string& dev_id)
             SharedAppenderPtr device_appender(new RollingFileAppender(file_path_t, 1 * 1024 * 1024, 2));
             device_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
 
-            Logger device_logger_c = init_logger(device_key_c, device_dir, dev_id, device_appender);
-            Logger device_logger_d = init_logger(device_key_d, device_dir, dev_id, device_appender);
-            logger_map[device_key_c]  = TypedLogger(device_logger_c, LOGTYPE_COM);
-            logger_map[device_key_d] = TypedLogger(device_logger_d, LOGTYPE_DATA);
+            Logger device_logger = init_logger(device_key, device_dir, dev_id, device_appender);
+            logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
 
-            DIY_WARNLOG(device_key_d.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
+            DIY_WARNLOG(dev_id.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
         }
 
         // 初始化监测点日志，monitor.<mp_id>.COM / .DATA
         for (size_t j = 0; j < term.line.size(); ++j) {
             const ledger_monitor& monitor = term.line[j];
             if (!monitor.monitor_id.empty()) {
-                std::ostringstream mon_key_c, mon_key_d, mon_path, mon_name;
-                mon_key_c << "monitor." << term.terminal_id << "." << monitor.logical_device_seq << ".COM";
-                mon_key_d << "monitor." << term.terminal_id << "." <<  monitor.logical_device_seq << ".DATA";
+                std::ostringstream mon_key, mon_path, mon_name;
+                mon_key << "monitor." << monitor.monitor_id;
                 mon_path << device_dir << "/monitor" << j;
                 mon_name << monitor.monitor_id;
 
                 // 判断监测点 logger 是否已存在
-                if (logger_map.find(mon_key_c.str()) == logger_map.end() &&
-                    logger_map.find(mon_key_d.str()) == logger_map.end()) {
+                if (logger_map.find(mon_key.str()) == logger_map.end()) {
 
                     // 所有监测点日志（com 和 data）写到同一个 monitor 日志文件中
                     std::string file_path_m = mon_path.str() + "/" + mon_name.str() + ".log";
@@ -352,12 +345,10 @@ void init_loggers_bydevid(const std::string& dev_id)
                     SharedAppenderPtr monitor_appender(new RollingFileAppender(file_path_m, 1 * 1024 * 1024, 2));
                     monitor_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
 
-                    Logger mon_logger_c = init_logger(mon_key_c.str(), mon_path.str(), mon_name.str(), monitor_appender);
-                    Logger mon_logger_d = init_logger(mon_key_d.str(), mon_path.str(), mon_name.str(), monitor_appender);
-                    logger_map[mon_key_c.str()] = TypedLogger(mon_logger_c, LOGTYPE_COM);
-                    logger_map[mon_key_d.str()] = TypedLogger(mon_logger_d, LOGTYPE_DATA);
+                    Logger mon_logger = init_logger(mon_key.str(), mon_path.str(), mon_name.str(), monitor_appender);
+                    logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
 
-                    DIY_WARNLOG(mon_key_d.str().c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕", monitor.monitor_name.c_str(), monitor.logical_device_seq.c_str());
+                    DIY_WARNLOG(monitor.monitor_id.c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕", monitor.monitor_name.c_str(), monitor.logical_device_seq.c_str());
                 }
             }
         }
@@ -384,8 +375,7 @@ void init_loggers()
         std::string ip_str = term.addr_str.empty() ? "unknown" : term.addr_str;
         std::string device_dir = base_dir + "/" + ip_str;
 
-        std::string device_key_c = std::string("terminal.") + term.terminal_id + ".COM";
-        std::string device_key_d = std::string("terminal.") + term.terminal_id + ".DATA";
+        std::string device_key = std::string("terminal.") + term.terminal_id;
 
         // 所有终端日志（com 和 data）写到同一个 device 日志文件中
         std::string file_path_t = device_dir + "/" + term.terminal_id + ".log";
@@ -394,22 +384,19 @@ void init_loggers()
         SharedAppenderPtr device_appender(new RollingFileAppender(file_path_t, 1 * 1024 * 1024, 2));
         device_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
 
-        Logger device_logger_c = init_logger(device_key_c, device_dir, term.terminal_id, device_appender);
-        Logger device_logger_d = init_logger(device_key_d, device_dir, term.terminal_id, device_appender);
+        Logger device_logger = init_logger(device_key, device_dir, term.terminal_id, device_appender);
 
-        logger_map[device_key_c] = TypedLogger(device_logger_c, LOGTYPE_COM);
-        logger_map[device_key_d] = TypedLogger(device_logger_d, LOGTYPE_DATA);
+        logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
 
-        DIY_WARNLOG(device_key_d.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
+        DIY_WARNLOG(term.terminal_id.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
 
         // 初始化监测点日志
         for (size_t i = 0; i < term.line.size(); ++i) {
             const ledger_monitor& monitor = term.line[i];
 
             if (!monitor.monitor_id.empty()) {
-                std::ostringstream mon_key_c, mon_key_d, mon_path, mon_name;
-                mon_key_c << "monitor." << term.terminal_id << "." << monitor.logical_device_seq << ".COM";
-                mon_key_d << "monitor." << term.terminal_id << "." <<  monitor.logical_device_seq << ".DATA";
+                std::ostringstream mon_key, mon_path, mon_name;
+                mon_key << "monitor." << monitor.monitor_id;
 
                 mon_path << device_dir << "/monitor" << i; // 用monitor+序号作为目录
                 mon_name << monitor.monitor_id;
@@ -420,13 +407,11 @@ void init_loggers()
                 SharedAppenderPtr monitor_appender(new RollingFileAppender(file_path_m, 1 * 1024 * 1024, 2));
                 monitor_appender->setLayout(std::unique_ptr<Layout>(new PatternLayout("%D{%Y-%m-%d %H:%M:%S} [%p] [%c] %m%n")));
 
-                Logger mon_logger_c = init_logger(mon_key_c.str(), mon_path.str(), mon_name.str(), monitor_appender);
-                Logger mon_logger_d = init_logger(mon_key_d.str(), mon_path.str(), mon_name.str(), monitor_appender);
+                Logger mon_logger = init_logger(mon_key.str(), mon_path.str(), mon_name.str(), monitor_appender);
 
-                logger_map[mon_key_c.str()] = TypedLogger(mon_logger_c, LOGTYPE_COM);
-                logger_map[mon_key_d.str()] = TypedLogger(mon_logger_d, LOGTYPE_DATA);
+                logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
 
-                DIY_WARNLOG(mon_key_d.str().c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕",
+                DIY_WARNLOG(mon_key.str().c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕",
                             monitor.monitor_name.c_str(), monitor.logical_device_seq.c_str());
             }
         }
@@ -441,17 +426,11 @@ void remove_loggers_by_terminal_id(const std::string& terminal_id) {
         if (term.terminal_id != terminal_id) continue;
 
         // 删除终端日志 logger
-        std::string com_key = "terminal." + terminal_id + ".COM";
-        std::string data_key = "terminal." + terminal_id + ".DATA";
+        std::string terminal_key = "terminal." + terminal_id;
 
-        if (logger_map.count(com_key)) {
-            logger_map[com_key].logger.removeAllAppenders();
-            logger_map.erase(com_key);
-        }
-
-        if (logger_map.count(data_key)) {
-            logger_map[data_key].logger.removeAllAppenders();
-            logger_map.erase(data_key);
+        if (logger_map.count(terminal_key)) {
+            logger_map[terminal_key].logger.removeAllAppenders();
+            logger_map.erase(terminal_key);
         }
 
         // 删除监测点日志 logger
@@ -459,18 +438,13 @@ void remove_loggers_by_terminal_id(const std::string& terminal_id) {
             const ledger_monitor& monitor = term.line[i];
             if (!monitor.monitor_id.empty()) {
                 std::string mon_prefix = "monitor." + monitor.monitor_id;
-                std::string mon_com_key = mon_prefix + ".COM";
-                std::string mon_data_key = mon_prefix + ".DATA";
+                std::string mon_key = mon_prefix;
 
-                if (logger_map.count(mon_com_key)) {
-                    logger_map[mon_com_key].logger.removeAllAppenders();
-                    logger_map.erase(mon_com_key);
+                if (logger_map.count(mon_key)) {
+                    logger_map[mon_key].logger.removeAllAppenders();
+                    logger_map.erase(mon_key);
                 }
 
-                if (logger_map.count(mon_data_key)) {
-                    logger_map[mon_data_key].logger.removeAllAppenders();
-                    logger_map.erase(mon_data_key);
-                }
             }
         }
 

LFtid1056/cloudfront/code/log4.h

@@ -105,44 +105,6 @@ void format_log_msg(char* buf, size_t buf_size, const char* fmt, ...);
 
 // ====================== ★新增结束 ======================
 
-
-// ====================== 日志宏区域 ======================
-// 原始不带 code 的实现（兼容/复用）
-#define DIY_LOG(LEVEL_FUNC, KEY, ...)                             \
-    do {                                                          \
-        char buf[256];                                            \
-        format_log_msg(buf, sizeof(buf), __VA_ARGS__);            \
-        LEVEL_FUNC(KEY, buf);                                     \
-    } while (0)
-
-// ★新增：带 code 的实现（C/C++ 通用，使用 TLS 保存/恢复）
-#define DIY_LOG_CODE(LEVEL_FUNC, KEY, CODE_INT, ...)              \
-    do {                                                          \
-        int __old_code__ = g_log_code_tls;                        /* 备份旧值 */ \
-        g_log_code_tls = (int)(CODE_INT);                         /* 设置本次日志 code */ \
-        char buf[256];                                            \
-        format_log_msg(buf, sizeof(buf), __VA_ARGS__);            \
-        LEVEL_FUNC(KEY, buf);                                     /* 输出日志 */ \
-        g_log_code_tls = __old_code__;                            /* 恢复旧值 */ \
-    } while (0)
-
-// ★修改：默认宏改为 code=0（兼容原有用法）
-#define DIY_ERRORLOG(KEY, ...) DIY_LOG_CODE(log_error, KEY, 0, __VA_ARGS__)   // ★修改：默认 code=0
-#define DIY_WARNLOG(KEY,  ...) DIY_LOG_CODE(log_warn,  KEY, 0, __VA_ARGS__)   // ★修改：默认 code=0
-#define DIY_INFOLOG(KEY,  ...) DIY_LOG_CODE(log_info,  KEY, 0, __VA_ARGS__)   // ★修改：默认 code=0
-#define DIY_DEBUGLOG(KEY, ...) DIY_LOG_CODE(log_debug, KEY, 0, __VA_ARGS__)   // ★修改：默认 code=0
-
-// ★新增：显式传入 code 的便捷宏
-// 用法示例：DIY_WARNLOG_CODE(full_key_m_c, warn_recallstart, "【WARN】监测点:%s ...", ...);
-#define DIY_ERRORLOG_CODE(KEY, CODE_INT, ...) DIY_LOG_CODE(log_error, KEY, CODE_INT, __VA_ARGS__)  // ★新增
-#define DIY_WARNLOG_CODE(KEY,  CODE_INT, ...) DIY_LOG_CODE(log_warn,  KEY, CODE_INT, __VA_ARGS__)  // ★新增
-#define DIY_INFOLOG_CODE(KEY,  CODE_INT, ...) DIY_LOG_CODE(log_info,  KEY, CODE_INT, __VA_ARGS__)  // ★新增
-#define DIY_DEBUGLOG_CODE(KEY, CODE_INT, ...) DIY_LOG_CODE(log_debug, KEY, CODE_INT, __VA_ARGS__)  // ★新增
-
-
-// ====================== 日志宏区域 ======================
-
-
 typedef enum LogCode {
     LOG_CODE_OTHER             = 99,   /* 其他类型 */
     LOG_CODE_LEDGER            = 100,  /* 台账类型 */
@@ -159,9 +121,77 @@ typedef enum LogCode {
     LOG_CODE_LOG_REQUEST       = 405,  /* 日志请求 */
     LOG_CODE_REPORT            = 500,  /* 报告处理 */
     LOG_CODE_COMM              = 600,  /* 通讯状态 */
-    LOG_CODE_SPACE_ALARM       = 700   /* 空间告警 */
+    LOG_CODE_SPACE_ALARM       = 700,  /* 空间告警 */
+    LOG_CODE_DEV_ALARM         = 800   /* 装置告警 */
 } LogCode;
 
+// ====================== 日志宏区域 ======================
+// 原始不带 code 的实现（兼容/复用）
+#define DIY_LOG(LEVEL_FUNC, KEY, ...)                             \
+    do {                                                          \
+        char buf[256];                                            \
+        format_log_msg(buf, sizeof(buf), __VA_ARGS__);            \
+        LEVEL_FUNC(KEY, buf);                                     \
+    } while (0)
+
+// ★新增：带 code 的实现（C/C++ 通用，使用 TLS 保存/恢复）
+#define DIY_LOG_CODE(LEVEL_FUNC, KEY, LEVEL_INT, CODE_INT, ...)                           \
+    do {                                                                                  \
+        /* 备份并设置本次日志 code */                                                     \
+        int __old_code__ = g_log_code_tls;                                                \
+        g_log_code_tls = static_cast<int>(CODE_INT);                                      \
+                                                                                          \
+        /* 组装消息 */                                                                    \
+        char __msg_buf__[256];                                                            \
+        format_log_msg(__msg_buf__, sizeof(__msg_buf__), __VA_ARGS__);                    \
+                                                                                          \
+        /* 将 KEY 规范化为字符串（兼容 const char* / std::string 等） */                 \
+        std::ostringstream __oss_key__;                                                   \
+        __oss_key__ << (KEY);                                                             \
+        std::string __key_raw__ = __oss_key__.str();                                      \
+                                                                                          \
+        /* 根据 LEVEL_INT 构造最终 key */                                                 \
+        char __key_buf__[256];                                                            \
+        switch (static_cast<int>(LEVEL_INT)) {                                            \
+            case 0:                                                                       \
+                std::snprintf(__key_buf__, sizeof(__key_buf__), "process");               \
+                break;                                                                    \
+            case 1:                                                                       \
+                std::snprintf(__key_buf__, sizeof(__key_buf__),                           \
+                              "terminal.%s", __key_raw__.c_str());                        \
+                break;                                                                    \
+            case 2:                                                                       \
+                std::snprintf(__key_buf__, sizeof(__key_buf__),                           \
+                              "monitor.%s", __key_raw__.c_str());                         \
+                break;                                                                    \
+            default:                                                                      \
+                std::snprintf(__key_buf__, sizeof(__key_buf__), "%s",                     \
+                              __key_raw__.c_str());                                       \
+                break;                                                                    \
+        }                                                                                 \
+                                                                                          \
+        /* 输出日志 */                                                                    \
+        LEVEL_FUNC(__key_buf__, __msg_buf__);                                             \
+                                                                                          \
+        /* 恢复旧值 */                                                                    \
+        g_log_code_tls = __old_code__;                                                    \
+    } while (0)
+
+// ★修改：默认宏改为 code=0（兼容原有用法）
+#define DIY_ERRORLOG(KEY, ...) DIY_LOG_CODE(log_error, KEY, 0, LOG_CODE_OTHER,__VA_ARGS__)   // ★修改：默认 code=0
+#define DIY_WARNLOG(KEY,  ...) DIY_LOG_CODE(log_warn,  KEY, 0, LOG_CODE_OTHER,__VA_ARGS__)   // ★修改：默认 code=0
+#define DIY_INFOLOG(KEY,  ...) DIY_LOG_CODE(log_info,  KEY, 0, LOG_CODE_OTHER,__VA_ARGS__)   // ★修改：默认 code=0
+#define DIY_DEBUGLOG(KEY, ...) DIY_LOG_CODE(log_debug, KEY, 0, LOG_CODE_OTHER,__VA_ARGS__)   // ★修改：默认 code=0
+
+// ★新增：显式传入 code 的便捷宏
+// 用法示例：DIY_WARNLOG_CODE(full_key_m_c, warn_recallstart, "【WARN】监测点:%s ...", ...);
+#define DIY_ERRORLOG_CODE(KEY, LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_error, KEY, LEVEL_INT, CODE_INT, __VA_ARGS__)  // ★新增
+#define DIY_WARNLOG_CODE(KEY,  LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_warn,  KEY, LEVEL_INT, CODE_INT, __VA_ARGS__)  // ★新增
+#define DIY_INFOLOG_CODE(KEY,  LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_info,  KEY, LEVEL_INT, CODE_INT, __VA_ARGS__)  // ★新增
+#define DIY_DEBUGLOG_CODE(KEY, LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_debug, KEY, LEVEL_INT, CODE_INT, __VA_ARGS__)  // ★新增
+
+
+// ====================== 日志宏区域 ======================
 
 #ifdef __cplusplus
 }

LFtid1056/cloudfront/code/log4cplus/log4.h

@@ -1,111 +0,0 @@
-#ifndef LOG4_H
-#define LOG4_H
-
-#ifdef __cplusplus
-
-#include <string>
-#include <map>
-
-//防止#include <log4cplus/logger.h>里的冲突
-#ifdef min
-#undef min
-#endif
-
-#ifdef max
-#undef max
-#endif
-//防止#include <log4cplus/logger.h>里的冲突
-
-#include "logger.h"
-#include <set>
-#include "loggingmacros.h"
-
-#include "appender.h"
-
-#define LOGTYPE_COM 1
-#define LOGTYPE_DATA 2
-
-struct TypedLogger {
-    log4cplus::Logger logger;
-    int logtype;
-    TypedLogger();
-    TypedLogger(const log4cplus::Logger& l, int t);
-};
-
-struct DebugSwitch {
-    bool debug_open;
-    std::set<std::string> targets;
-    int min_level;
-    std::map<int, bool> type_enable;
-
-    DebugSwitch();
-    void open();
-    void close();
-    void set_target(const std::string& name);
-    void set_level(int level);
-    void enable_type(int type);
-    void disable_type(int type);
-    bool match(const std::string& logger_name, int level, int logtype);
-};
-
-extern std::map<std::string, TypedLogger> logger_map;
-extern DebugSwitch g_debug_switch;
-
-extern void send_reply_to_queue(const std::string& guid, const std::string& step, const std::string& result);
-
-
-//std::string get_front_type_from_subdir();
-
-
-// 不带 Appender 的版本
-log4cplus::Logger init_logger(const std::string& full_name,
-                              const std::string& file_dir,
-                              const std::string& base_file);
-
-// 带 Appender 的版本
-log4cplus::Logger init_logger(const std::string& full_name,
-                              const std::string& file_dir,
-                              const std::string& base_file,
-                              log4cplus::SharedAppenderPtr fileAppender);
-
-void process_log_command(const std::string& id, const std::string& level, const std::string& grade, const std::string& logtype_str);
-
-
-void update_log_entries_countdown();
-
-extern "C" {
-#endif
-void remove_loggers_by_terminal_id(const std::string& terminal_id_cstr);
-void init_logger_process();
-void init_loggers();
-void init_loggers_bydevid(const std::string& dev_id);
-
-void log_debug(const char* key, const char* msg);
-void log_info(const char* key, const char* msg);
-void log_warn(const char* key, const char* msg);
-void log_error(const char* key, const char* msg);
-
-void send_reply_to_queue_c(const char* guid, const char* step, const char* result);
-void format_log_msg(char* buf, size_t buf_size, const char* fmt, ...);
-
-//宏定义
-#define DIY_LOG(LEVEL_FUNC, KEY, ...)                             \
-    do {                                                          \
-        char buf[256];                                            \
-        format_log_msg(buf, sizeof(buf), __VA_ARGS__);            \
-        LEVEL_FUNC(KEY, buf);                                     \
-    } while (0)
-
-#define DIY_ERRORLOG(KEY, ...) DIY_LOG(log_error, KEY, __VA_ARGS__)
-#define DIY_WARNLOG(KEY, ...)  DIY_LOG(log_warn,  KEY, __VA_ARGS__)
-#define DIY_INFOLOG(KEY, ...)  DIY_LOG(log_info,  KEY, __VA_ARGS__)
-#define DIY_DEBUGLOG(KEY, ...) DIY_LOG(log_debug, KEY, __VA_ARGS__)
-
-
-#ifdef __cplusplus
-}
-#endif
-
-
-
-#endif // LOG4_H

LFtid1056/cloudfront/code/main.cpp

@@ -29,7 +29,7 @@
 ////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 #include "interface.h"          //用于访问接口
-#include "log4cplus/log4.h"     //用于日志
+#include "log4.h"     //用于日志
 #include "curl/curl.h"          //用于访问接口
 #include "nlohmann/json.hpp"    //用于构造json
 #include "worker.h"             //shell接口
@@ -401,14 +401,14 @@ void Front::mqconsumerThread()
         std::vector<rocketmq::Subscription> subscriptions;
 
         //if (g_node_id == THREE_SECS_DATA_BASE_NODE_ID) {
-        subscriptions.emplace_back(FRONT_INST + "_" + G_MQCONSUMER_TOPIC_RT, G_MQCONSUMER_TAG_RT, myMessageCallbackrtdata);
+        subscriptions.emplace_back(G_MQCONSUMER_TOPIC_RT, FRONT_INST, myMessageCallbackrtdata);
         //}
-        subscriptions.emplace_back(FRONT_INST + "_" + G_MQCONSUMER_TOPIC_UD, G_MQCONSUMER_TAG_UD, myMessageCallbackupdate);
+        subscriptions.emplace_back(G_MQCONSUMER_TOPIC_UD, FRONT_INST, myMessageCallbackupdate);
         //if (g_node_id == RECALL_HIS_DATA_BASE_NODE_ID) {
-        subscriptions.emplace_back(FRONT_INST + "_" + G_MQCONSUMER_TOPIC_RC, G_MQCONSUMER_TAG_RC, myMessageCallbackrecall);
+        subscriptions.emplace_back(G_MQCONSUMER_TOPIC_RC, FRONT_INST, myMessageCallbackrecall);
         //}
-        subscriptions.emplace_back(FRONT_INST + "_" + G_MQCONSUMER_TOPIC_SET, G_MQCONSUMER_TAG_SET, myMessageCallbackset);
-        subscriptions.emplace_back(FRONT_INST + "_" + G_MQCONSUMER_TOPIC_LOG, G_MQCONSUMER_TAG_LOG, myMessageCallbacklog);
+        subscriptions.emplace_back(G_MQCONSUMER_TOPIC_SET, FRONT_INST, myMessageCallbackset);
+        subscriptions.emplace_back(G_MQCONSUMER_TOPIC_LOG, FRONT_INST, myMessageCallbacklog);
 
         m_mqConsumer = make_unique<rocketmq::DefaultMQPushConsumer>(consumerGroup);
         m_mqConsumer->setNamesrvAddr(nameServer);

LFtid1056/cloudfront/code/rocketmq.cpp

@@ -33,7 +33,7 @@
 
 #include "rocketmq.h"
 #include "nlohmann/json.hpp"
-#include "log4cplus/log4.h"
+#include "log4.h"
 #include "interface.h"
 #include "front.h"
 
@@ -266,14 +266,16 @@ void ShutdownAndDestroyProducer()
 // 使用 C++ 接口封装的 RocketMQProducer 类
 void rocketmq_producer_send(rocketmq::RocketMQProducer* producer,
                                    const std::string& body,
-                                   const std::string& topic) {
+                                   const std::string& topic,                               
+                                   const std::string& tags,
+                                   const std::string& keys) {
     if (!producer) {
         std::cerr << "[rocketmq_producer_send] producer 不可用，未初始化\n";
         return;
     }
 
-    const std::string& tags = G_ROCKETMQ_TAG;
-    const std::string& keys = G_ROCKETMQ_KEY;
+    //const std::string& tags = G_ROCKETMQ_TAG;
+    //const std::string& keys = G_ROCKETMQ_KEY;
 
     try {
         producer->sendMessage(body, topic, tags, keys);
@@ -306,7 +308,9 @@ void my_rocketmq_send(queue_data_t& data,rocketmq::RocketMQProducer* producer)
 		init = true;
 	}
 
-	std::string key = data.mp_id;
+	std::string key = data.key;
+    std::string tag = data.tag;
+
 	std::string senddata = data.strText;
 	if (data.strTopic == "HISDATA")
 	{
@@ -334,7 +338,7 @@ void my_rocketmq_send(queue_data_t& data,rocketmq::RocketMQProducer* producer)
 		
 	}
 
-	rocketmq_producer_send(producer,senddata,topic);
+	rocketmq_producer_send(producer,senddata,topic,tag,key);
 }
 
 /////////////////////////////////////////////////////////////////////////////////////////////////查找台账下标
@@ -1247,7 +1251,7 @@ std::string prepare_update(const std::string& code_str, const terminal_dev& json
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////终端连接消息
 
-void connect_status_to_queue(const std::string& id, const std::string& datetime, int status)
+void connect_status_to_queue(const std::string& id, const std::string& datetime, int status)//这个不使用，使用新的带有时间封装的
 {
     try {
         // 构造 JSON
@@ -1260,6 +1264,8 @@ void connect_status_to_queue(const std::string& id, const std::string& datetime,
         queue_data_t data;
         data.strTopic = G_CONNECT_TOPIC;
         data.strText = jsonObject.dump();  // 转换为字符串
+        data.tag = G_CONNECT_TAG;
+        data.key = G_CONNECT_KEY;
 
         //if (g_node_id == STAT_DATA_BASE_NODE_ID) {
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
@@ -1288,6 +1294,8 @@ void send_reply_to_queue(const std::string& guid, const std::string& step, const
         queue_data_t connect_info;
         connect_info.strTopic = Topic_Reply_Topic;
         connect_info.strText = obj.dump();  // 序列化为 JSON 字符串
+        connect_info.tag = Topic_Reply_Tag;
+        connect_info.key = Topic_Reply_Key;
 
         // 加入发送队列（线程安全）
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
@@ -1311,6 +1319,8 @@ void send_heartbeat_to_queue(const std::string& status) {
         queue_data_t connect_info;
         connect_info.strTopic = Heart_Beat_Topic;
         connect_info.strText = obj.dump();  // 紧凑格式 JSON
+        connect_info.tag = Heart_Beat_Tag;
+        connect_info.key = Heart_Beat_Key;
 
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);
         queue_data_list.push_back(connect_info);
@@ -1412,6 +1422,8 @@ void rocketmq_test_300(int mpnum, int front_index, int type, Front* front) {
                     }
 
                     data.strText = modified_strText;
+                    data.tag = G_ROCKETMQ_TAG_TEST;
+                    data.key = G_ROCKETMQ_KEY_TEST;
                     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
                     queue_data_list.push_back(data);
 
@@ -1449,6 +1461,8 @@ void rocketmq_test_300(int mpnum, int front_index, int type, Front* front) {
                 }
 
                 data.strText = modified_strText;
+                data.tag = G_ROCKETMQ_TAG_TEST;
+                data.key = G_ROCKETMQ_KEY_TEST;
                 std::lock_guard<std::mutex> lock(queue_data_list_mutex);
                 queue_data_list.push_back(data);
 
@@ -1482,7 +1496,8 @@ void rocketmq_test_rt(Front* front)//用来测试实时数据
 
     data.strText = std::string(buffer.str());
     data.mp_id = "123123";
-    //my_rocketmq_send(data,front->m_producer);
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -1505,7 +1520,8 @@ void rocketmq_test_ud(Front* front)//用来测试台账更新
 
     data.strText = std::string(buffer.str());
     data.mp_id = "123123";
-    //my_rocketmq_send(data,front->m_producer);
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -1528,7 +1544,8 @@ void rocketmq_test_set(Front* front)//用来测试进程控制脚本
 
     data.strText = std::string(buffer.str());
     data.mp_id = "123123";
-    //my_rocketmq_send(data,front->m_producer);
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -1551,7 +1568,8 @@ void rocketmq_test_rc(Front* front)//用来测试补招
 
     data.strText = std::string(buffer.str());
     data.mp_id = "123123";
-    //my_rocketmq_send(data,front->m_producer);
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -1890,6 +1908,8 @@ void send_reply_to_cloud(int reply_code, const std::string& dev_id, int type) {
         queue_data_t connect_info;
         connect_info.strTopic = Topic_Reply_Topic;
         connect_info.strText  = obj.dump();  // 序列化为字符串
+        connect_info.tag      = Topic_Reply_Tag;
+        connect_info.key      = Topic_Reply_Key;
 
         {
             std::lock_guard<std::mutex> lock(queue_data_list_mutex);
@@ -1981,6 +2001,8 @@ void rocketmq_test_getdir(Front* front)//用来测试目录获取
 
     data.strText = std::string(buffer.str());
     data.mp_id = "123123";
+    data.tag = G_ROCKETMQ_TAG_TEST;
+    data.key = G_ROCKETMQ_KEY_TEST;
     std::lock_guard<std::mutex> lock(queue_data_list_mutex);
     queue_data_list.push_back(data);
 }
@@ -2008,6 +2030,8 @@ void connect_status_update(const std::string& id, int status)
     queue_data_t connect_info;
     connect_info.strTopic = G_CONNECT_TOPIC;
     connect_info.strText  = j.dump(); // 转成字符串
+    connect_info.tag      = G_CONNECT_TAG;
+    connect_info.key      = G_CONNECT_KEY;
 
     {
         std::lock_guard<std::mutex> lock(queue_data_list_mutex);

LFtid1056/cloudfront/code/rocketmq.h

@@ -60,6 +60,10 @@ extern std::string TOPIC_RTDATA;
 extern std::string G_ROCKETMQ_TAG;
 extern std::string G_ROCKETMQ_KEY;
 
+//添加rt的tagkey
+extern std::string G_RT_TAG;
+extern std::string G_RT_KEY;
+
 extern std::string G_ROCKETMQ_CONSUMER;
 extern std::string G_MQCONSUMER_IPPORT;
 extern std::string G_MQCONSUMER_ACCESSKEY;

LFtid1056/cloudfront/code/worker.cpp

@@ -35,7 +35,7 @@
 #include "worker.h"
 #include "interface.h"
 #include "rocketmq.h"
-#include "log4cplus/log4.h"
+#include "log4.h"
 #include "front.h"
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////

LFtid1056/config/front.cfg

@@ -11,8 +11,11 @@ SngTopic=SngTopic
 QUEUE_TAG=stat
 QUEUE_KEY=stat
 
+RT_TAG=rt
+RT_KEY=rt
+
 [Flag]
-FrontInst=
+FrontInst=914b94563ca7f272c90ee8580ed6adc6
 FrontIP=192.168.1.138
 
 [Ledger]
@@ -30,7 +33,7 @@ WebFiledownload=
 
 [RocketMq]
 producer=Group_producer
-Ipport=192.168.1.24:9876
+Ipport=192.168.1.103:9876
 AccessKey=rmqroot
 SecretKey=001@#njcnmq
 
@@ -41,10 +44,10 @@ Testflag=1
 Testnum=0
 Testtype=0
 TestPort=11000
-TestList=8ad28e2e36dfbd19906f9e2a4894b375
+TestList=
 
 consumer=Group_consumer
-ConsumerIpport=192.168.1.24:9876
+ConsumerIpport=192.168.1.103:9876
 ConsumerTopicRT=ask_real_data_topic
 ConsumerTagRT=Test_Tag
 ConsumerKeyRT=Test_Keys