the heartbeat reply ledgerupdate is ok

cfg_parse/SimpleProducer.cpp

@@ -44,6 +44,8 @@ extern std::string G_ROCKETMQ_TOPIC;//topie
 extern std::string G_ROCKETMQ_TAG;//tag
 extern std::string G_ROCKETMQ_KEY;//key
 
+extern std::string FRONT_INST;
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -283,14 +285,14 @@ RocketMQConsumer::~RocketMQConsumer()
 // 在 RocketMQConsumer 类中新增函数用来设置消费模式
 void RocketMQConsumer::setConsumerMessageModel(const std::string& topic)
 {
-    if (topic == G_MQCONSUMER_TOPIC_SET) {
+    /*if (topic == G_MQCONSUMER_TOPIC_SET) {
         // 设置为普通消费模式
         if (SetPushConsumerMessageModel(consumer_, CLUSTERING) != 0) {
             std::cout << "Error setting message model to CLUSTERING for topic: " << topic << std::endl;
         } else {
             std::cout << "Set consumer to CLUSTERING for topic: " << topic << std::endl;
         }
-    } else {
+    } else*/ {
         // 默认设置为广播消费模式
         if (SetPushConsumerMessageModel(consumer_, BROADCASTING) != 0) {
             std::cout << "Error setting message model to BROADCASTING for topic: " << topic << std::endl;
@@ -649,7 +651,7 @@ void rocketmq_test_rt()
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_RT);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_RT);
     std::ifstream file("rt.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容
@@ -663,7 +665,7 @@ void rocketmq_test_ud()//用来测试台账更新
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_UD);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_UD);
     std::ifstream file("ud.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容
@@ -677,7 +679,7 @@ void rocketmq_test_set()//用来测试进程控制脚本
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_SET);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_SET);
     std::ifstream file("set.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容
@@ -691,7 +693,7 @@ void rocketmq_test_only()//用来测试进程控制脚本
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_SET);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_SET);
     std::ifstream file("set_debug.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容
@@ -706,7 +708,7 @@ void rocketmq_test_rc()
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_RC);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_RC);
     std::ifstream file("rc.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容
@@ -721,7 +723,7 @@ void rocketmq_test_log()
 {
     Ckafka_data_t data;
     data.monitor_id = 123123;
-    data.strTopic = QString::fromStdString(G_MQCONSUMER_TOPIC_LOG);
+    data.strTopic = QString::fromStdString(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_LOG);
     std::ifstream file("log_test.txt");  // 文件中存储长字符串
     std::stringstream buffer;
     buffer << file.rdbuf();  // 读取整个文件内容

cfg_parse/cfg_parser.cpp

@@ -221,7 +221,7 @@ char* UDS_DELETE_URL;
 
 int  FILE_FLAG;
 int  SEND_FLAG;
-int FRONT_INST;
+std::string FRONT_INST;//lnk20250512改为string
 char* FRONT_IP;
 int CITY_FLAG;
 
@@ -331,10 +331,16 @@ extern pthread_mutex_t mtx;
 
 /*lnk 2024-10-21 */
 std::string intToString(int number);
+
+//lnk20250512
+void send_heartbeat_to_kafka(const std::string& status);
 //////////////////////////////////////////////////////////////////////////
 
 extern int server_socket; //Web Socket服务端实例
 extern unsigned int	g_node_id; //前置程序类型(100-500)
+
+extern QMutex kafka_data_list_mutex; //Kafka发送数据锁
+extern QList<Ckafka_data_t> kafka_data_list; //kafka发送数据链表
 //WW 2023-08-20 end
 ////////////////////////////////////////////////////////////////////////////////////////////////////
 
@@ -401,8 +407,13 @@ void init_config() {
 	qDebug() << "Read SEND_FLAG:" << SEND_FLAG << endl;
 	FILE_FLAG = settings.value("Flag/FileFlag", 0).toInt();
 	qDebug() << "Read FILE_FLAG:" << FILE_FLAG << endl;
-	FRONT_INST = settings.value("Flag/FrontInst", 0).toInt();
+
-	qDebug() << "Read FRONT_INST:" << FRONT_INST << endl;
+	//FRONT_INST = settings.value("Flag/FrontInst", 0).toInt();
+	//qDebug() << "Read FRONT_INST:" << FRONT_INST << endl;
+	ba = settings.value("Flag/FrontInst", "").toString().toLatin1();
+	FRONT_INST = strdup(ba.data());
+	std::cout << "Read FRONT_INST:" << FRONT_INST << endl;
+
 	ba = settings.value("Flag/FrontIP", "").toString().toLatin1();
 	FRONT_IP = strdup(ba.data());
 	qDebug() << "Read FRONT_IP:" << FRONT_IP << endl;
@@ -686,6 +697,12 @@ void init_config() {
 	std::cout << "Read G_CONNECT_TOPIC:" << G_CONNECT_TOPIC << std::endl;
 	std::cout << "Read G_CONNECT_TAG:" << G_CONNECT_TAG << std::endl;
 	std::cout << "Read G_CONNECT_KEY:" << G_CONNECT_KEY << std::endl;
+	std::cout << "Read Heart_Beat_Topic:" << Heart_Beat_Topic << std::endl;
+	std::cout << "Read Heart_Beat_Tag:" << Heart_Beat_Tag << std::endl;
+	std::cout << "Read Heart_Beat_Key:" << Heart_Beat_Key << std::endl;
+	std::cout << "Read Topic_Reply_Topic:" << Topic_Reply_Topic << std::endl;
+	std::cout << "Read Topic_Reply_Tag:" << Topic_Reply_Tag << std::endl;
+	std::cout << "Read Topic_Reply_Key:" << Topic_Reply_Key << std::endl;
 //消费者相关打印
 	std::cout << "Read G_ROCKETMQ_CONSUMER:" << G_ROCKETMQ_CONSUMER << std::endl;
 	std::cout << "Read G_MQCONSUMER_IPPORT:" << G_MQCONSUMER_IPPORT << std::endl;
@@ -3525,28 +3542,6 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
         return 1;
     }
 
-	//后续修改为根据index来获取对应的台账，将不再均分，获取所有台账并判断进程号
-#if 0
-	int base_size = 0;
-	int remainder = 0;
-	int start_index = 0;
-	int end_index = data_size;
-
-	if(1 == MULTIPLE_NODE_FLAG){
-		// 计算每份的大小
-    	base_size = data_size / num;
-    	remainder = data_size % num;
-
-    	// 计算当前份的起始和结束索引
-    	start_index = index * base_size + (index < remainder ? index : remainder);
-    	end_index = start_index + base_size + (index < remainder ? 1 : 0);
-
-    	// 确保结束索引不超过数组大小
-    	if (end_index > data_size) {
-    	    end_index = data_size;
-    	}
-	}
-#endif
 	int start_index = 0;
 	int end_index = data_size;
 
@@ -4287,7 +4282,7 @@ int parse_model_cfg_web()
 	}
 }
 ////////////////////////////////////////////////////////////icd模型重构函数lnk20250116
-char* parse_model_cfg_web_one(ied_t* ied)
+char* parse_model_cfg_web_one(ied_t* ied, char* out_model)
 {
 	std::vector<std::string> codes;//入参集合
 	QMap<QString, icd_model*>  icd_model_map;
@@ -4323,19 +4318,28 @@ char* parse_model_cfg_web_one(ied_t* ied)
     		if (value != nullptr) {
         		// 找到容器中对应名称并取值
         		strncpy(model_id, value->model_id, sizeof(model_id) - 1);
+				model_id[sizeof(model_id) - 1] = '\0';
+
         		strncpy(tmnl_type, value->tmnl_type, sizeof(tmnl_type) - 1);
         		strncpy(file_path, value->file_path, sizeof(file_path) - 1);
         		strncpy(file_name, value->file_name, sizeof(file_name) - 1);
 
-				std::cout << "model_id" << model_id << std::endl;
+				std::cout << "model_id:" << model_id << std::endl;
-				std::cout << "tmnl_type" << tmnl_type << std::endl;
+				std::cout << "tmnl_type:" << tmnl_type << std::endl;
-				std::cout << "filepath" << file_path << std::endl;
+				std::cout << "filepath:" << file_path << std::endl;
-				std::cout << "filename" << file_name << std::endl;
+				std::cout << "filename:" << file_name << std::endl;
 
 				Set_xml_databaseinfo(model_id, tmnl_type, file_path, file_name, timestamp.year, timestamp.month, timestamp.day, timestamp.hour, timestamp.minute, timestamp.second);
-    		}
+    		
+				if (out_model != NULL) {
+					strncpy(out_model, model_id, 64);
+					out_model[63] = '\0';
+				}
+
+				return model_id;
+			}
 		}
