modify recall

LFtid1056/cloudfront/code/cfg_parser.cpp

@@ -205,7 +205,10 @@ std::vector<std::string> TESTARRAY;		//解析的列表数组
 
 //////////////////////////////////////////////////////////////////////////////////////////////////////////////////其他文件定义的函数引用声明
 
-
+bool enqueue_direct_download(const std::string& dev_id,
+                             const std::string& monitor_id,
+                             const std::string& filename,
+                             const std::vector<std::string>& dir_candidates);
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////当前文件函数声明
 
@@ -1060,6 +1063,49 @@ void create_recall_xml()
 }
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////补招部分
 // 工具函数：将时间字符串转为 time_t（秒级）
+// ▲新增：从 monitorId 提取结尾数字（不含前导非数字部分），失败返回空串
+static std::string extract_monitor_digits(const std::string& monitorId) {
+    // 例： "00B78D0171091" -> "171091"（按你的“对应的数字”的规则可定制）
+    // 这里实现：取 monitorId 中最后一段连续数字
+    std::string digits;
+    for (int i = static_cast<int>(monitorId.size()) - 1; i >= 0; --i) {
+        if (std::isdigit(static_cast<unsigned char>(monitorId[i]))) {
+            digits.push_back(monitorId[i]);
+        } else if (!digits.empty()) {
+            break;
+        }
+    }
+    std::reverse(digits.begin(), digits.end());
+    return digits;
+}
+
+// ▲新增：把 "YYYY-MM-DD HH:MM:SS[.ffffff]" -> "YYYYMMDDHHMMSS"（忽略小数部分）
+static std::string compact_ts_for_filename(const std::string& ts) {
+    // 允许 "2025-10-10 14:38:07.000000"
+    // 输出 "20251010143807"
+    std::string ymdhms;
+    ymdhms.reserve(14);
+    for (size_t i = 0; i < ts.size(); ++i) {
+        char c = ts[i];
+        if (std::isdigit(static_cast<unsigned char>(c))) {
+            ymdhms.push_back(c);
+            if (ymdhms.size() == 14) break;
+        }
+    }
+    return (ymdhms.size() == 14) ? ymdhms : std::string();
+}
+
+// ▲新增：按“数字_时间后缀.后缀”拼直下文件名（返回{*.cfg, *.dat}两种）
+static std::vector<std::string> build_direct_filenames(const std::string& monitorDigits,
+                                                       const std::string& ts_compact)
+{
+    std::vector<std::string> out;
+    if (monitorDigits.empty() || ts_compact.empty()) return out;
+    out.push_back(monitorDigits + "_" + ts_compact + ".cfg");
+    out.push_back(monitorDigits + "_" + ts_compact + ".dat");
+    return out;
+}
+
 static long long parse_time_to_epoch(const std::string& time_str) {
     std::tm tm = {};
     std::istringstream ss(time_str);
@@ -1138,9 +1184,9 @@ int recall_json_handle_from_mq(const std::string& body)
         }
 
         // 解析 messageBody 内层 JSON
-        nlohmann::json messageBody;
+        nlohmann::json mb;
         try {
-            messageBody = nlohmann::json::parse(messageBodyStr);
+            mb = nlohmann::json::parse(messageBodyStr);
         } catch (const std::exception& e) {
             std::cerr << "Failed to parse 'messageBody' JSON: " << e.what() << std::endl;
             DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,messageBody的json结构不正确",
@@ -1148,209 +1194,147 @@ int recall_json_handle_from_mq(const std::string& body)
             return 10004;
         }
 
