fix recall

LFtid1056/cloudfront/code/cfg_parser.cpp

@@ -1288,7 +1288,7 @@ int recall_json_handle_from_mq(const std::string& body)
                     if (ClientManager::instance().get_dev_status(targetDev->terminal_id) != 1) {
                         std::cout << "terminalId对应装置不在线: " << targetDev->terminal_id << std::endl;
                         std::string msg = std::string("装置：") + targetDev->terminal_name + " 不在线，无法补招";
-                        send_reply_to_kafka_recall(guid, dt, static_cast<int>(ResponseCode::INTERNAL_ERROR), msg, targetDev->terminal_id, "", "", "");
+                        send_reply_to_kafka_recall(guid, dt, static_cast<int>(ResponseCode::NOT_FOUND), msg, targetDev->terminal_id, "", "", "");
                         continue;
                     }
                 }
@@ -1363,6 +1363,7 @@ int recall_json_handle_from_mq(const std::string& body)
                                 for (size_t i = 0; i < recallinfo_list_hour.size(); ++i) {
                                     const RecallInfo& info = recallinfo_list_hour[i];
                                     RecallMonitor rm;
+                                    rm.recall_guid   = guid;
                                     rm.recall_status = 0;
                                     rm.StartTime     = epoch_to_datetime_str(info.starttime);
                                     rm.EndTime       = epoch_to_datetime_str(info.endtime);
@@ -1379,12 +1380,14 @@ int recall_json_handle_from_mq(const std::string& body)
                     }
                 } else {
                     // 未知 dataType，忽略
+                    std::cout << "unknown dataType: " << dt << std::endl;
                     continue;
                 }
             }
         }
         else {
             // 不支持的 messageBody 形态
+            std::cout << "unknown messageBody form" << std::endl;
             return 10004;
         }
     }