-		return model_id;
+		
 	}
 	catch (otl_exception& e)
 	{
@@ -4377,8 +4381,11 @@ void OnTimerThread::run()
 	int pgmin = 0;
 	int mp_num_hour = 0;
 
-	//添加入参容器
+	//时间计数用
-	std::vector<std::string> codes;
+	int counter = 0;
+
+	//初始化发送一次心跳
+	send_heartbeat_to_kafka("1");
 
 	while (1)
 	{
@@ -4408,6 +4415,13 @@ void OnTimerThread::run()
 		//添加日志开关控制lnk20250508
 		update_log_entries_countdown();
 
+		//添加进程心跳
+		if (counter >= 30) {
+        	send_heartbeat_to_kafka("1");  // 每30秒发送一次心跳
+        	counter = 0;
+    	}
+		counter++;
+
 		msleep(1000);
 	}
 	printf(">>>OnTimerThread::run() is end!!!\n");
@@ -4611,37 +4625,7 @@ bool threadmsghttp(int fun);
 #ifdef __cplusplus
 }
 #endif
-#if 0
-void RecallJsonResponse(int result) {
-    char* ptr=NULL;
-	
-	// 创建 JSON 对象
-    cJSON* json_response = cJSON_CreateObject();
 
-    // 添加字段
-	if(result == 000000){
-		cJSON_AddStringToObject(json_response, "code", "A0000");
-		cJSON_AddStringToObject(json_response, "msg", "数据补招执行成功");
-	}
-	else{
-		cJSON_AddStringToObject(json_response, "code", "A0002");
-		cJSON_AddStringToObject(json_response, "msg", "数据补招执行失败");
-	}
-    
-    cJSON_AddNullToObject(json_response, "data"); // 设置为 null
-
-    // 将 JSON 对象转换为字符串
-    char* json_string = cJSON_Print(json_response);
-    std::cout << json_string << std::endl;
-
-	//发送回复给对方，只发一次
-	SendJsonAPI_web("调用方", "", json_string ,&ptr);
-
-    // 清理
-    cJSON_Delete(json_response);
-    free(json_string); // 释放打印字符串的内存
-}
-#endif
 
 void WebhttpThread::run()
 {

cfg_parse/log4.cpp

@@ -166,6 +166,9 @@ protected:
 
 public:
     SendAppender() {}
+    virtual ~SendAppender() {
+        destructorImpl();  // 重要！释放 log4cplus 基类资源
+    }
 };
 
 //用来控制日志上送的结构
