add log realdata send funtion

cfg_parse/cfg_parser.cpp

@@ -312,6 +312,12 @@ std::string G_MQCONSUMER_KEY_RC = "";//consumer key
 std::string G_MQCONSUMER_TOPIC_SET = "";//consumer topie
 std::string G_MQCONSUMER_TAG_SET = "";//consumer tag
 std::string G_MQCONSUMER_KEY_SET = "";//consumer key
+std::string G_LOG_TOPIC = "";//topie
+std::string G_LOG_TAG = "";//tag
+std::string G_LOG_KEY = "";//key
+std::string G_MQCONSUMER_TOPIC_LOG = "";//consumer topie
+std::string G_MQCONSUMER_TAG_LOG = "";//consumer tag
+std::string G_MQCONSUMER_KEY_LOG = "";//consumer key
 
 int G_TEST_FLAG = 0;
 int G_TEST_NUM = 0;
@@ -896,6 +902,19 @@ void init_config() {
 	G_MQCONSUMER_TAG_SET = strdup(ba.data());
 	ba = settings.value("RocketMq/ConsumerKeySET", "").toString().toLatin1();
 	G_MQCONSUMER_KEY_SET = strdup(ba.data());
+
+	ba = settings.value("RocketMq/ConsumerTopicLOG", "").toString().toLatin1();
+	G_MQCONSUMER_TOPIC_LOG = strdup(ba.data());
+	ba = settings.value("RocketMq/ConsumerTagLOG", "").toString().toLatin1();
+	G_MQCONSUMER_TAG_LOG = strdup(ba.data());
+	ba = settings.value("RocketMq/ConsumerKeyLOG", "").toString().toLatin1();
+	G_MQCONSUMER_KEY_LOG = strdup(ba.data());
+	ba = settings.value("RocketMq/LOGTopic", "").toString().toLatin1();
+	G_LOG_TOPIC = strdup(ba.data());
+	ba = settings.value("RocketMq/LOGTag", "").toString().toLatin1();
+	G_LOG_TAG = strdup(ba.data());
+	ba = settings.value("RocketMq/LOGKey", "").toString().toLatin1();
+	G_LOG_KEY = strdup(ba.data());
 //MQ<4D><51><EFBFBD><EFBFBD>
 	G_TEST_FLAG = settings.value("RocketMq/Testflag", 0).toInt();
 	G_TEST_NUM = settings.value("RocketMq/Testnum", 0).toInt();
@@ -909,6 +928,9 @@ void init_config() {
 	std::cout << "Read G_ROCKETMQ_TAG:" << G_ROCKETMQ_TAG << std::endl;
 	std::cout << "Read G_ROCKETMQ_KEY:" << G_ROCKETMQ_KEY << std::endl;
 	std::cout << "Read QUEUENUM:" << QUEUENUM << std::endl;
+	std::cout << "Read G_LOG_TOPIC:" << G_LOG_TOPIC << std::endl;
+	std::cout << "Read G_LOG_TAG:" << G_LOG_TAG << std::endl;
+	std::cout << "Read G_LOG_KEY:" << G_LOG_KEY << std::endl;
 //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ش<EFBFBD>ӡ
 	std::cout << "Read G_ROCKETMQ_CONSUMER:" << G_ROCKETMQ_CONSUMER << std::endl;
 	std::cout << "Read G_MQCONSUMER_IPPORT:" << G_MQCONSUMER_IPPORT << std::endl;
@@ -927,6 +949,9 @@ void init_config() {
 	std::cout << "Read G_MQCONSUMER_TOPIC_SET:" << G_MQCONSUMER_TOPIC_SET << std::endl;
 	std::cout << "Read G_MQCONSUMER_TAG_SET:" << G_MQCONSUMER_TAG_SET << std::endl;
 	std::cout << "Read G_MQCONSUMER_KEY_SET:" << G_MQCONSUMER_KEY_SET << std::endl;
+	std::cout << "Read G_MQCONSUMER_TOPIC_LOG:" << G_MQCONSUMER_TOPIC_LOG << std::endl;
+	std::cout << "Read G_MQCONSUMER_TAG_LOG:" << G_MQCONSUMER_TAG_LOG << std::endl;
+	std::cout << "Read G_MQCONSUMER_KEY_LOG:" << G_MQCONSUMER_KEY_LOG << std::endl;
 //Mq<4D><71><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ش<EFBFBD>ӡ
 	std::cout << "Read G_TEST_FLAG:" << G_TEST_FLAG << std::endl;
 	std::cout << "Read G_TEST_NUM:" << G_TEST_NUM << std::endl;
@@ -15137,6 +15162,115 @@ void rocketmq_test_300(int mpnum,int front_index) {
     std::cout << "Finished sending " << total_messages << " messages." << std::endl;
 }
 
+///////////////////////////////////////////////////////////////////////////////lnkʵʱ<CAB5><CAB1>־<EFBFBD><D6BE><EFBFBD><EFBFBD>20250205
+// ȫ<><C8AB><EFBFBD>б<EFBFBD><D0B1>Ϳ<EFBFBD><CDBF><EFBFBD>
+std::list<std::string> errorList;
+std::list<std::string> warnList;
+std::list<std::string> normalList;
+
+// <20><><EFBFBD><EFBFBD>ȫ<EFBFBD>ֻ<EFBFBD><D6BB><EFBFBD><EFBFBD><EFBFBD>
+pthread_mutex_t errorListMutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t warnListMutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t normalListMutex = PTHREAD_MUTEX_INITIALIZER;
+	
+bool errorOutputEnabled = false;
+bool warnOutputEnabled = false;
+bool normalOutputEnabled = false;
+
+// <20>Զ<EFBFBD><D4B6><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+class RedirectStreamBuf : public std::streambuf {
+	public:
+		RedirectStreamBuf(std::list<std::string>& targetList, pthread_mutex_t& targetMutex)
+			: targetList(targetList), targetMutex(targetMutex) {}
+	
+	protected:
+		virtual int_type overflow(int_type ch) override {
+			if (ch != EOF) {
+				char c = static_cast<char>(ch);
+	
+				// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD> targetList
+				pthread_mutex_lock(&targetMutex);
+				targetList.push_back(std::string(1, c));
+				pthread_mutex_unlock(&targetMutex);
+			}
+			return ch;
+		}
+	
+	private:
+		std::list<std::string>& targetList;
+		pthread_mutex_t& targetMutex;
+};
+	
+// <20>Զ<EFBFBD><D4B6><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+void redirectErrorOutput(bool enabled) {
+	errorOutputEnabled = enabled;
+	if (enabled) {
+		static RedirectStreamBuf errorBuf(errorList, errorListMutex);
+		std::cerr.rdbuf(&errorBuf);
+	} else {
+		std::cerr.rdbuf(nullptr); // <20>ָ<EFBFBD><D6B8><EFBFBD><EFBFBD><EFBFBD>׼<EFBFBD><D7BC><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+	}
+}
+
+// <20>Զ<EFBFBD><D4B6><EFBFBD><EFBFBD>澯<EFBFBD><E6BEAF><EFBFBD><EFBFBD>
+void redirectWarnOutput(bool enabled) {
+	warnOutputEnabled = enabled;
+	if (enabled) {
+		static RedirectStreamBuf warnBuf(warnList, warnListMutex);
+		std::clog.rdbuf(&warnBuf);
+	} else {
+		std::clog.rdbuf(nullptr); // <20>ָ<EFBFBD><D6B8><EFBFBD><EFBFBD><EFBFBD>׼<EFBFBD>澯<EFBFBD><E6BEAF>
+	}
+}
+
+// <20>Զ<EFBFBD><D4B6><EFBFBD><EFBFBD><EFBFBD>ͨ<EFBFBD><CDA8><EFBFBD><EFBFBD>
+void redirectNormalOutput(bool enabled) {
+	normalOutputEnabled = enabled;
+	if (enabled) {
+		static RedirectStreamBuf normalBuf(normalList, normalListMutex);
+		std::cout.rdbuf(&normalBuf);
+	} else {
+		std::cout.rdbuf(nullptr); // <20>ָ<EFBFBD><D6B8><EFBFBD><EFBFBD><EFBFBD>׼<EFBFBD><D7BC><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+	}
+}
 
 
+// <20>Զ<EFBFBD><D4B6><EFBFBD> printf <20><><EFBFBD><EFBFBD>
+int customPrintf(const char* format, ...) {
+    va_list args;
+    va_start(args, format);
+    char buffer[1024];
+    vsnprintf(buffer, sizeof(buffer), format, args);
+    va_end(args);
+
+    // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>п<EFBFBD><D0BF>ض<EFBFBD>û<EFBFBD><C3BB><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʹ<EFBFBD><CAB9>ԭ<EFBFBD><D4AD> printf <20><><EFBFBD><EFBFBD>
+    if (!errorOutputEnabled && !warnOutputEnabled && !normalOutputEnabled) {
+        vprintf(format, args);
+        return 0;  // <20><><EFBFBD><EFBFBD>ֵΪ<D6B5>Ѵ<EFBFBD>ӡ<EFBFBD>ַ<EFBFBD><D6B7><EFBFBD>
+    }
+
+    // <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>õĿ<C3B5><C4BF>ؽ<EFBFBD><D8BD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ӵ<EFBFBD><D3B5><EFBFBD>Ӧ<EFBFBD><D3A6><EFBFBD>б<EFBFBD><D0B1><EFBFBD>
+    if (errorOutputEnabled) {
+        pthread_mutex_lock(&errorListMutex);  // <20><><EFBFBD><EFBFBD> errorList
+        errorList.push_back(buffer);
+        pthread_mutex_unlock(&errorListMutex);  // <20><><EFBFBD><EFBFBD> errorList
+    }
+
+    if (warnOutputEnabled) {
+        pthread_mutex_lock(&warnListMutex);  // <20><><EFBFBD><EFBFBD> warnList
+        warnList.push_back(buffer);
+        pthread_mutex_unlock(&warnListMutex);  // <20><><EFBFBD><EFBFBD> warnList
+    }
+
+    if (normalOutputEnabled) {
+        pthread_mutex_lock(&normalListMutex);  // <20><><EFBFBD><EFBFBD> normalList
+        normalList.push_back(buffer);
+        pthread_mutex_unlock(&normalListMutex);  // <20><><EFBFBD><EFBFBD> normalList
+    }
+
+    return 0;  // <20><><EFBFBD><EFBFBD>ֵΪ<D6B5>Ѵ<EFBFBD>ӡ<EFBFBD>ַ<EFBFBD><D6B7><EFBFBD>
+}
+
+///////////////////////////////////////////////////////////////////////////////
+
 

cfg_parse/custom_printf.h

@@ -0,0 +1,26 @@
+// custom_printf.h
+#ifndef CUSTOM_PRINTF_H
+#define CUSTOM_PRINTF_H
+
+#include <stdio.h>
+#include <stdarg.h>
+#include <list>
+#include <string>
+
+// 假设这些是你管理输出的列表
+extern std::list<std::string> errorList;
+extern std::list<std::string> warnList;
+extern std::list<std::string> normalList;
+
+// 开关
+extern bool errorOutputEnabled;
+extern bool warnOutputEnabled;
+extern bool normalOutputEnabled;
+
+// 自定义的 printf 函数
+int customPrintf(const char* format, ...);
+
+// 使用宏将 printf 替换为 customPrintf
+#define printf customPrintf
+
+#endif // CUSTOM_PRINTF_H

json/save2json.cpp

@@ -118,6 +118,9 @@ extern std::string G_MQCONSUMER_KEY_RC;//key
 extern std::string G_MQCONSUMER_TOPIC_SET;//topie_recall
 extern std::string G_MQCONSUMER_TAG_SET;//tag
 extern std::string G_MQCONSUMER_KEY_SET;//key
+extern std::string G_MQCONSUMER_TOPIC_LOGSET;//topie_log
+extern std::string G_MQCONSUMER_TAG_LOGSET;//tag
+extern std::string G_MQCONSUMER_KEY_LOGSET;//key
 
 #define APRTIME_8H  (28800000000ULL)
 #define APRTIME_1H  (3600000000ULL)
@@ -127,11 +130,8 @@ const int MAX_LIST_SIZE = 16;
 static QMap<int, QList<long long> > real_data_report_map;
 static QMap<QString, json_block_data*> json_data_map;//CZY 2023-08-17 ww 2023<32><33>3<EFBFBD><33>13<31><33>17:23:17<31><37>չMap<61><70><EFBFBD><EFBFBD><EFBFBD>ڱ<EFBFBD><DAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
 static QMap<QString, json_block_data*> json_flicker_data_map;//CZY 2023-09-11 չMap<61><70><EFBFBD><EFBFBD><EFBFBD>ڱ<EFBFBD><DAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
-
 static QMap<QString, json_block_data*> json_pst_data_map;//CZY 2023-09-11 չMap<61><70><EFBFBD><EFBFBD><EFBFBD>ڱ<EFBFBD><DAB1><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>·<EFBFBD><C2B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
 
-
-
 int	urcbRealDataHasReceived(int dev_index, LD_info_t* LD_info, long long Time)
 {
 	QList<long long>& ts_list = real_data_report_map[LD_info->line_id];
@@ -584,6 +584,21 @@ 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<32><35><EFBFBD><EFBFBD><EFBFBD><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>
+		Ckafka_data_t log_send;
+		log_send.strTopic
+		bool log_gotten;
+
+		log_gotten = false;
+		pthread_mutex_lock(&targetMutex);
+		if (!kafka_data_list.isEmpty()) {
+			data_gotten = true;
+			log_send = kafka_data_list.takeFirst();
+		}
+		kafka_data_list_mutex.unlock();
+
+		if (log_gotten && ) {
 
 		/*if (data_gotten) {
 			LD_info_t* LD_info = find_LD_info_only_from_mp_id(data.mp_id.toAscii().data());
@@ -1358,6 +1373,131 @@ int StringToInt(const std::string& str) {
     return number;
 }
 
+// <20><><EFBFBD><EFBFBD> JSON <20>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD>ִ<EFBFBD><D6B4><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6><EFBFBD><EFBFBD>
+void parse_log(const std::string& json_str, const std::string& output_dir) {
+    // <20><><EFBFBD><EFBFBD> JSON <20>ַ<EFBFBD><D6B7><EFBFBD>
+    cJSON* root = cJSON_Parse(json_str.c_str());
+    if (root == nullptr) {
+        std::cout << "Error parsing JSON." << std::endl;
+        return;
+    }
+
+	// <20><>ȡ "messageBody" <20><><EFBFBD><EFBFBD>
+	cJSON* messageJson = cJSON_GetObjectItem(root, "messageBody");
+	if (messageJson == NULL || messageJson->type != cJSON_String) {
+	   std::cerr << "'messageJson' is missing or is not an cJSON_String" << std::endl;
+	   cJSON_Delete(root);
+	   return ;
+	}
+
+	// <20><><EFBFBD><EFBFBD> messageBody <20>е<EFBFBD> JSON <20>ַ<EFBFBD><D6B7><EFBFBD>
+	const char* messageBodyStr = messageJson->valuestring;
+	if (messageBodyStr == nullptr || strlen(messageBodyStr) == 0) {
+		std::cerr << "Failed to parse 'messageBody' JSON or it's empty." << std::endl;
+		cJSON_Delete(root);
+		return;
+	}
+
+	cJSON* messageBody = cJSON_Parse(messageBodyStr); // <20><><EFBFBD><EFBFBD> messageBody <20>ַ<EFBFBD><D6B7><EFBFBD>
+	if (messageBody == NULL) {
+	    std::cerr << "Failed to parse 'messageBody' JSON." << std::endl;
+	    cJSON_Delete(root);
+	    return ;
+	}
+
+    // <20><>ȡ code <20>ֶ<EFBFBD>
+    cJSON* code = cJSON_GetObjectItem(messageBody, "code");
+    if (code == nullptr) {
+        std::cout << "Missing 'code' in JSON." << std::endl;
+        cJSON_Delete(root);
+        return;
+    }
+
+	cJSON* index = cJSON_GetObjectItem(messageBody, "index");
+	if (index == nullptr) {
+        std::cout << "Missing 'index' in JSON." << std::endl;
+        cJSON_Delete(root);
+        return;
+    }
+
+	//<2F>ж<EFBFBD><D0B6>Dz<EFBFBD><C7B2><EFBFBD><EFBFBD>Լ<EFBFBD><D4BC><EFBFBD><EFBFBD>̺ţ<CCBA>
+	int index_value = index->valueint;
+	//string index_value_str = index->valuestring;
+	//int index_value = StringToInt(index_value_str);
+
+
+	//<2F><><EFBFBD>̺<EFBFBD>Ϊ0<CEAA>Ľ<EFBFBD><C4BD>̴<EFBFBD><CCB4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>̨<EFBFBD>˸<EFBFBD><CBB8><EFBFBD><EFBFBD><EFBFBD>Ϣ
+	if (index_value != g_front_seg_index) {
+    	std::cout << "msg index:"<< index_value <<"doesnt match self index:" << g_front_seg_index << std::endl;
+        cJSON_Delete(root);
+        return;
+	}
+
+	//<2F><><EFBFBD>̺<EFBFBD>ƥ<EFBFBD><C6A5><EFBFBD><EFBFBD>
+	std::cout << "msg index:"<< index_value <<"  self index:" << g_front_seg_index << std::endl;
+
+    // <20><><EFBFBD><EFBFBD> code <20>ֶ<EFBFBD>ִֵ<D6B5>в<EFBFBD>ͬ<EFBFBD>Ľ<EFBFBD><C4BD><EFBFBD><EFBFBD>߼<EFBFBD>
+    std::string code_str = code->valuestring;
+
+	if (code_str == "set_log") {
+
+		// <20><><EFBFBD><EFBFBD> set_process
+        cJSON* data = cJSON_GetObjectItem(messageBody, "data");
+        if (data != nullptr && data->type == cJSON_Array) {
+            int data_size = cJSON_GetArraySize(data);
+            for (int i = 0; i < data_size; i++) {
+                cJSON* item = cJSON_GetArrayItem(data, i);
+                std::string fun = cJSON_GetObjectItem(item, "fun")->valuestring;
+                std::string level = cJSON_GetObjectItem(item, "level")->valuestring;
+                std::string frontType = cJSON_GetObjectItem(item, "frontType")->valuestring;
+				//У<><D0A3><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+				if(frontType == subdir){
+					if(fun == "open"){
+						if (code_str == "ERROR"){
+							// <20><><EFBFBD>ô<EFBFBD><C3B4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+							redirectErrorOutput(true);
+						}
+						else if (code_str == "WARN"){
+							// <20><><EFBFBD>ø澯<C3B8><E6BEAF><EFBFBD><EFBFBD>
+							redirectWarnOutput(true);
+						}
+						else if (code_str == "NORMAL"){
+							// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͨ<EFBFBD><CDA8><EFBFBD><EFBFBD>
+							redirectNormalOutput(true);
+						}
+						else{
+							std::cout << "code_str error" <<std::endl;
+						}
+					}
+					else{
+						if (code_str == "ERROR"){
+							// <20>رմ<D8B1><D5B4><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+							redirectErrorOutput(false);
+						}
+						else if (code_str == "WARN"){
+							// <20><><EFBFBD>ø澯<C3B8><E6BEAF><EFBFBD><EFBFBD>
+							redirectWarnOutput(false);
+						}
+						else if (code_str == "NORMAL"){
+							// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ͨ<EFBFBD><CDA8><EFBFBD><EFBFBD>
+							redirectNormalOutput(false);
+						}
+						else{
+							std::cout << "code_str error" <<std::endl;
+						}
+					} 
+				}
+				else{
+					std::cout << "front type doesnt match" <<std::endl;
+				}      
+        	}
+			std::cout << "this msg should only execute once" <<std::endl;
+        }
+
+    // <20>ͷ<EFBFBD> JSON <20><><EFBFBD><EFBFBD>
+    cJSON_Delete(root);
+}	
+
 // <20><><EFBFBD><EFBFBD> JSON <20>ַ<EFBFBD><D6B7><EFBFBD><EFBFBD><EFBFBD>ִ<EFBFBD><D6B4><EFBFBD><EFBFBD>Ӧ<EFBFBD><D3A6><EFBFBD><EFBFBD>
 void parse_control(const std::string& json_str, const std::string& output_dir) {
     // <20><><EFBFBD><EFBFBD> JSON <20>ַ<EFBFBD><D6B7><EFBFBD>
@@ -1794,6 +1934,40 @@ int myMessageCallbackset(CPushConsumer* consumer, CMessageExt* msg)
     return E_CONSUME_SUCCESS;
 }
 
+int myMessageCallbacklog(CPushConsumer* consumer, CMessageExt* msg)
+{
+    if (msg == NULL) {
+        std::cerr << "Received null message." << std::endl;
+        return E_RECONSUME_LATER;
+    }
+
+    const char* body = GetMessageBody(msg);
+    const char* key = GetMessageKeys(msg);
+
+    if (body == NULL) {
+        std::cerr << "Message body is NULL." << std::endl;
+        return E_RECONSUME_LATER;
+    }
+	else{
+		//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
+		// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD><CFA2><EFBFBD><EFBFBD><EFBFBD>磬<EFBFBD><E7A3AC>ӡ<EFBFBD><D3A1>Ϣ<EFBFBD><CFA2><EFBFBD>ݣ<EFBFBD>
+    	std::cout << "process Callback received message: " << body << std::endl;
+    	if (key) {
+    	    std::cout << "Message Key: " << key << std::endl;
+    	}
+    	else {
+    	    std::cout << "Message Key: N/A" << std::endl;
+    	}
+		
+		//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD≯<EFBFBD><CCB8><EFBFBD><EFBFBD><EFBFBD>Ϣ
+		parse_log(body);
+
+	}
+
+    // <20><><EFBFBD><EFBFBD>ҵ<EFBFBD><D2B5><EFBFBD>߼<EFBFBD><DFBC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>״̬
+    return E_CONSUME_SUCCESS;
+}
+
 int myMessageCallbackrecall(CPushConsumer* consumer, CMessageExt* msg)
 {
 	//<2F><><EFBFBD><EFBFBD>
@@ -1843,6 +2017,7 @@ int myMessageCallbackrecall(CPushConsumer* consumer, CMessageExt* msg)
     return E_CONSUME_SUCCESS;
 }
 
+
 void mqconsumerThread::run()
 {
 	// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>߲<EFBFBD><DFB2><EFBFBD>
@@ -1870,6 +2045,9 @@ void mqconsumerThread::run()
 		subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_SET, G_MQCONSUMER_TAG_SET, myMessageCallbackset));
 	}
 
+	// <20><>ʼ<EFBFBD><CABC><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>5 //<2F><><EFBFBD>н<EFBFBD><D0BD>̶<EFBFBD><CCB6>ᶩ<EFBFBD><E1B6A9><EFBFBD><EFBFBD>־<EFBFBD><D6BE><EFBFBD><EFBFBD>topic<69><63><EFBFBD><EFBFBD>ͬ<EFBFBD><CDAC><EFBFBD>ܽ<EFBFBD><DCBD>̵<EFBFBD><CCB5><EFBFBD>־<EFBFBD><D6BE><EFBFBD>Ͳ<EFBFBD><CDB2>ܻ<EFBFBD><DCBB><EFBFBD>Ӱ<EFBFBD><D3B0>
+	subscriptions.push_back(Subscription(G_MQCONSUMER_TOPIC_LOG, G_MQCONSUMER_TAG_LOG, myMessageCallbacklog));
+
     try {
         rocketmq_consumer_receive(consumerName, nameServer, subscriptions);
     }

mms/db_interface.h

@@ -17,6 +17,7 @@
 #include "apr_time.h"
 
 #include <stdbool.h>//lnk20241022
+#include "../cfgparse/custom_printf.h"//lnk20250225
 
 #define  LOG_IDX  (0)
 #define  RPT_IDX  (1)

mms/rdb_client.c

@@ -401,7 +401,7 @@ static void* APR_THREAD_FUNC rtdb_worker(apr_thread_t* thd, void* data)
 		/* Maintenance the clients request */
 	while (1) {
 
-		pthread_mutex_lock(&mtx);	printf("work hold lock !!!!!!!!!!!");
+		//pthread_mutex_lock(&mtx);	printf("work hold lock !!!!!!!!!!!");
 		
 		//<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
 		printf("check error4 !!!!!!!!!!!!!!\n");
@@ -421,6 +421,8 @@ static void* APR_THREAD_FUNC rtdb_worker(apr_thread_t* thd, void* data)
 		//check_recall_config();//<2F><><EFBFBD>ٽ<EFBFBD><D9BD>̶<EFBFBD>ȡ<EFBFBD><C8A1><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ
 		create_recall_xml();//<2F><><EFBFBD>ɴ<EFBFBD><C9B4><EFBFBD><EFBFBD><EFBFBD>xml<6D>ļ<EFBFBD>
 
+		pthread_mutex_lock(&mtx);	printf("work hold lock !!!!!!!!!!!");
+
 		check_ledger_update();//lnk20250113<31><33>ȡ̨<C8A1>˸<EFBFBD><CBB8>£<EFBFBD><C2A3><EFBFBD><EFBFBD><EFBFBD>̨<EFBFBD>˸<EFBFBD><CBB8><EFBFBD>
 
 		pthread_mutex_unlock(&mtx); printf("work free lock !!!!!!!!!!!");