-        // 提取 guid 并立即回执
+        if (mb.is_array()) {
-        if (!messageBody.contains("guid") || !messageBody["guid"].is_string()) {
+            // ====== 新格式（数组）：支持 dataType=0/1 的区间补招 & dataType=2 的直下文件 ======
-            std::cerr << "'guid' is missing or is not a string" << std::endl;
+            for (const auto& rec : mb) {
-            DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有guid字段",
+                if (!rec.is_object()) continue;
-                         g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
+
-            return 10004;
+                // 必要字段
+                std::string guid       = rec.value("guid", "");
+                std::string terminalId = rec.value("terminalId", "");
+                if (terminalId.empty()) continue;
+
+                // ▲dataType 可能是字符串"0"/"1"或数字2
+                int dt = -1;
+                if (rec["dataType"].is_number_integer()) {
+                    dt = rec["dataType"].get<int>();
+                } else if (rec["dataType"].is_string()) {
+                    std::string s = rec["dataType"].get<std::string>();
+                    if (s == "0") dt = 0; else if (s == "1") dt = 1;
                 }
-        std::string guid = messageBody["guid"].get<std::string>();
-        //send_reply_to_queue(guid, static_cast<int>(ResponseCode::OK), "收到补招指令");
 
-        // 提取 data 数组
+                // 统一 monitorId 为数组形式
-        if (!messageBody.contains("data") || !messageBody["data"].is_array()) {
+                std::vector<std::string> monitors;
-            std::cerr << "'data' is missing or is not an array" << std::endl;
+                if (rec.contains("monitorId")) {
-            DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有data字段",
+                    if (rec["monitorId"].is_array()) {
-                         g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
+                        for (auto& m : rec["monitorId"]) if (m.is_string()) monitors.push_back(m.get<std::string>());
-            return 10004;
+                    } else if (rec["monitorId"].is_string()) {
+                        monitors.push_back(rec["monitorId"].get<std::string>());
                     }
-        // 仅用于保留你原先的调试输出
+                }
-        std::string data_dump;
+                if (monitors.empty()) continue;
-        try { data_dump = messageBody["data"].dump(); } catch (...) { data_dump.clear(); }
-        std::cout << "parseJsonMessageRC: " << data_dump << std::endl;
 
-        // 不指定稳态/暂态则全部补招
+                // ▲沿用：校验终端归属 + 在线性
-        int stat = 0;
-        int voltage = 0;
-
-        // 2. 遍历每个补招项（这里直接用已解析的 messageBody["data"]）
-        for (auto& item : messageBody["data"]) {
-            // 获取必需字段
-            // ★修改：强制要求 terminalId；
-            if (!item.contains("terminalId") ||
-                !item.contains("monitor") ||
-                !item.contains("timeInterval") ||
-                !item.contains("dataType"))
                 {
-                std::cout << "json内容解析错误" << std::endl;
-                return 10000;
-            }
-
-            // ★新增：读取 terminalId（必填）
-            std::string terminalId = item["terminalId"].get<std::string>();
-            if (terminalId.empty()) {
-                std::cout << "terminalId为空" << std::endl;
-                continue;
-            }
-
-            // 2.1 解析 dataType（仅保留稳态/暂态）
-            std::string datatype = item["dataType"].get<std::string>();
-            if (!datatype.empty()) {
-                if (datatype == "0" || datatype == "稳态" || datatype == "steady" || datatype == "STEADY") {
-                    stat = 1;  voltage = 0;   // 稳态
-                } else if (datatype == "1" || datatype == "暂态" || datatype == "voltage" || datatype == "VOLTAGE") {
-                    stat = 0;  voltage = 1;   // 暂态
-                } else {
-                    stat = voltage = 1;       // 其他情况按全补
-                }
-            } else {
-                stat = voltage = 1;           // 全补
-            }
-
-            // ★新增：定位并校验该 terminal 是否归属当前进程
                     std::lock_guard<std::mutex> lock(ledgermtx);
-
                     const terminal_dev* targetDev = NULL;
-            for (std::vector<terminal_dev>::const_iterator it = terminal_devlist.begin();
+                    for (std::vector<terminal_dev>::const_iterator it = terminal_devlist.begin(); it != terminal_devlist.end(); ++it) {
-                 it != terminal_devlist.end(); ++it)
+                        if (it->terminal_id == terminalId) { targetDev = &(*it); break; }
-            {
-                if (it->terminal_id == terminalId) {
-                    targetDev = &(*it);
-                    break;
                     }
-            }
+                    if (!targetDev) { std::cout << "terminalId未匹配当前进程内装置: " << terminalId << std::endl; continue; }
-            if (!targetDev) {
-                std::cout << "terminalId未匹配当前进程内装置: " << terminalId << std::endl;
-                continue;
-            }
-
-            // 添加判断装置在线，不注册guid，异步补招
                     if (ClientManager::instance().get_dev_status(targetDev->terminal_id) != 1) {
                         std::cout << "terminalId对应装置不在线: " << targetDev->terminal_id << std::endl;
-
-                // 响应 web
                         std::string msg = std::string("装置：") + targetDev->terminal_name + " 不在线，无法补招";
                         send_reply_to_kafka_recall("12345", "2", static_cast<int>(ResponseCode::INTERNAL_ERROR), msg, targetDev->terminal_id, "", "", "");
-                continue;//处理下一个装置的补招记录
-            }
-
-            // ★新增：按新结构解析 monitor 层级
-            nlohmann::json& monitors = item["monitor"];
-            if (!monitors.is_array() || monitors.empty()) {
-                std::cout << "monitor数组为空或非数组类型" << std::endl;
                         continue;
                     }
-
-            for (auto& mobj : monitors) {
-                if (!mobj.contains("monitorId") || !mobj.contains("timeInterval")) {
-                    std::cout << "monitor项缺少 monitorId 或 timeInterval" << std::endl;
-                    continue;
                 }
 
-                std::string monitorId = mobj["monitorId"].get<std::string>();
+                if (dt == 2) {
-                if (monitorId.empty()) continue;
+                    // ▲直下文件：timeList -> fun1/fun2 -> enqueue_direct_download
+                    if (!rec.contains("timeList") || !rec["timeList"].is_array()) continue;
 
-                // 不只是判断存在，还要拿到指针 lm 以便后续 push_back
+                    for (const auto& monId : monitors) {
+                        // fun1：提取 monitor 数字
+                        std::string digits = extract_monitor_digits(monId);
+                        if (digits.empty()) { std::cout << "monitorId数字提取失败: " << monId << std::endl; continue; }
+
+                        for (const auto& t : rec["timeList"]) {
+                            if (!t.is_string()) continue;
+                            std::string ts_compact = compact_ts_for_filename(t.get<std::string>());
+                            if (ts_compact.empty()) { std::cout << "时间解析失败: " << t << std::endl; continue; }
+
+                            // fun2：生成 *.cfg/*.dat 两个文件名
+                            std::vector<std::string> fns = build_direct_filenames(digits, ts_compact);
+                            // 加入候选目录（使用 RecallFile 缺省的四个；如需定制可在此传自定义列表）
+                            static const std::vector<std::string> DIRS {
+                                "/cf/COMTRADE", "/bd0/COMTRADE", "/sd0/COMTRADE", "/sd0:1/COMTRADE"
+                            };
+                            for (const auto& fn : fns) {
+                                bool ok = enqueue_direct_download(terminalId, monId, fn, DIRS);
+                                std::cout << "[direct] enqueue " << (ok ? "ok " : "fail ")
+                                          << "dev=" << terminalId << " mon=" << monId
+                                          << " file=" << fn << std::endl;
+                            }
+                        }
+                    }
+                } else if (dt == 0 || dt == 1) {
+                    // ▲保持老逻辑（与“对象+data”一致）：timeInterval 数组
+                    if (!rec.contains("timeInterval") || !rec["timeInterval"].is_array()) continue;
+
+                    // 解析 dataType-> stat/voltage
+                    int stat = (dt == 0) ? 1 : 0;
+                    int voltage = (dt == 1) ? 1 : 0;
+
+                    // 把每个 monitor 的区间写入 recall_list / recall_list_static
+                    std::lock_guard<std::mutex> lock(ledgermtx);
+                    // 找终端
+                    terminal_dev* dev_nc = NULL;
+                    for (auto& d : terminal_devlist) if (d.terminal_id == terminalId) { dev_nc = &d; break; }
+                    if (!dev_nc) continue;
+
+                    for (const auto& monId : monitors) {
+                        // 找监测点
                         ledger_monitor* lm = NULL;
-                // 注意：这里需要非常量指针，取 const_cast 后遍历
+                        for (auto itLm = dev_nc->line.begin(); itLm != dev_nc->line.end(); ++itLm) {
-                terminal_dev* dev_nc = const_cast<terminal_dev*>(targetDev);
+                            if (!itLm->monitor_id.empty() && itLm->monitor_id == monId) { lm = &(*itLm); break; }
-                for (std::vector<ledger_monitor>::iterator itLm = dev_nc->line.begin();
-                     itLm != dev_nc->line.end(); ++itLm)
-                {
-                    if (!itLm->monitor_id.empty() && itLm->monitor_id == monitorId) {
-                        lm = &(*itLm);
-                        break;
-                    }
-                }
-                if (!lm) {
-                    std::cout << "monitorId未在terminal内找到: " << monitorId
-                              << " @ " << terminalId << std::endl;
-                    continue;
                         }
+                        if (!lm) { std::cout << "monitorId未在terminal内找到: " << monId << " @ " << terminalId << std::endl; continue; }
 
-                nlohmann::json& tiArr = mobj["timeInterval"];
+                        for (const auto& ti : rec["timeInterval"]) {
-                if (!tiArr.is_array() || tiArr.empty()) {
+                            if (!ti.is_string()) continue;
-                    std::cout << "timeInterval为空或非数组类型: monitorId=" << monitorId << std::endl;
+                            std::string s = ti.get<std::string>();
-                    continue;
+                            std::string::size_type pos = s.find('~');
-                }
+                            if (pos == std::string::npos) { std::cout << "timeInterval格式错误: " << s << std::endl; continue; }
+                            std::string start = s.substr(0, pos);
+                            std::string end   = s.substr(pos + 1);
 
-                // 这里拆分时间段并 push 到 lm->recall_list / lm->recall_list_static
+                            RecallFile rm_all;
-                for (auto& timeItem : tiArr) {
+                            rm_all.recall_status = 0;
-                    std::string ti = timeItem.get<std::string>();
+                            rm_all.StartTime     = start;
-                    std::string::size_type pos = ti.find('~');
-                    if (pos == std::string::npos) {
-                        std::cout << "timeInterval格式错误: " << ti << std::endl;
-                        continue;
-                    }
-                    std::string start = ti.substr(0, pos);
-                    std::string end   = ti.substr(pos + 1);
-
-                    // 仅对 recall_list（事件）进行 1 小时拆分；recall_list_static（稳态）不拆分
-                    {
-                        // 公共字段（整体区间，不拆分）用于 recall_list_static
-                        RecallFile rm_all;                 // ★类型正确：稳态列表的元素
-                        rm_all.recall_status = 0;          // 初始状态：未补招
-                        rm_all.StartTime     = start;      // 直接使用字符串
                             rm_all.EndTime       = end;
                             rm_all.STEADY        = std::to_string(stat);
                             rm_all.VOLTAGE       = std::to_string(voltage);
 
-                        // 仅当需要事件补招（voltage==1）时，才进行 1 小时拆分并压入 recall_list
                             if (voltage == 1) {
-                            // 拆分时间段为 1 小时一段，并存入 recall_list
                                 std::vector<RecallInfo> recallinfo_list_hour;
                                 Get_Recall_Time_Char(start, end, recallinfo_list_hour);
-
+                                for (size_t i = 0; i < recallinfo_list_hour.size(); ++i) {
-                            for (std::size_t i = 0; i < recallinfo_list_hour.size(); ++i) {
                                     const RecallInfo& info = recallinfo_list_hour[i];
-
+                                    RecallMonitor rm;
-                                RecallMonitor rm;             // ★类型正确：事件列表的元素
+                                    rm.recall_status = 0;
-                                rm.recall_status = 0;         // 初始状态：未补招
                                     rm.StartTime     = epoch_to_datetime_str(info.starttime);
                                     rm.EndTime       = epoch_to_datetime_str(info.endtime);
                                     rm.STEADY        = std::to_string(stat);
                                     rm.VOLTAGE       = std::to_string(voltage);
-
                                     lm->recall_list.push_back(rm);
-
-                                // 事件补招列表（recall_list）调试打印
-                                std::cout << "[recall_json_handle] terminal=" << terminalId
-                                          << " monitor=" << monitorId
-                                          << " [recall_list] start=" << lm->recall_list.back().StartTime
-                                          << " end="   << lm->recall_list.back().EndTime
-                                          << " steady="<< lm->recall_list.back().STEADY
-                                          << " voltage="<< lm->recall_list.back().VOLTAGE
-                                          << std::endl;
                                 }
                             }
-
-                        // 仅当需要稳态补招（stat==1）时，不拆分，直接压入 recall_list_static
                             if (stat == 1) {
-                            lm->recall_list_static.push_back(rm_all); // 不拆分，整体区间
+                                lm->recall_list_static.push_back(rm_all);
-
-                            // 稳态补招列表（recall_list_static）调试打印
-                            std::cout << "[recall_json_handle] terminal=" << terminalId
-                                      << " monitor=" << monitorId
-                                      << " [recall_list_static] start=" << lm->recall_list_static.back().StartTime
-                                      << " end="   << lm->recall_list_static.back().EndTime
-                                      << " steady="<< lm->recall_list_static.back().STEADY
-                                      << " voltage="<< lm->recall_list_static.back().VOLTAGE
-                                      << std::endl;
-                        }
-
-                        // 非法输入保护（保留你原来的保护与返回码）
-                        if (stat == 0 && voltage == 0) {
-                            std::cout << "[recall_json_handle] skip: stat=0 && voltage=0, monitor=" << monitorId
-                                      << " terminal=" << terminalId
-                                      << " start=" << rm_all.StartTime
-                                      << " end="   << rm_all.EndTime
-                                      << std::endl;
-                            return 10003;
                             }
+                            if (stat == 0 && voltage == 0) return 10003;
+                        }
+                    }
+                } else {
+                    // 未知 dataType，忽略
+                    continue;
                 }
             }
         }
+        else {
+            // 不支持的 messageBody 形态
+            return 10004;
         }
     }
     catch (const std::exception& e) {
@@ -4355,15 +4339,14 @@ void check_recall_file() {
                             auto it = front.dir_files.find(front.cur_dir);
                             if (it != front.dir_files.end()) {
                                 if (front.direct_mode) {
-                                    // ★新增：直下文件（精确匹配文件名）
+                                    // ▲直下：支持“目标名列表”
+                                    std::set<std::string> want(front.target_filenames.begin(), front.target_filenames.end());
                                     for (const auto& ent : it->second) {
                                         if (ent.flag != 1) continue; // 只要文件
-                                        // 安全从 char[64] 转成 std::string
                                         size_t n = ::strnlen(ent.name, sizeof(ent.name));
                                         std::string fname(ent.name, n);
-                                        if (fname == front.target_filename) {
+                                        if (want.find(fname) != want.end()) {
                                             front.download_queue.push_back(front.cur_dir + "/" + fname);
-                                            break; // 找到一个就足够
                                         }
                                     }
                                 } else {
@@ -4546,7 +4529,7 @@ bool enqueue_direct_download(const std::string& dev_id,
     rf.EndTime         = "1970-01-01 00:00:01";
     rf.dir_candidates  = dir_candidates;          // 传入要检索的目录列表
     rf.direct_mode     = true;                    // ★关键：直下文件
-    rf.target_filename = filename;                // ★关键：匹配的目标文件名
+    rf.target_filenames.push_back(filename);      // ▲单个文件名入“列表”
 
     lm_it->recall_list_static.push_back(std::move(rf));
 

LFtid1056/cloudfront/code/interface.h

@@ -76,7 +76,7 @@ public:
 
     //暂态文件用
     bool        direct_mode = false;     // 直下文件开关：true 表示不按时间窗，仅按目标文件名
-    std::string target_filename;         // 直下文件名（不含目录）
+    std::vector<std::string>    target_filenames;         // 直下文件名（不含目录）
 
     std::list<std::string> file_paths;     // 已下载/要上报的完整路径（用于最终结果）
 
@@ -118,7 +118,7 @@ public:
         // ★新增：按需保留直下文件开关和目标名
         if (!keep_direct) {
             direct_mode = false;
-            target_filename.clear();
+            target_filenames.clear(); // ▲列表清空
         }
     }
 };

LFtid1056/cloudfront/code/rocketmq.cpp

@@ -88,6 +88,20 @@ bool createXmlFile(int devindex, int mpindex, bool realData, bool soeData, int l
 std::string prepare_update(const std::string& code_str, const terminal_dev& json_data,const std::string& guid);
 bool writeToFile(const std::string& filePath, const std::string& xmlContent);
 
+//////////////////////////////////////////////////////////////////////////////////////////////////////消费起始控制
+
+static const int64_t G_APP_START_MS = []() -> int64_t {
+    using namespace std::chrono;
+    return duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count();
+}();
+
+static const int64_t G_START_SKEW_MS = 1000; // 容错 1s，可按需调整
+
+static inline 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);
+}
+
 //////////////////////////////////////////////////////////////////////////////////////////////////////
 
 namespace rocketmq {
@@ -423,34 +437,49 @@ bool parseJsonMessageSET(const std::string& json_str) {
         return false;
     }
 
-    if (!root.contains("messageBody") || !root["messageBody"].is_string()) {
+    // [MOD] 允许 messageBody 既可为字符串也可为对象；保持原有错误打印
-        std::cerr << "'messageBody' is missing or is not a string" << std::endl;
+    // ----- MOD BEGIN: messageBody 兼容 string/object -----
+    json messageBody;
+    if (!root.contains("messageBody")) {
+        std::cerr << "missing 'messageBody'" << std::endl;
         return false;
     }
-
+    if (root["messageBody"].is_string()) {
-    std::string messageBodyStr = root["messageBody"];
+        std::string messageBodyStr = root["messageBody"].get<std::string>();
         if (messageBodyStr.empty()) {
             std::cerr << "'messageBody' is empty" << std::endl;
             return false;
         }
-
-    json messageBody;
         try {
             messageBody = json::parse(messageBodyStr);
         } catch (const std::exception& e) {
             std::cerr << "Failed to parse 'messageBody': " << e.what() << std::endl;
             return false;
         }
+    } else if (root["messageBody"].is_object()) {
+        messageBody = root["messageBody"];
+    } else {
+        std::cerr << "'messageBody' is neither string nor object" << std::endl;
+        return false;
+    }
+    // ----- MOD END: messageBody 兼容 string/object -----
 
-    // 获取字段
+    // [MOD] 基础必填字段仅校验 guid/code/processNo/fun；frontType、processNum 改为按功能分支再校验
-    if (!messageBody.contains("guid") || !messageBody.contains("code") ||
+    // ----- MOD BEGIN: 基础字段按需校验 -----
-        !messageBody.contains("processNo") || !messageBody.contains("fun") ||
+    if (!messageBody.contains("guid") ||
-        !messageBody.contains("frontType")) {
+        !messageBody.contains("code") ||
+        !messageBody.contains("processNo") ||
+        !messageBody.contains("fun")) {
         std::cout << "Missing one or more required fields in messageBody." << std::endl;
         return false;
     }
+    // ----- MOD END: 基础字段按需校验 -----
 
-    std::string guid, code_str, fun, frontType;
+    std::string guid, code_str, fun;
+    // [MOD] frontType 改为可选，给默认值 "all"
+    // ----- MOD BEGIN: frontType 可选 -----
+    std::string frontType = "all";
+    // ----- MOD END: frontType 可选 -----
     int index_value = 0;
 
     try {
@@ -458,13 +487,20 @@ bool parseJsonMessageSET(const std::string& json_str) {
         code_str    = messageBody["code"].get<std::string>();
         index_value = messageBody["processNo"].get<int>();
         fun         = messageBody["fun"].get<std::string>();
+
+        // [MOD] 仅当存在 frontType 且为 string 时再读取，保持兼容
+        // ----- MOD BEGIN: frontType 存在才解析 -----
+        if (messageBody.contains("frontType") && messageBody["frontType"].is_string()) {
             frontType = messageBody["frontType"].get<std::string>();
+        }
+        // ----- MOD END: frontType 存在才解析 -----
+
     } catch (const std::exception& e) {
         std::cerr << "Field parsing error: " << e.what() << std::endl;
         return false;
     }
 
-    // 判断进程号是否匹配
+    // 判断进程号是否匹配（保留原逻辑）
     if (index_value != g_front_seg_index && g_front_seg_index != 0) {
         std::cout << "msg index: " << index_value << " doesn't match self index: " << g_front_seg_index << std::endl;
         return true;
@@ -472,9 +508,17 @@ bool parseJsonMessageSET(const std::string& json_str) {
 
     std::cout << "msg index: " << index_value << "  self index: " << g_front_seg_index << std::endl;
 
-    DIY_INFOLOG("process", "【NORMAL】前置的%d号进程处理topic:%s_%s的进程控制消息", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_SET.c_str());
+    DIY_INFOLOG("process", "【NORMAL】前置的%d号进程处理topic:%s_%s的进程控制消息",
+                g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_SET.c_str());
 
     if (code_str == "set_process") {
+
+        // [MOD] 按功能分支分别校验参数：
+        //       reset/add 需要 processNum（且可选 frontType，默认 all）；
+        //       delete 不需要 frontType/processNum
+        // ----- MOD BEGIN: 分功能按需校验与执行 -----
+        if (fun == "reset" || fun == "add") {
+
             if (!messageBody.contains("processNum")) {
                 std::cout << "Missing 'processNum' in JSON." << std::endl;
                 return false;
@@ -488,9 +532,8 @@ bool parseJsonMessageSET(const std::string& json_str) {
                 return false;
             }
 
-        // 校验参数并执行
+            // 校验参数并执行（保留你原校验条件，frontType 允许默认 all）
-        if ((fun == "reset" || fun == "add") &&
+            if ((processNum >= 1 && processNum < 10) &&
-            (processNum >= 1 && processNum < 10) &&
                 (frontType == "cloudfront" || frontType == "all")) {
 
                 // if (g_node_id == STAT_DATA_BASE_NODE_ID && g_front_seg_index == 1) {
@@ -498,26 +541,34 @@ bool parseJsonMessageSET(const std::string& json_str) {
 
                     execute_bash(fun, processNum, frontType);
 
-                DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,reset表示重启所有进程,add表示添加进程", g_front_seg_index, fun.c_str());
+                    DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,reset表示重启所有进程,add表示添加进程",
+                                g_front_seg_index, fun.c_str());
 
                     send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到重置进程指令,重启所有进程!");
                     std::cout << "this msg should only execute once" << std::endl;
                 } else {
                     std::cout << "only cfg_stat_data index 1 can control process, this process not handle this msg" << std::endl;
                 }
-        }
+            } else {
-        else if (fun == "delete") {
-
-            send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到删除进程指令,这个进程将会重启 ");
-
-            DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,即将重启", g_front_seg_index, fun.c_str());
-
-            std::this_thread::sleep_for(std::chrono::seconds(10));
-            ::_exit(-1039); // 进程退出
-        }
-        else {
                 std::cout << "param is not executable" << std::endl;
             }
+
+        } else if (fun == "delete") {
+
+            // delete 分支：不要求 frontType/processNum
+            send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到删除进程指令,这个进程将会重启 ");
+
+            DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,即将重启",
+                        g_front_seg_index, fun.c_str());
+
+            std::this_thread::sleep_for(std::chrono::seconds(3));
+            ::_exit(-1039); // 进程退出
+
+        } else {
+            std::cout << "param is not executable" << std::endl;
+        }
+        // ----- MOD END: 分功能按需校验与执行 -----
+
     } else {
         std::cout << "set process code str error" << std::endl;
     }
@@ -927,6 +978,20 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
         return rocketmq::RECONSUME_LATER;
     }
 
+    // [MOD] 仅消费启动后的消息：历史消息直接跳过并 ACK（即使并发也安全）
+    // ----- MOD BEGIN: 启动后消息过滤 -----
+    if (!should_process_after_start(msg)) {
+        std::cout << "[SET] skip old message: "
+                  << "topic=" << msg.getTopic()
+                  << ", queueId=" << msg.getQueueId()
+                  << ", offset=" << msg.getQueueOffset()
+                  << ", bornTs=" << msg.getBornTimestamp()
+                  << ", appStart=" << G_APP_START_MS
+                  << std::endl;
+        return rocketmq::CONSUME_SUCCESS; // 确认成功，避免重投
+    }
+    // ----- MOD END: 启动后消息过滤 -----
+
     std::string body = msg.getBody();
     std::string key = msg.getKeys();
 
@@ -992,6 +1057,20 @@ rocketmq::ConsumeStatus myMessageCallbackupdate(const rocketmq::MQMessageExt& ms
         return rocketmq::RECONSUME_LATER;
     }
 
+    // [MOD] 仅消费启动后的消息：历史消息直接跳过并 ACK（即使并发也安全）
+    // ----- MOD BEGIN: 启动后消息过滤 -----
+    if (!should_process_after_start(msg)) {
+        std::cout << "[SET] skip old message: "
+                  << "topic=" << msg.getTopic()
+                  << ", queueId=" << msg.getQueueId()
+                  << ", offset=" << msg.getQueueOffset()
+                  << ", bornTs=" << msg.getBornTimestamp()
+                  << ", appStart=" << G_APP_START_MS
+                  << std::endl;
+        return rocketmq::CONSUME_SUCCESS; // 确认成功，避免重投
+    }
+    // ----- MOD END: 启动后消息过滤 -----
+
     std::string body = msg.getBody();
     std::string key = msg.getKeys();
 
@@ -1023,6 +1102,20 @@ rocketmq::ConsumeStatus myMessageCallbackset(const rocketmq::MQMessageExt& msg)
         return rocketmq::RECONSUME_LATER;
     }
 
+    // [MOD] 仅消费启动后的消息：历史消息直接跳过并 ACK（即使并发也安全）
+    // ----- MOD BEGIN: 启动后消息过滤 -----
+    if (!should_process_after_start(msg)) {
+        std::cout << "[SET] skip old message: "
+                  << "topic=" << msg.getTopic()
+                  << ", queueId=" << msg.getQueueId()
+                  << ", offset=" << msg.getQueueOffset()
+                  << ", bornTs=" << msg.getBornTimestamp()
+                  << ", appStart=" << G_APP_START_MS
+                  << std::endl;
+        return rocketmq::CONSUME_SUCCESS; // 确认成功，避免重投
+    }
+    // ----- MOD END: 启动后消息过滤 -----
+
     std::string body = msg.getBody();
     std::string key = msg.getKeys();
 
@@ -1041,7 +1134,7 @@ rocketmq::ConsumeStatus myMessageCallbackset(const rocketmq::MQMessageExt& msg)
 
     // 调用业务处理逻辑
     if (!parseJsonMessageSET(body)) {
-        DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的进程控制消息失败,消息的json结构不正确", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_SET.c_str());
+        DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s - tag:%s的进程控制消息失败,消息的json结构不正确", g_front_seg_index, G_MQCONSUMER_TOPIC_SET.c_str(), FRONT_INST.c_str());
     }
 
     return rocketmq::CONSUME_SUCCESS;
@@ -1053,6 +1146,20 @@ rocketmq::ConsumeStatus myMessageCallbacklog(const rocketmq::MQMessageExt& msg)
         return rocketmq::RECONSUME_LATER;
     }
 
+    // [MOD] 仅消费启动后的消息：历史消息直接跳过并 ACK（即使并发也安全）
+    // ----- MOD BEGIN: 启动后消息过滤 -----
+    if (!should_process_after_start(msg)) {
+        std::cout << "[SET] skip old message: "
+                  << "topic=" << msg.getTopic()
+                  << ", queueId=" << msg.getQueueId()
+                  << ", offset=" << msg.getQueueOffset()
+                  << ", bornTs=" << msg.getBornTimestamp()
+                  << ", appStart=" << G_APP_START_MS
+                  << std::endl;
+        return rocketmq::CONSUME_SUCCESS; // 确认成功，避免重投
+    }
+    // ----- MOD END: 启动后消息过滤 -----
+
     std::string body = msg.getBody();
     std::string key = msg.getKeys();
 
@@ -1084,6 +1191,20 @@ rocketmq::ConsumeStatus myMessageCallbackrecall(const rocketmq::MQMessageExt& ms
         return rocketmq::RECONSUME_LATER;
     }
 
+    // [MOD] 仅消费启动后的消息：历史消息直接跳过并 ACK（即使并发也安全）
+    // ----- MOD BEGIN: 启动后消息过滤 -----
+    if (!should_process_after_start(msg)) {
+        std::cout << "[SET] skip old message: "
+                  << "topic=" << msg.getTopic()
+                  << ", queueId=" << msg.getQueueId()
+                  << ", offset=" << msg.getQueueOffset()
+                  << ", bornTs=" << msg.getBornTimestamp()
+                  << ", appStart=" << G_APP_START_MS
+                  << std::endl;
+        return rocketmq::CONSUME_SUCCESS; // 确认成功，避免重投
+    }
+    // ----- MOD END: 启动后消息过滤 -----
+
     // 调试输出
     std::cout << "myMessageCallbackrecall" << std::endl;