@@ -452,6 +455,10 @@ extern "C" {
     void log_info (const char* key, const char* msg) { log4_log_with_level(key, msg, 1); }
     void log_warn (const char* key, const char* msg) { log4_log_with_level(key, msg, 2); }
     void log_error(const char* key, const char* msg) { log4_log_with_level(key, msg, 3); }
+
+    void send_reply_to_kafka_c(const char* guid, const char* step, const char* result) {
+        send_reply_to_kafka(std::string(guid), std::string(step), std::string(result));
+    }
     
 }
 

json/save2json.cpp

@@ -73,6 +73,8 @@ extern "C" {
 
 extern int INITFLAG;
 
+extern std::string FRONT_INST;
+
 extern QMutex kafka_data_list_mutex;
 extern QList<Ckafka_data_t> kafka_data_list;
 
@@ -525,87 +527,6 @@ void KafkaSendThread::run()
 			printf("END   my_kafka_send no.%i -------->>>>>>>>>>>> %s \n\n", count++,
 				QDateTime::currentDateTime().toString("yyyy-MM-dd hh:mm:ss.zzz").toAscii().data());
 		}
-		
-		/*//lnk20250225添加日志上送
-		Ckafka_data_t log_send;
-		log_send.strTopic = QString::fromStdString(G_LOG_TOPIC);
-		bool log_gotten;
-		log_gotten = false;
-
-		if(!showinshellflag){
-			if (debugOutputEnabled) {
-				// 如果 normalOutputEnabled 和 warnOutputEnabled 都为 0，且 errorOutputEnabled 为 1，取 errorList 输出
-				// 处理 errorList 的输出
-				pthread_mutex_lock(&debugListMutex);
-				if (!debugList.empty()) {
-					log_gotten = true;
-					log_send.strText = QString::fromStdString(debugList.front());//请确保list正确
-					// 检查是否为空字符串（去掉空白字符后是否为空）
-					if (log_send.strText.trimmed().isEmpty()) {
-						debugList.pop_front();  // 直接丢弃这条日志
-						log_gotten = false;     // 标记没有获取到有效日志
-					} else {
-						debugList.pop_front();  // 只有非空白字符串才会真正返回
-					}
-				}
-				pthread_mutex_unlock(&debugListMutex);
-			}
-			else if (normalOutputEnabled) {
-				// 如果 normalOutputEnabled 为 1，优先从 normalList 获取输出
-				// 处理 normalList 的输出
-				pthread_mutex_lock(&normalListMutex);
-				if (!normalList.empty()) {
-
-					log_gotten = true;
-					log_send.strText = QString::fromStdString(normalList.front());
-					// 检查是否为空字符串（去掉空白字符后是否为空）
-					if (log_send.strText.trimmed().isEmpty()) {
-						normalList.pop_front();  // 直接丢弃这条日志
-						log_gotten = false;     // 标记没有获取到有效日志
-					} else {
-						normalList.pop_front();  // 只有非空白字符串才会真正返回
-					}
-				}
-				pthread_mutex_unlock(&normalListMutex);
-			} else if (warnOutputEnabled) {
-				// 如果 normalOutputEnabled 为 0，且 warnOutputEnabled 为 1，优先从 warnList 获取输出
-				// 处理 warnList 的输出
-				pthread_mutex_lock(&warnListMutex);
-				if (!warnList.empty()) {
-					log_gotten = true;
-					log_send.strText = QString::fromStdString(warnList.front());
-					// 检查是否为空字符串（去掉空白字符后是否为空）
-					if (log_send.strText.trimmed().isEmpty()) {
-						warnList.pop_front();  // 直接丢弃这条日志
-						log_gotten = false;     // 标记没有获取到有效日志
-					} else {
-						warnList.pop_front();  // 只有非空白字符串才会真正返回
-					}
-				}
-				pthread_mutex_unlock(&warnListMutex);
-			} else if (errorOutputEnabled) {
-				// 如果 normalOutputEnabled 和 warnOutputEnabled 都为 0，且 errorOutputEnabled 为 1，取 errorList 输出
-				// 处理 errorList 的输出
-				pthread_mutex_lock(&errorListMutex);
-				if (!errorList.empty()) {
-					log_gotten = true;
-					log_send.strText = QString::fromStdString(errorList.front());
-					// 检查是否为空字符串（去掉空白字符后是否为空）
-					if (log_send.strText.trimmed().isEmpty()) {
-						errorList.pop_front();  // 直接丢弃这条日志
-						log_gotten = false;     // 标记没有获取到有效日志
-					} else {
-						errorList.pop_front();  // 只有非空白字符串才会真正返回
-					}
-				}
-				pthread_mutex_unlock(&errorListMutex);
-			}
-		}
-
-		if (log_gotten) {
-			static uint32_t count = 0;
-			my_rocketmq_send(log_send);
-		}*/
 
 		QThread::msleep(10);  // 避免 CPU 空转lnk20250326
 