@@ -3406,45 +3409,76 @@ bool send_file_list(terminal_dev* dev, const std::vector<tag_dir_info>& FileList
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////检查云前置终端的mq业务超时
 std::string get_type_by_state(int state) {
-    switch (static_cast<DeviceState>(state)) {
-        case DeviceState::READING_STATS:
-            return "读取统计数据";  //读数据
-        case DeviceState::READING_STATS_TIME:
+    switch (state) {
+        case static_cast<int>(DeviceState::IDLE):
+            return "空闲状态";
+
+        case static_cast<int>(DeviceState::READING_STATS):
+            return "读取统计数据";
+
+        case static_cast<int>(DeviceState::READING_STATS_TIME):
             return "读取统计时间";
-        case DeviceState::READING_REALSTAT:
+
+        case static_cast<int>(DeviceState::READING_REALSTAT):
             return "读取实时数据";
-        case DeviceState::READING_FIXEDVALUE:
-            return "读取定值";
-        case DeviceState::READING_FIXEDVALUEDES:
-            return "读取定值描述";
-        case DeviceState::SET_FIXEDVALUE:
-            return "设置定值";
-        case DeviceState::READING_INTERFIXEDVALUE:
-            return "读取内部定值";
-        case DeviceState::READING_INTERFIXEDVALUEDES:
-            return "读取内部定值描述";
-        case DeviceState::READING_CONTROLWORD:
-            return "读取控制字";
-        case DeviceState::SET_INTERFIXEDVALUE:
-            return "设置内部定值";
-            //return 0x2106;  //读数据
 
-        case DeviceState::READING_FILEMENU:
+        case static_cast<int>(DeviceState::READING_EVENTFILE):
+            return "暂态波形文件下载";
+
+        case static_cast<int>(DeviceState::READING_FILEMENU):
             return "读取文件目录";
-            //return 0x2131;  //读目录
 
-        case DeviceState::READING_EVENTFILE:
-            return "读取事件文件目录";
-            //return 0x2133;  //读事件文件目录
-        case DeviceState::READING_FILEDATA:
+        case static_cast<int>(DeviceState::READING_FILEDATA):
             return "读取文件数据";
-            //return 0x2132;  //读文件
+
+        case static_cast<int>(DeviceState::READING_FIXEDVALUE):
+            return "读取测点定值";
+
+        case static_cast<int>(DeviceState::READING_FIXEDVALUEDES):
+            return "读取测点定值描述";
+
+        case static_cast<int>(DeviceState::SET_FIXEDVALUE):
+            return "设置测点定值";
+
+        case static_cast<int>(DeviceState::READING_INTERFIXEDVALUE):
+            return "读取内部定值";
+
+        case static_cast<int>(DeviceState::READING_INTERFIXEDVALUEDES):
+            return "读取内部定值描述";
+
+        case static_cast<int>(DeviceState::READING_CONTROLWORD):
+            return "读取控制字描述";
+
+        case static_cast<int>(DeviceState::SET_INTERFIXEDVALUE):
+            return "设置内部定值";
+
+        case static_cast<int>(DeviceState::READING_RUNNINGINFORMATION_1):
+            return "读取装置运行信息(主动触发)";
+
+        case static_cast<int>(DeviceState::READING_RUNNINGINFORMATION_2):
+            return "读取装置运行信息(定时执行)";
+
+        case static_cast<int>(DeviceState::READING_DEVVERSION):
+            return "读取装置版本配置信息";
+
+        case static_cast<int>(DeviceState::SET_RIGHTTIME):
+            return "设置装置对时";
+
+        case static_cast<int>(DeviceState::READING_EVENTLOG):
+            return "补招事件日志";
+
+        case static_cast<int>(DeviceState::READING_STATSFILE):
+            return "补招稳态数据文件";
+
+        case static_cast<int>(DeviceState::CUSTOM_ACTION):
+            return "自定义动作";
 
         default:
-            return "未知业务"; // 没有对应的type
+            return "未知业务"; // 未匹配的类型
     }
 }
 
+
 // 定时检查业务超时
 void check_device_busy_timeout()
 {
@@ -3458,7 +3492,7 @@ void check_device_busy_timeout()
 
             if (dev.busytype == static_cast<int>(DeviceState::READING_FILEDATA))    //下载文件业务
             {
-                if (dev.busytimecount > 30)
+                if (dev.busytimecount > 60)
                 {
                     std::cout << "[Timeout] Device " << dev.terminal_id
                               << " busytype=READING_FILEDATA 超时(" 
@@ -3478,7 +3512,7 @@ void check_device_busy_timeout()
             }
             else    //其他业务
             {
-                if (dev.busytimecount > 10)
+                if (dev.busytimecount > 20)
                 {
                     std::cout << "[Timeout] Device " << dev.terminal_id
                               << " busytype=" << dev.busytype 
@@ -4064,6 +4098,7 @@ void check_recall_event() {
         for (auto& dev : terminal_devlist) {
             //如果该终端不是正在补招或者idle则直接跳过，节省运行时间
             if(dev.busytype != static_cast<int>(DeviceState::READING_EVENTLOG) && dev.busytype != static_cast<int>(DeviceState::IDLE)){
+                std::cout << "[check_recall_event] skip dev=" << dev.terminal_id << std::endl;
                 continue;
             }
             // 对正在补招或idle终端的所有监测点的待补招列表进行处理
@@ -4085,7 +4120,9 @@ void check_recall_event() {
                         send_reply_to_kafka_recall(dev.guid,1,static_cast<int>(ResponseCode::OK),msg,dev.terminal_id,lm.monitor_id,front.StartTime,front.EndTime);
 
                         lm.recall_list.pop_front(); // 弹掉首条
-                    }else if (front.recall_status == static_cast<int>(RecallStatus::EMPTY)) {
+                        break;
+                    }
+                    else if (front.recall_status == static_cast<int>(RecallStatus::EMPTY)) {
                         std::cout << "[check_recall_event] EMPTY dev=" << dev.terminal_id
                                   << " monitor=" << lm.monitor_id
                                   << " " << front.StartTime << " ~ " << front.EndTime << std::endl;
@@ -4098,6 +4135,7 @@ void check_recall_event() {
                         send_reply_to_kafka_recall(dev.guid,1,static_cast<int>(ResponseCode::NOT_FOUND),msg,dev.terminal_id,lm.monitor_id,front.StartTime,front.EndTime);
 
                         lm.recall_list.pop_front(); // 弹掉首条
+                        break;
                     }  
                     else if (front.recall_status == static_cast<int>(RecallStatus::FAILED)) {
                         std::cout << "[check_recall_event] FAILED dev=" << dev.terminal_id
@@ -4112,31 +4150,16 @@ void check_recall_event() {
                         send_reply_to_kafka_recall(dev.guid,1,static_cast<int>(ResponseCode::BAD_REQUEST),msg,dev.terminal_id,lm.monitor_id,front.StartTime,front.EndTime);
 
                         lm.recall_list.pop_front(); // 弹掉首条
+                        break;
                     } else {
-                        break; // 首条不是 2/3，停，如果是正在处理其他业务或者idle的装置写入了待补招列表，应该都是0；如果是正在补招的装置，新增的部分不会影响原有顺序
+                        std::cout << "[check_recall_event] skip line=" << lm.monitor_name<< std::endl;
+                        break; // 首条不是 2/3/4，停，如果是正在处理其他业务或者idle的装置写入了待补招列表，应该都是0；如果是正在补招的装置，新增的部分不会影响原有顺序
                     }
                 }
                 if (!lm.recall_list.empty()) any_non_empty = true;  // 处理了成功和失败的以后只要有一条非空就标记，可能是待处理或者正在处理的补招
             }
 
-            if (!any_non_empty && dev.busytype == static_cast<int>(DeviceState::READING_EVENTLOG)) {
-                // 该终端本轮已无任何补招条目，且处于补招暂态事件的状态清空运行态
-                //通知补招全部完成
-
-                dev.guid.clear();       // 清空 guid
-                dev.busytype = 0;       // 业务类型归零
-                dev.isbusy = 0;         // 清空业务标志
-                dev.busytimecount = 0;  // 计时归零
-                continue;
-            }
-            //如果没有待补招任务,或者正在进行其他业务，应该跳过
-            else if(!any_non_empty || (dev.busytype != static_cast<int>(DeviceState::IDLE) && dev.busytype != static_cast<int>(DeviceState::READING_EVENTLOG))){
-                continue;
-            }
-
-            //有待补招任务且处于补招事件状态或者idle状态，继续补招处理
-
-            // 2) 若该装置任一 monitor 的首条为 RUNNING，则该终端正在补招中 -> 跳过该终端,不会下发新的补招请求
+            // pop后判断是否仍有 RUNNING（pop后应该都为unstarted，没pop的才会是running）
             bool has_running = false;
             for (auto& lm : dev.line) {
                 if (!lm.recall_list.empty() &&
@@ -4145,20 +4168,49 @@ void check_recall_event() {
                     break;
                 }
             }
-            if (has_running) continue;//跳过这个装置
+
+            // 有条目但目前存在 RUNNING：继续等待该 RUNNING 完成，本轮不新发
+            if (any_non_empty && has_running) {
+                std::cout << "[check_recall_event] skip dev=" << dev.terminal_id
+                          << " already running recall" << std::endl;
+                continue;
+            }
+
+            //pop后无任何补招条目，且处于补招状态：清空运行态
+            if (!any_non_empty && dev.busytype == static_cast<int>(DeviceState::READING_EVENTLOG)) {
+                // 该终端本轮已无任何补招条目，且处于补招暂态事件的状态清空运行态
+                std::cout << "[check_recall_event] finish recall dev=" << dev.terminal_id << std::endl;
+                //通知补招全部完成
+
+                dev.guid.clear();       // 清空 guid
+                dev.busytype = static_cast<int>(DeviceState::IDLE);       // 业务类型归零
+                dev.isbusy = 0;         // 清空业务标志
+                dev.busytimecount = 0;  // 计时归零
+                continue;   //处理完处理下一个装置
+            }
+
+            
+
+            //有待补招任务，且idle或者在补招继续补招处理
+            std::cout << "[check_recall_event] idle or continue recall dev=" << dev.terminal_id << std::endl;
 
             // 若无 RUNNING，则说明该终端空闲，可以挑选新的补招任务
+            if (any_non_empty && !has_running) {
 
                 // 3) 选择该终端的“第一条 NOT_STARTED(0)”作为本终端本轮任务
                 bool picked = false;
                 for (auto& lm : dev.line) {
-                if (lm.recall_list.empty()) continue;   //跳过空的监测点
+                    if (lm.recall_list.empty()) {
+                        std::cout << "[check_recall_event] skip empty line=" << lm.monitor_name<< std::endl;
+                        continue;   //跳过空的监测点
+                    }
 
                     RecallMonitor& front = lm.recall_list.front();  //取非空测点的列表的第一条
                     if (front.recall_status == static_cast<int>(RecallStatus::NOT_STARTED)) {//未补招
                         // 标记为 RUNNING，并设置终端忙状态
                         front.recall_status = static_cast<int>(RecallStatus::RUNNING);//该补招记录刷新为补招中
 
+                        dev.guid          = front.recall_guid; // 记录本次补招的 guid
                         dev.isbusy        = 1;      //标记为忙
                         dev.busytype      = static_cast<int>(DeviceState::READING_EVENTLOG);//装置状态正在补招和idle的都刷新为正在补招
                         dev.busytimecount = 0;      //刷新业务超时计数
@@ -4175,7 +4227,15 @@ void check_recall_event() {
                     }
                 }
                 // 若该终端没有 NOT_STARTED 的首条（可能剩下的都是 RUNNING 或内部已经被清空），
+                 if (!picked) {
+                    // 没挑到 NOT_STARTED（例如都是“首条非空但非 NOT_STARTED/非 RUNNING”的情况）：
+                    // 这种情况下本终端留给下一轮处理即可。
+                    std::cout << "[check_recall_event] skip dev=" << dev.terminal_id
+                              << " no NOT_STARTED at head" << std::endl;
+                }
                 // 就留待下一轮；不要清空运行态，直到所有条目被处理为止。
+                continue;
+            }
         }
     } // 解锁
 

LFtid1056/cloudfront/code/front.h

@@ -94,6 +94,11 @@ public:
     std::atomic<bool> m_bIsFrontThreadCancle{false};
     std::atomic<bool> m_IsMQConsumerCancel{false};
 
+    std::atomic<bool> m_frontRunning{false};
+    std::atomic<bool> m_consumerRunning{false};
+    std::atomic<bool> m_producerRunning{false};
+    std::atomic<bool> m_timerRunning{false};
+
     std::mutex m_threadCheckMutex;
     std::atomic<bool> m_needRestartFrontThread{false};
     std::atomic<bool> m_needRestartConsumerThread{false};
@@ -110,6 +115,12 @@ public:
     void StartMQProducerThread();
     void StartTimerThread();
 
+    // [ADD] 统一的停止接口（便于重启前先停干净）
+    void StopFrontThread();
+    void StopMQConsumerThread();
+    void StopMQProducerThread();
+    void StopTimerThread();
+
     void FrontThread();
     void mqconsumerThread();
     void mqproducerThread();

LFtid1056/cloudfront/code/interface.h

@@ -44,6 +44,7 @@ public:
 class RecallMonitor
 {
 public:
+    std::string recall_guid; // 本次补招的唯一标识 GUID
     int recall_status;      //补招状态 0-未补招 1-补招中 2-补招完成 3-补招失败 4-无数据
     std::string StartTime; //数据补招起始时间
 	std::string EndTime;   //数据补招结束时间

LFtid1056/cloudfront/code/main.cpp

@@ -246,7 +246,7 @@ std::string get_parent_directory() {
     }
 
     // ============ 关闭所有运行中的线程============ 
-    void Front::FormClosing() {
+    /*void Front::FormClosing() {
         //确保testshell关闭
         m_worker.stopServer();
     
@@ -286,11 +286,19 @@ std::string get_parent_directory() {
             m_MQConsumerThread.join();
         }
    
+    }*/
+    void Front::FormClosing() {
+        m_worker.stopServer();
+
+        StopFrontThread();
+        StopTimerThread();
+        StopMQProducerThread();
+        StopMQConsumerThread();
     }
 
     //============ 线程函数 ============
 
-    void Front::StartFrontThread() {
+    /*void Front::StartFrontThread() {
         m_bIsFrontThreadCancle = false;
         m_FrontThread = std::thread(&Front::FrontThread, this);
     }
@@ -308,6 +316,136 @@ std::string get_parent_directory() {
     void Front::StartTimerThread() {
         m_IsTimerCancel = false;
         m_TimerThread = std::thread(&Front::OnTimerThread, this);
+    }  */
+   void Front::StartFrontThread() {
+        bool expected = false;
+        if (!m_frontRunning.compare_exchange_strong(expected, true)) {
+            std::cout << "[FrontThread] already running, skip\n";
+            return;
+        }
+        if (m_FrontThread.joinable()) m_FrontThread.join();
+
+        m_bIsFrontThreadCancle = false;
+        m_FrontThread = std::thread([this]{
+            try {
+                this->FrontThread();
+            } catch (const std::exception& e) {
+                std::cerr << "[FrontThread] exception: " << e.what() << "\n";
+            } catch (...) {
+                std::cerr << "[FrontThread] unknown exception\n";
+            }
+            m_frontRunning = false; // 线程真正退出后复位
+        });
+    }
+
+    void Front::StartMQConsumerThread() {
+        bool expected = false;
+        if (!m_consumerRunning.compare_exchange_strong(expected, true)) {
+            std::cout << "[MQConsumer] already running, skip\n";
+            return;
+        }
+        if (m_MQConsumerThread.joinable()) m_MQConsumerThread.join();
+
+        m_IsMQConsumerCancel = false;
+        m_MQConsumerThread = std::thread([this]{
+            try {
+                this->mqconsumerThread();
+            } catch (const std::exception& e) {
+                std::cerr << "[mqconsumerThread] exception: " << e.what() << "\n";
+            } catch (...) {
+                std::cerr << "[mqconsumerThread] unknown exception\n";
+            }
+            m_consumerRunning = false;
+        });
+    }
+
+    void Front::StartMQProducerThread() {
+        bool expected = false;
+        if (!m_producerRunning.compare_exchange_strong(expected, true)) {
+            std::cout << "[MQProducer] already running, skip\n";
+            return;
+        }
+        if (m_MQProducerThread.joinable()) m_MQProducerThread.join();
+
+        m_IsMQProducerCancel = false;
+        m_MQProducerThread = std::thread([this]{
+            try {
+                this->mqproducerThread();
+            } catch (const std::exception& e) {
+                std::cerr << "[mqproducerThread] exception: " << e.what() << "\n";
+            } catch (...) {
+                std::cerr << "[mqproducerThread] unknown exception\n";
+            }
+            m_producerRunning = false;
+        });
+    }
+
+    void Front::StartTimerThread() {
+        bool expected = false;
+        if (!m_timerRunning.compare_exchange_strong(expected, true)) {
+            std::cout << "[Timer] already running, skip StartTimerThread\n";
+            return; // 已有定时线程在跑，直接跳过
+        }
+
+        // 若有旧线程尚未 join，先回收
+        if (m_TimerThread.joinable()) {
+            m_TimerThread.join();
+        }
+
+        m_IsTimerCancel.store(false, std::memory_order_relaxed);
+
+        m_TimerThread = std::thread([this]{
+            try {
+                this->OnTimerThread();
+            } catch (const std::exception& e) {
+                std::cerr << "[Timer] exception: " << e.what() << "\n";
+            } catch (...) {
+                std::cerr << "[Timer] unknown exception\n";
+            }
+            m_timerRunning.store(false); // 线程真正退出后复位
+        });
+    }
+
+    void Front::StopFrontThread() {
+        if (!m_frontRunning.load()) return;
+        m_bIsFrontThreadCancle = true;
+        if (m_FrontThread.joinable()) m_FrontThread.join();
+        m_frontRunning = false;
+    }
+
+    void Front::StopMQConsumerThread() {
+        if (!m_consumerRunning.load()) return;
+
+        m_IsMQConsumerCancel = true;          // 你的线程函数可能不轮询此标志，但先置上
+        // 关闭 MQ 对象（避免内部阻塞线程仍在）
+        if (m_mqConsumer) {
+            try { m_mqConsumer->shutdown(); } catch (...) {}
+            m_mqConsumer.reset();
+        }
+        m_listener.reset();
+
+        if (m_MQConsumerThread.joinable()) m_MQConsumerThread.join();
+        m_consumerRunning = false;
+    }
+
+    void Front::StopMQProducerThread() {
+        if (!m_producerRunning.load()) return;
+
+        m_IsMQProducerCancel = true;
+        if (m_MQProducerThread.joinable()) m_MQProducerThread.join();
+        m_producerRunning = false;
+
+        // 如需销毁/关闭 producer，对应你的初始化方式：
+        // if (m_producer) { ShutdownProducer(m_producer); m_producer = nullptr; }
+    }
+
+    void Front::StopTimerThread() {
+        if (!m_timerRunning.load()) return;  // 没跑就不处理
+        m_IsTimerCancel.store(true);
+        if (m_TimerThread.joinable()) {
+            m_TimerThread.join();            // 等它退出
+        }
+        m_timerRunning.store(false);
     }
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////主功能线程
@@ -596,7 +734,7 @@ void* cloudfrontthread(void* arg) {
         {
             std::lock_guard<std::mutex> lock(FrontProcess->m_threadCheckMutex);
 
-            if (FrontProcess->m_needRestartFrontThread) {
+            /*if (FrontProcess->m_needRestartFrontThread) {
                 std::cout << "[Monitor] Restarting FrontThread..." << std::endl;
                 FrontProcess->StartFrontThread();
                 FrontProcess->m_needRestartFrontThread = false;
@@ -618,6 +756,34 @@ void* cloudfrontthread(void* arg) {
                 std::cout << "[Monitor] Restarting TimerThread..." << std::endl;
                 FrontProcess->StartTimerThread();
                 FrontProcess->m_needRestartTimerThread = false;
+            }*/
+
+            if (FrontProcess->m_needRestartFrontThread) {
+                std::cout << "[Monitor] Restarting FrontThread..." << std::endl;
+                FrontProcess->StopFrontThread();
+                FrontProcess->StartFrontThread();
+                FrontProcess->m_needRestartFrontThread = false;
+            }
+            
+            if (FrontProcess->m_needRestartConsumerThread) {
+                std::cout << "[Monitor] Restarting MQConsumerThread..." << std::endl;
+                FrontProcess->StopMQConsumerThread();
+                FrontProcess->StartMQConsumerThread();
+                FrontProcess->m_needRestartConsumerThread = false;
+            }
+            
+            if (FrontProcess->m_needRestartProducerThread) {
+                std::cout << "[Monitor] Restarting MQProducerThread..." << std::endl;
+                FrontProcess->StopMQProducerThread();
+                FrontProcess->StartMQProducerThread();
+                FrontProcess->m_needRestartProducerThread = false;
+            }
+
+            if (FrontProcess->m_needRestartTimerThread) {
+                std::cout << "[Monitor] Restarting TimerThread..." << std::endl;
+                FrontProcess->StopTimerThread();   // 先停
+                FrontProcess->StartTimerThread();  // 再启
+                FrontProcess->m_needRestartTimerThread = false;
             }
         }
 

LFtid1056/cloudfront/code/worker.cpp

@@ -425,6 +425,12 @@ void Worker::printLedgerinshell(const terminal_dev& dev, int fd) {
 
     std::ostringstream os;
     os << "\r\x1B[K------------------------------------\n";
+
+    os << "\r\x1B[K|-- guid             : " << dev.guid          << "\n";
+    os << "\r\x1B[K|-- busytype         : " << dev.busytype      << "\n";
+    os << "\r\x1B[K|-- isbusy           : " << dev.isbusy        << "\n";
+    os << "\r\x1B[K|-- busytimecount    : " << dev.busytimecount << "\n";
+
     //os << "\r\x1B[K|-- dev_index        : " << dev.dev_index   << "\n";
     os << "\r\x1B[K|-- terminal_id      : " << dev.terminal_id   << "\n";
     os << "\r\x1B[K|-- terminal_name    : " << dev.terminal_name << "\n";