@@ -649,15 +570,17 @@ std::string extractDataJson(const char* inputJson) {
 	    return "";
 	}
 
-	/*//添加guid
+	//添加guid
 	// 提取 "guid" 部分
     cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
-    if (guidstr == NULL || data->type != cJSON_String) {
+    if (guidstr == NULL || guidstr->type != cJSON_String) {
         std::cerr << "'guid' is missing or is not an array" << std::endl;
         cJSON_Delete(root);
         return "";
     }
-	//guid回复*/
+	//guid回复
+	std::string guid = guidstr->valuestring;
+	send_reply_to_kafka(guid,"1","收到补招指令");
 
     // 提取 "data" 部分
     cJSON* data = cJSON_GetObjectItem(messageBody, "data");
@@ -737,8 +660,9 @@ bool parseJsonMessageRT(const std::string& body, std::string& devSeries, std::st
     cJSON* limitItem = cJSON_GetObjectItem(messageBody, "limit");
 
 	//添加guid
-	//cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
+	std::string guid;
-	//if(guidstr)std::string guid = guidstr->valuestring;
+	cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
+	if(guidstr)guid = guidstr->valuestring;
 
     if (devSeriesItem && lineItem && realDataItem && soeDataItem && limitItem) {
         devSeries = devSeriesItem->valuestring;
@@ -748,7 +672,8 @@ bool parseJsonMessageRT(const std::string& body, std::string& devSeries, std::st
         limit = limitItem->valueint;
 
 		//回复消息
-		//guid
+		//执行结果直接看实时数据，不需要再回复，1是收到消息
+		send_reply_to_kafka(guid,"1","收到三秒数据指令");
 
     } else {
         std::cerr << "Missing expected fields in JSON message." << std::endl;
@@ -1031,8 +956,7 @@ void parse_set(const std::string& json_str) {
 
 	std::string frontType = front->valuestring;
 
-	//进程号为0的进程处理所有控制消息
+	if (index_value != g_front_seg_index) {
-	if (index_value != g_front_seg_index && g_front_seg_index !=0) {
     	std::cout << "msg index:"<< index_value <<"doesnt match self index:" << g_front_seg_index << std::endl;
         cJSON_Delete(root);
         return;
@@ -1068,6 +992,7 @@ void parse_set(const std::string& json_str) {
 
 				//脚本在3秒后执行
 				//回复消息
+				send_reply_to_kafka(guid,"1","收到重置进程指令,重启所有进程!");
 
 				//上送日志
 
@@ -1081,6 +1006,11 @@ void parse_set(const std::string& json_str) {
 			//等待一会后退出进程
 			MVL_LOG_ACSE0("MYLOG: recive delete msg, so exit to restart ");
 
+			//回复消息
+			send_reply_to_kafka(guid,"1","收到删除进程指令,这个进程将会重启 ");
+
+			//上送日志
+
 			apr_sleep(apr_time_from_sec(10));
 			exit(-1039);
 		}
@@ -1444,7 +1374,6 @@ void parse_log(const std::string& json_str) {
 	
 	std::string frontType = frontTypestr->valuestring;
 
-	//进程号为0的进程处理所有台账更新消息
 	if (processNo != g_front_seg_index) {
     	std::cout << "msg index:"<< processNo <<"doesnt match self index:" << g_front_seg_index << std::endl;
         cJSON_Delete(root);
@@ -1457,10 +1386,13 @@ void parse_log(const std::string& json_str) {
         return;
 	}
 
-	//进程号匹配上
+	//进程号和匹配上
 	std::cout << "msg index:"<< processNo <<"  self index:" << g_front_seg_index << std::endl;
 	std::cout << "msg frontType:"<< frontType <<"  self frontType:" << subdir << std::endl;
 
+	//回复消息
+	send_reply_to_kafka(guid,"1","收到实时日志指令");
+
 	if (code_str == "set_log") {
 		//校验数据
 		if((level == "terminal" || level == "measurepoint") &&
@@ -1560,7 +1492,8 @@ void parse_control(const std::string& json_str, const std::string& output_dir) {
 	std::cout << "msg index:"<< process_No <<"  self index:" << g_front_seg_index << std::endl;
 
 	//匹配后响应收到台账更新消息
-	//guid
+	//除了回复收到消息，执行结束后还要回复结果
+	send_reply_to_kafka(guid,"1","收到台账更新指令");
 
     if (code_str == "add_terminal" || code_str == "ledger_modify") {
 
@@ -2032,24 +1965,22 @@ void mqconsumerThread::run()
     std::vector<Subscription> subscriptions;
 	// 初始化消费者1 //lnk20241230只有实时进程会订阅实时topic，不订阅实时topic的进程无法触发实时数据
 	if(g_node_id == THREE_SECS_DATA_BASE_NODE_ID){
-		subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_RT, G_MQCONSUMER_TAG_RT, myMessageCallbackrtdata));
+		subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_RT, G_MQCONSUMER_TAG_RT, myMessageCallbackrtdata));
 	}
 
 	// 初始化消费者2 //所有进程都会订阅台账更新topic，不同功能进程的台账不能互相影响
-	subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_UD, G_MQCONSUMER_TAG_UD, myMessageCallbackupdate));
+	subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_UD, G_MQCONSUMER_TAG_UD, myMessageCallbackupdate));
 
 	// 初始化消费者3 //lnk20241230只有补招进程会订阅补招topic，不订阅补招topic的进程无法触发补招数据
 	if(g_node_id == RECALL_HIS_DATA_BASE_NODE_ID){
-		subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_RC, G_MQCONSUMER_TAG_RC, myMessageCallbackrecall));
+		subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_RC, G_MQCONSUMER_TAG_RC, myMessageCallbackrecall));
 	}
 
-	// 初始化消费者4 //lnk20250108只有稳态进程1会订阅控制topic，不订阅控制topic的进程无法触发进程重置
+	// 初始化消费者4 //lnk20250108只有稳态进程1会控制reset，
-	if(g_node_id == STAT_DATA_BASE_NODE_ID && g_front_seg_index == 1){
+	subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_SET, G_MQCONSUMER_TAG_SET, myMessageCallbackset));
-		subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_SET, G_MQCONSUMER_TAG_SET, myMessageCallbackset));
-	}
 
 	// 初始化消费者5 //所有进程都会订阅日志上送topic，不同功能进程的日志上送不能互相影响
-	subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_LOG, G_MQCONSUMER_TAG_LOG, myMessageCallbacklog));
+	subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_LOG, G_MQCONSUMER_TAG_LOG, myMessageCallbacklog));
 
     try {
         rocketmq_consumer_receive(consumerName, nameServer, subscriptions);

log4cplus/log4.h

@@ -50,8 +50,10 @@ struct DebugSwitch {
 extern std::map<std::string, TypedLogger> logger_map;
 extern DebugSwitch g_debug_switch;
 
+extern void send_reply_to_kafka(const std::string& guid, const std::string& step, const std::string& result);
 
 
+std::string get_front_type_from_subdir();
 
 
 log4cplus::Logger init_logger(const std::string& full_name, const std::string& file_dir, const std::string& base_file);
@@ -73,6 +75,8 @@ 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_kafka_c(const char* guid, const char* step, const char* result);
 #ifdef __cplusplus
 }
 #endif

mms/main.c

@@ -315,14 +315,14 @@ int main(int argc, const char **argv)
 		}
 
 		//lnk20241211 添加测试开关
-		pthread_mutex_lock(&mtx);
+		/*pthread_mutex_lock(&mtx);
 		if (!G_TEST_FLAG && g_front_num_count >= 30 && g_onlyIP[0] == 0 && g_node->n_clients>10) {//30分钟连接数量过低且不是单连且台账大于十个终端
 			MVL_LOG_ACSE0("MYLOG: g_front_num_count>=20, so exit to restart ");
 
 			apr_sleep(apr_time_from_sec(10));
 			exit(-1039);
 		}
-		pthread_mutex_unlock(&mtx);
+		pthread_mutex_unlock(&mtx);*/
 		
 	}
 

mms/mms_process.c

@@ -671,10 +671,17 @@ void check_3s_config()
 int isValidModelId(const char* model_id) {
 	size_t i;
 	
-    if (model_id == NULL) return 0;  // NULL 无效
+    if (model_id == NULL) 
+	{
+		printf("!!!model_id null!!!\n");
+		return 0;
+	}  // NULL 无效
 
     size_t len = strlen(model_id);
-    if (len < 4) return 0;  // 长度 < 4 无效
+    if (len < 4) {
+		printf("!!!model_id length < 4!!!\n");
+		return 0;  // 长度 < 4 无效
+	}
 
     // 检查是否全是空格
     for (i = 0; i < len; i++) {
@@ -682,6 +689,7 @@ int isValidModelId(const char* model_id) {
             return 1;  // 只要包含非空格字符，就是合法的
         }
     }
+	printf("!!!model_id only space!!!\n");
     return 0;  // 仅包含空格，无效
 }
 
@@ -774,6 +782,9 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 
 					//添加mq响应台账添加失败：台账挂满
 					//update[i].guid
+					char msg[256];
+					sprintf(msg, "终端 id: %s 台账更新失败, 这个进程的台账空间已满，达到了配置台账数量的最大值", update[i].terminal_id);
+					send_reply_to_kafka_c(update[i].guid, "2", msg);
 
 					return;
 				}
@@ -798,14 +809,24 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 				
 				ied_usr = (ied_usr_t*)apr_pcalloc(g_init_pool, sizeof(ied_usr_t));
 				ied->usr_ext = ied_usr;//内存挂到ied上
-				if (ied_usr == NULL)
+				if (ied_usr == NULL){
-					return APR_ENOMEM;
+					char msg[256];
+					snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,没有找到台账的终端空间", update[i].terminal_id);
+					send_reply_to_kafka_c(update[i].guid, "2", msg);
+					return ;
+				}
+					
 
 				ied_usr->last_call_wavelist_time = sGetMsTime() + g_pt61850app->giTime * 1000;//从FeProject/子功能目录/etc/pt61850netd_pqfe.xml中读取的总查询时间
 				
 				ied_usr->LD_info = (LD_info_t*)apr_pcalloc(g_init_pool, MAX_CPUNO * sizeof(LD_info_t));//内存挂到ied上
-				if (ied_usr->LD_info == NULL)
+				if (ied_usr->LD_info == NULL){
-					return APR_ENOMEM;
+					char msg[256];
+					snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,没有找到台账的监测点空间", update[i].terminal_id);
+					send_reply_to_kafka_c(update[i].guid, "2", msg);
+					return ;
+				}
+					
 
 				ied_usr->dev_flag = ENABLE;//终端有效
 
@@ -821,22 +842,27 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 			int ret = update_one_terminal_ledger(update,i,ied,terminal_index,ied_take);
 			if(ret){
 				printf("ledger can not be update!!!!!quit process!!!!!\n");	
-				return 0;
+				char msg[256];
+				snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,无法写入台账", update[i].terminal_id);
+				send_reply_to_kafka_c(update[i].guid, "2", msg);
 			}
 			//3-写入台账内容///////////////////////////////////
 
 			//4-配置映射文件//////////////////////////////
 			char model[64];
     		// 获取模型ID，检查是否返回 NULL
-    		char* model_id = parse_model_cfg_web_one(ied);//存储在/FeProject/dat/
+    		parse_model_cfg_web_one(ied,model);//存储在/FeProject/dat/
 
-    		if (isValidModelId(model_id)) { //lnk20250313防止拿不到映射文件
+    		if (isValidModelId(model)) { //lnk20250313防止拿不到映射文件
     		    // 安全拷贝字符串到 model 数组
-    		    strncpy(model, model_id, sizeof(model) - 1);
+    		    strncpy(model, model, sizeof(model) - 1);
     		    model[sizeof(model) - 1] = '\0';  // 确保以 null 结尾
     		    printf("ledger Model ID: %s\n", model);
     		} else {
     		    printf("ledger No model ID found.quit\n");
+				char msg[256];
+				snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,没有找到装置型号", update[i].terminal_id);
+				send_reply_to_kafka_c(update[i].guid, "2", msg);
 				return ;
     		}
 			char full_path[128];
@@ -865,6 +891,9 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 
 			//8响应添加成功
 			//update[i].guid
+			char msg[256];
+			snprintf(msg, sizeof(msg), "终端 id: %s 台账添加成功", update[i].terminal_id);
+			send_reply_to_kafka_c(update[i].guid, "2", msg);
 			
 		}
 	} 
@@ -908,22 +937,31 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 				int ret = update_one_terminal_ledger(update,i,ied,ied_usr->dev_idx,1);//1：更新已有的ied
 				if(ret){
 					printf("ledger can not be update!!!!!quit process!!!!!\n");	
-					return 0;
+
+					char msg[256];
+					snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,台账无法写入", update[i].terminal_id);
+					send_reply_to_kafka_c(update[i].guid, "2", msg);
+					return ;
 				}
 				//3-写入台账内容////////////////////////////////////////////
 
 				//4-配置映射文件///////////////////////////////////////////
 				char model[64] = {0};
     			// 获取模型ID，检查是否返回 NULL
-    			char* model_id = parse_model_cfg_web_one(ied);//存储在/FeProject/dat/
+    			parse_model_cfg_web_one(ied,model);//存储在/FeProject/dat/
 
-    			if (model_id != NULL) {
+    			if (isValidModelId(model)) {
     			    // 安全拷贝字符串到 model 数组
-    			    strncpy(model, model_id, sizeof(model) - 1);
+    			    strncpy(model, model, sizeof(model) - 1);
     			    model[sizeof(model) - 1] = '\0';  // 确保以 null 结尾
     			    printf("ledger Model ID: %s\n", model);
     			} else {
     			    printf("ledger No model ID found.\n");
+
+					char msg[256];
+					snprintf(msg, sizeof(msg), "终端 id: %s 台账更新失败,没有找到装置型号", update[i].terminal_id);
+					send_reply_to_kafka_c(update[i].guid, "2", msg);
+					return ;
     			}
 				char full_path[128];
         		snprintf(full_path, sizeof(full_path), "/FeProject/dat/%s.xml", model);  // 拼接路径
@@ -949,6 +987,9 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 				
 				//8响应添加成功
 				//update[i].guid
+				char msg[256];
+				snprintf(msg, sizeof(msg), "终端 id: %s 台账修改成功", update[i].terminal_id);
+				send_reply_to_kafka_c(update[i].guid, "2", msg);
 			}
 		}		
 
@@ -957,6 +998,9 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 
 			//添加mq响应台账添加失败：台账找不到
 			//update[i].guid
+			char msg[256];
+			sprintf(msg, "终端 id: %s 台账修改失败, 无法找到这个终端", update[i].terminal_id);
+			send_reply_to_kafka_c(update[i].guid, "2", msg);
 		}		
 	}
 ///////////////////////////////////////////////////////////////////////////////delete
@@ -1032,7 +1076,10 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 				//更新数据//////////////////////////////////////////////////////////////////////
 
 				//7响应添加成功
-				//update[i].guid	
+				//update[i].guid
+				char msg[256];
+				snprintf(msg, sizeof(msg), "终端 id: %s 台账删除成功", update[i].terminal_id);
+				send_reply_to_kafka_c(update[i].guid, "2", msg);
 			}
 		}		
 
@@ -1041,6 +1088,10 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
 
 			//添加mq响应台账添加失败：台账找不到
 			//update[i].guid
+			
+			char msg[256];
+			sprintf(msg, "终端 id: %s 台账删除失败, 无法找到这个终端", update[i].terminal_id);
+			send_reply_to_kafka_c(update[i].guid, "2", msg);
 
 		}		
 	}

mms/rdb_client.h

@@ -472,7 +472,7 @@ bool isCharPtrEmpty(const char* str);
 int parse_ledger_update_xml(trigger_update_xml_t* trigger_update_xml);
 int update_one_terminal_ledger(terminal* update, int i,ied_t* ied,int terminal_index,int ied_take);
 void print_trigger_update_xml(const trigger_update_xml_t* trigger_update);
-char* parse_model_cfg_web_one(ied_t* ied);
+char* parse_model_cfg_web_one(ied_t* ied,char* out_model);
 void Set_xml_nodeinfo_one(char* dev_type);
 void create_ledger_log(trigger_update_xml_t* ledger_update_xml);
 ied_t* find_ied_unused();

rocketmq/SimpleProducer.h

@@ -8,7 +8,7 @@
 #include "../rocketmq/CPushConsumer.h"
 
 #include <vector>
-
+#include <iostream>
 
 /*添加测试函数lnk10-10*/
 void producer_send0();
@@ -42,7 +42,7 @@ struct Subscription {
     MessageCallBack callback;
     
     Subscription(const std::string& t, const std::string& tg, MessageCallBack cb)
-        : topic(t), tag(tg), callback(cb) {}
+        : topic(t), tag(tg), callback(cb) {std::cout << "Subscription topic: " << topic << std::endl;}
 };
 
 void rocketmq_consumer_receive(