zhangwen/front_linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '测试2' of http://192.168.1.22:3000/zhangwen/front_linux into 测试2

Browse Source
This commit is contained in:
zhangwen
2026-01-07 11:15:26 +08:00
parent cab2ab3453 02127869c0
commit 7c676162ef
10 changed files with 269 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
LFtid1056/cloudfront/code/cfg_parser.cpp
View File

@@ -566,7 +566,7 @@ void init_config() {
std::cout << "this is multiple process of index:" << g_front_seg_index << std::endl;
if(g_front_seg_index > g_front_seg_num){
DIY_ERRORLOG("process","【ERROR】前置当前进程的进程号为:%d,前置的多进程最大进程号为:%d,当前进程的进程号应该为1到最大进程号范围内的整数,退出该进程",g_front_seg_index,g_front_seg_num);
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG,"进程号应该为1到配置的最大进程号范围内的整数,退出当前进程");
exit(-1039);
}
@@ -576,7 +576,7 @@ void init_config() {
std::cout << "this is single process" << std::endl;
}
else{
DIY_ERRORLOG("process","【ERROR】前置当前进程的进程号为:%d,前置的多进程最大进程号为:%d,应该为大于0的整数,退出该进程",g_front_seg_index,g_front_seg_num);
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG,"进程号应该为1到配置的最大进程号范围内的整数,退出当前进程");
exit(-1039);
}
@@ -811,7 +811,6 @@ int parse_3s_xml(trigger_3s_xml_t* trigger_3s_xml) {
}
std::cout << "/etc/trigger3s/*.xml success..." << std::endl;
DIY_WARNLOG("process", "【WARN】前置读取实时数据触发文件成功,即将注册实时数据报告");
return 0;
}
@@ -882,7 +881,7 @@ int parse_recall_xml(recall_xml_t* recall_xml, const std::string& id) {
DIR* dir = opendir(cfg_dir.c_str());
if (!dir) {
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 无法解析补招文件，补招文件路径FRONT_PATH + /etc/recall/不存在", g_front_seg_index);
return false;
}
@@ -894,7 +893,7 @@ int parse_recall_xml(recall_xml_t* recall_xml, const std::string& id) {
std::string filepath = cfg_dir + "/" + filename;
tinyxml2::XMLDocument doc;
if (doc.LoadFile(filepath.c_str()) != tinyxml2::XML_SUCCESS) {
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 无法解析补招文件%s,补招内容无效", g_front_seg_index, filepath.c_str());
continue;
}
@@ -988,7 +987,7 @@ void DeletcRecallXml() {
DIR* dir = opendir(cfg_dir.c_str());
if (!dir) {
std::cerr << "folder does not exist!" << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 删除旧的补招文件失败,补招文件路径FRONT_PATH + /etc/recall/不存在", g_front_seg_index);
return;
}
@@ -1007,7 +1006,7 @@ void DeletcRecallXml() {
if (stat(fullpath.c_str(), &file_stat) == 0) {
if (file_stat.st_mtime < cutoff) {
if (remove(fullpath.c_str()) == 0) {
DIY_INFOLOG("process", "【NORMAL】前置的%d号进程 删除超过两天的补招文件", g_front_seg_index);
} else {
std::cerr << "Failed to remove file: " << fullpath << std::endl;
}
@@ -1029,7 +1028,7 @@ void CreateRecallXml() {
g_StatisticLackList_list_mutex.lock();
if (!g_StatisticLackList.empty()) {
DIY_INFOLOG("process", "【NORMAL】前置的%d号进程 开始写入补招文件", g_front_seg_index);
std::map<std::string, std::list<JournalRecall>> id_map;
for (const auto& jr : g_StatisticLackList) {
@@ -1069,7 +1068,7 @@ void CreateRecallXml() {
tinyxml2::XMLError save_result = doc.SaveFile(path.str().c_str());
if (save_result != tinyxml2::XML_SUCCESS) {
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 无法将补招文件写入路径: %s", g_front_seg_index, path.str().c_str());
continue;
}
}
@@ -1233,23 +1232,22 @@ int recall_json_handle_from_mq(const std::string& body)
} catch (const std::exception& e) {
std::cerr << "Error parsing JSON: " << e.what() << std::endl;
// ★与原逻辑等价：无法解析，不再进入 recall_json_handle
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,消息的json结构不正确",
g_front_seg_index, G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}
// 提取 "messageBody"（字符串）
if (!root.contains("messageBody") || !root["messageBody"].is_string()) {
std::cerr << "'messageBody' is missing or is not a string" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有messageBody字段",
g_front_seg_index, G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",
G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}
std::string messageBodyStr = root["messageBody"].get<std::string>();
if (messageBodyStr.empty()) {
std::cerr << "'messageBody' is empty" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,messageBody为空",
g_front_seg_index, G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",
G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}
@@ -1259,8 +1257,7 @@ int recall_json_handle_from_mq(const std::string& body)
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结构不正确",
g_front_seg_index, G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",G_MQCONSUMER_TOPIC_RC.c_str(), FRONT_INST.c_str());
return 10004;
}
@@ -1780,14 +1777,14 @@ int parse_ledger_update_xml(trigger_update_xml_t& trigger_update_xml)
std::this_thread::sleep_for(std::chrono::milliseconds(100));
if (!load_ledger_update_from_xml(trigger_update_xml, filename)) {
DIY_WARNLOG("process", "【WARN】成功读取台账更新文件: %s", filename.c_str());
}
if (std::remove(filename.c_str()) != 0) {
DIY_ERRORLOG("process", "【ERROR】删除台账更新文件失败: %s", filename.c_str());
return 1;
} else {
DIY_INFOLOG("process", "【NORMAL】成功删除台账更新文件: %s", filename.c_str());
}
}
@@ -5815,7 +5812,7 @@ void on_device_response_minimal(int response_code,
<< " rc=" << response_code << " recall_status=" << front.recall_status << std::endl; //错误响应码
//记录日志
DIY_ERRORLOG_CODE(matched_monitor->monitor_id.c_str(),2,static_cast<int>(LogCode::LOG_CODE_RECALL),"【ERROR】监测点:%s 补招数据失败 - 失败时间点:%s 至 %s",matched_monitor->monitor_id.c_str(),front.StartTime.c_str(),front.EndTime.c_str());
DIY_ERRORLOG_CODE(matched_monitor->monitor_id.c_str(),2,static_cast<int>(LogCode::LOG_CODE_RECALL),"监测点:%s 补招数据失败 - 失败时间点:%s 至 %s",matched_monitor->monitor_id.c_str(),front.StartTime.c_str(),front.EndTime.c_str());
}
else { //补招失败
front.recall_status = static_cast<int>(RecallStatus::FAILED);
@@ -5825,7 +5822,7 @@ void on_device_response_minimal(int response_code,
<< " rc=" << response_code << " recall_status=" << front.recall_status<< std::endl; //错误响应码
//记录日志
DIY_ERRORLOG_CODE(matched_monitor->monitor_id.c_str(),2,static_cast<int>(LogCode::LOG_CODE_RECALL),"【ERROR】监测点:%s 补招数据失败 - 失败时间点:%s 至 %s",matched_monitor->monitor_id.c_str(),front.StartTime.c_str(),front.EndTime.c_str());
DIY_ERRORLOG_CODE(matched_monitor->monitor_id.c_str(),2,static_cast<int>(LogCode::LOG_CODE_RECALL),"监测点:%s 补招数据失败 - 失败时间点:%s 至 %s",matched_monitor->monitor_id.c_str(),front.StartTime.c_str(),front.EndTime.c_str());
}
updated = true;
} else { //首条不是 RUNNING 状态，不应该收到这条响应

91
LFtid1056/cloudfront/code/interface.cpp
View File

@@ -147,14 +147,14 @@ void handleUploadResponse(const std::string& response, std::string& wavepath) {
}
catch (const json::parse_error& e) {
std::cerr << "Error parsing response: " << e.what() << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置上传暂态录波文件失败,web返回的消息不是json格式");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置上传文件失败,web响应异常");
return;
}
// 提取字段
if (!json_data.contains("code") || !json_data.contains("data")) {
std::cerr << "Error: Missing expected fields in JSON response." << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置上传暂态录波文件失败,web返回的消息没有远端文件名");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置上传文件失败,web响应异常");
return;
}
@@ -167,7 +167,7 @@ void handleUploadResponse(const std::string& response, std::string& wavepath) {
auto& data = json_data["data"];
if (!data.contains("name") || !data.contains("fileName") || !data.contains("url")) {
std::cerr << "Error: Missing expected fields in JSON data object." << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置上传暂态录波文件失败,web返回的消息没有远端文件名");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置上传文件失败,web响应异常");
return;
}
@@ -195,7 +195,7 @@ void handleUploadResponse(const std::string& response, std::string& wavepath) {
wavepath = nameWithoutExt;
std::cout << "wavepath: " << wavepath << std::endl;
DIY_INFOLOG("process", "【NORMAL】前置上传暂态录波文件成功,远端文件名:%s", wavepath.c_str());
DIY_INFOLOG("process", "前置上传暂态录波文件成功,远端文件名:%s", wavepath.c_str());
}
//上传文件
@@ -271,7 +271,7 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
if (res != CURLE_OK) {
const char* em = errbuf[0] ? errbuf : curl_easy_strerror(res);
std::cerr << "http web failed: " << em << std::endl;
DIY_ERRORLOG("process","【ERROR】前置上传暂态录波文件 %s 失败，请检查文件上传接口配置",localpath.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG, "前置上传文件 %s 失败，请检查文件上传接口配置",localpath.c_str());
} else {
std::cout << "http web success, response: " << resPost0 << std::endl;
handleUploadResponse(resPost0, wavepath); // 处理响应
@@ -389,7 +389,7 @@ void download_xml_for_icd(const std::string& MODEL_ID,
std::cout << "remote file name:" << remote_file_name << "local save name:" << save_name << std::endl;
// mq日志
DIY_WARNLOG("process","【WARN】前置获取到终端类型%s,该终端类型对应的映射文件为%s,映射文件将下载并保存在本地为%s",TMNL_TYPE.c_str(),FILE_PATH.c_str(),save_name.c_str());
std::string fileContent;
std::string fullPath = std::string("filePath=") + filepath; //填写远端路径作为入参
@@ -409,14 +409,14 @@ void download_xml_for_icd(const std::string& MODEL_ID,
outFile.close();
std::cout << "File saved successfully!" << std::endl;
DIY_WARNLOG("process","【WARN】前置下载映射文件%s成功",save_name.c_str());
} else {
std::cerr << "Error: Unable to open file for writing." << std::endl;
DIY_ERRORLOG("process","【ERROR】前置写入本地映射文件%s失败",save_name.c_str());
}
} else {
std::cerr << "Error: Unable to download file." << std::endl;
DIY_ERRORLOG("process","【ERROR】前置调用文件下载接口下载远端文件文件%s失败",FILE_PATH.c_str());
}
}
@@ -550,7 +550,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
{
if (inputstring.empty()) {
std::cerr << "Error: inputstring is empty\n";
DIY_ERRORLOG("process","【ERROR】前置的%d号进程调用web台账接口的入参为空", g_front_seg_index);
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG,"台账接口的入参为空");
return 1;
}
@@ -569,7 +569,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
break;
}
std::cerr << "data 无效或为空数组，重试\n";
DIY_ERRORLOG("process","【ERROR】前置从web接口中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s",inputparm.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"前置从web接口中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s",inputparm.c_str());
} catch (const nlohmann::json::parse_error& e) {
std::cerr << "parse error: " << e.what() << ", retrying...\n";
}
@@ -586,7 +586,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
if (json_data.contains("data") && json_data["data"].is_array() && !json_data["data"].empty()) {
break;
}
DIY_ERRORLOG("process", "【ERROR】前置从本地台账中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s",inputparm.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "前置从本地台账中获取的台账信息为空或者无效信息无法解析,请核对前置使用的入参信息:%s", inputparm.c_str());
} catch (const nlohmann::json::parse_error& e) {
std::cerr << "local parse error: " << e.what() << "\n";
}
@@ -709,7 +709,7 @@ int parse_device_cfg_web()
input_jstr += "}";
std::cout << "input_jstr: " << input_jstr << std::endl;
DIY_DEBUGLOG("process","【DEBUG】前置的%d号进程调用web接口获取台账使用的请求输入为:%s", g_front_seg_index, input_jstr.c_str());
DIY_DEBUGLOG("process","前置获取台账使用的请求输入为:%s", input_jstr.c_str());
// 2. 调用接口
std::map<std::string, terminal_dev> terminal_dev_map;
@@ -738,10 +738,10 @@ int parse_device_cfg_web()
if (max_process_num != max_index) {
if (max_process_num >= 1 && max_process_num <= 9) {
DIY_WARNLOG("process", "【WARN】前置比对台账获取的进程数:%d和本地配置的进程数:%d,不匹配,按照台账进程数重置前置的进程数量",max_process_num, max_index);
DIY_WARNLOG_CODE("process",0,LOG_CODE_CONFIG, "前置比对台账获取的进程数:%d和本地配置的进程数:%d,不匹配,按照台账进程数重置前置的进程数量",max_process_num, max_index);
execute_bash("reset", max_process_num, "all");
} else {
DIY_ERRORLOG("process","【ERROR】前置从台账获取的进程数:%d不符合范围1~9,按照本地配置进程数启动进程",max_process_num);
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG, "前置从台账获取的进程数:%d不符合范围1~9,按照本地配置进程数启动进程",max_process_num);
}
}
}
@@ -749,13 +749,13 @@ int parse_device_cfg_web()
// 5. 台账数量与配置比对
int count_cfg = static_cast<int>(terminal_dev_map.size());
std::cout << "terminal_ledger_num: " << count_cfg << std::endl;
DIY_DEBUGLOG("process", "【DEBUG】前置的%d号进程调用获取到的台账的数量为:%d", g_front_seg_index, count_cfg);
DIY_DEBUGLOG("process", "前置获取到的台账的数量为:%d",count_cfg);
if (IED_COUNT < count_cfg) {
std::cout << "!!!!!!!!!!single process has ledger count more than config!!!!!!!" << std::endl;
//DIY_WARNLOG("process","【WARN】前置的%d号进程获取到的台账的数量大于配置文件中给单个进程配置的台账数量:%d,这个进程将按照获取到的台账的数量来创建台账空间，这个进程不能直接通过台账添加来新增台账，只能通过重启进程或者先删除已有台账再添加台账的方式来添加新台账", g_front_seg_index, IED_COUNT);
} else {
//DIY_INFOLOG("process","【NORMAL】前置的%d号进程根据配置文件中给单个进程配置的台账数量:%d来创建台账空间", g_front_seg_index, IED_COUNT);
}
///////////////////////////////////////////////////////////////////////////////用例这里将局部的map拷贝到全局map，后续根据协议台账修改
@@ -807,7 +807,7 @@ int parse_device_cfg_web()
if (!mon.terminal_connect.empty() && mon.terminal_connect != "0") {
isdelta_flag = 1;
std::cout << "monitor_id " << mon.monitor_id<< " v_wiring_type: " << mon.terminal_connect << " is delta wiring: " << isdelta_flag << std::endl;
DIY_WARNLOG("process","【WARN】前置连接的监测点%s是角形接线,对应终端为%s 终端类型是%s",mon.monitor_id.c_str(),dev.terminal_id.c_str(),dev.dev_type.c_str());
DIY_WARNLOG_CODE("process",0,LOG_CODE_WIRETYPE,"装置：%s - 监测点: %s 是角形接线",dev.terminal_name.c_str(),mon.monitor_name.c_str());
}
}
}
@@ -938,7 +938,7 @@ int parse_model_cfg_web()
// 3. 调用接口
std::map<std::string, icd_model*> icd_model_map;
if (parse_model_web(&icd_model_map, input_jstr)) {
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 icd模型接口异常,将使用默认的icd模型,请检查接口配置", g_front_seg_index);
// 确保释放 map
for (auto& kv : icd_model_map) delete kv.second;
return 0;
@@ -992,7 +992,7 @@ std::string parse_model_cfg_web_one(const std::string& terminal_type)
// 2. 拉取并解析
if (parse_model_web(&icd_model_map, input_jstr) != 0) {
std::cerr << "parse_model_web failed for type: " << terminal_type << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程 icd模型接口异常,将使用默认的icd模型,请检查接口配置", g_front_seg_index);
// 清理（即使 map 为空，也安全）
for (auto& kv : icd_model_map) delete kv.second;
return "";
@@ -1054,7 +1054,7 @@ static void writeJsonToFile(const std::string& filePath, const std::string& json
{
FILE* fp = fopen(filePath.c_str(), "w");
if (!fp) {
DIY_ERRORLOG("process", "【ERROR】无法将暂态事件写入本地缓存");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_TRANSIENT_COMM, "前置无法将暂态事件写入本地缓存");
std::cerr << "Failed to write in file : " << filePath << std::endl;
return;
}
@@ -1166,7 +1166,7 @@ static void scanAndResendOfflineFiles(const std::string& dirPath)
// 读取 JSON 文件内容
std::ifstream inFile(file.fileName.c_str());
if (!inFile) {
DIY_ERRORLOG("process", "【ERROR】无法打开本地缓存的暂态事件");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_TRANSIENT_COMM, "无法打开本地缓存的暂态事件");
std::cerr << "fail to open existing file: " << file.fileName << std::endl;
continue;
}
@@ -1185,14 +1185,14 @@ static void scanAndResendOfflineFiles(const std::string& dirPath)
try {
json j_r = json::parse(response);
DIY_WARNLOG("process", "【WARN】前置重发暂态事件成功");
DIY_WARNLOG_CODE("process",0,LOG_CODE_TRANSIENT_COMM, "前置重发暂态事件成功");
std::cout << "old file send success, remove it" << std::endl;
std::remove(file.fileName.c_str());
} catch (...) {
std::cout << "old file send fail (response parse failed)" << std::endl;
DIY_WARNLOG("process", "【WARN】前置重发暂态事件失败");
DIY_ERRORLOG_CODE("process",0,LOG_CODE_TRANSIENT_COMM, "前置重发暂态事件失败");
handleCommentResponse(response); // 仍然处理文本响应
}
} else {
@@ -1230,12 +1230,21 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
//找监测点id
std::string mpid;
//找测点名称
std::string mpname;
//找装置名称
std::string devname;
get_monitor_id_by_dev_and_seq(dev_id, monitor_id, mpid);
if(mpid.empty()) {
std::cout << "qvvr send error ,monitorId is null" << std::endl;
return 1;
}
get_terminal_name_by_terminal_id(dev_id, devname);
get_monitor_name_by_monitor_id(mpid, mpname);
root["monitorId"] = mpid;
root["devId"] = dev_id;
root["CpuNo"] = monitor_id;
@@ -1279,7 +1288,7 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
// 有效响应，略过
} catch (...) {
// 响应异常，保存 json
DIY_ERRORLOG(mpid.c_str(), "【ERROR】暂态接口响应异常,无法上送装置%s监测点%u的暂态事件",dev_id.c_str(), (unsigned)monitor_id);
DIY_ERRORLOG_CODE(mpid.c_str(),2,LOG_CODE_TRANSIENT_COMM, "暂态接口响应异常,无法上送装置%s - 监测点%s的暂态事件",devname.c_str(), mpname.c_str());
std::cout << "qvvr send fail ,store in local" << std::endl;
std::string qvvrDir = FRONT_PATH + "/dat/qvvr/";
@@ -1289,7 +1298,7 @@ int transfer_json_qvvr_data(const std::string& dev_id, ushort monitor_id,
}
} else {
// 无响应，保存 json
DIY_ERRORLOG(mpid.c_str(), "【ERROR】暂态接口无响应,无法上送装置%s监测点%u的暂态事件",dev_id.c_str(), (unsigned)monitor_id);
DIY_ERRORLOG_CODE(mpid.c_str(),2,LOG_CODE_TRANSIENT_COMM,"暂态接口无响应,无法上送装置%s - 监测点%s的暂态事件",devname.c_str(), mpname.c_str());
std::cout << "qvvr send fail ,store in local" << std::endl;
std::string qvvrDir = FRONT_PATH + "/dat/qvvr/";
@@ -1385,5 +1394,33 @@ bool get_monitor_id_by_dev_and_seq(const std::string& terminal_id,
return false;
}
// 根据 monitor_id 查 monitor_name
bool get_monitor_name_by_monitor_id(const std::string& monitor_id,
std::string& out_monitor_name)
{
std::lock_guard<std::mutex> lk(ledgermtx);
for (const auto& dev : terminal_devlist) {
for (const auto& mon : dev.line) {
if (mon.monitor_id == monitor_id) {
out_monitor_name = mon.monitor_name;
return true;
}
}
}
return false;
}
bool get_terminal_name_by_terminal_id(const std::string& terminal_id,
std::string& out_terminal_name)
{
std::lock_guard<std::mutex> lk(ledgermtx);
for (const auto& dev : terminal_devlist) {
if (dev.terminal_id == terminal_id) {
out_terminal_name = dev.terminal_name;
return true;
}
}
return false;
}

5
LFtid1056/cloudfront/code/interface.h
View File

@@ -729,6 +729,11 @@ void on_device_response_minimal(int response_code,
bool get_monitor_id_by_dev_and_seq(const std::string& terminal_id,
unsigned short logical_seq,
std::string& out_monitor_id);
bool get_monitor_name_by_monitor_id(const std::string& monitor_id,
std::string& out_monitor_name);
bool get_terminal_name_by_terminal_id(const std::string& terminal_id,
std::string& out_terminal_name);
//处理补招的任务
void check_recall_event();

155
LFtid1056/cloudfront/code/log4.cpp
View File

@@ -125,14 +125,13 @@ bool DebugSwitch::match(const std::string& logger_name, int level, int logtype)
std::map<std::string, TypedLogger> logger_map;
DebugSwitch g_debug_switch;
class SendAppender : public Appender {
/*class SendAppender : public Appender {
protected:
void append(const spi::InternalLoggingEvent& event) {
std::string logger_name = event.getLoggerName();
int level = event.getLogLevel();
std::string msg = event.getMessage();
int logtype = (logger_name.find(".COM") != std::string::npos) ? LOGTYPE_COM : LOGTYPE_DATA;
std::string level_str;
if (logger_name.find("process") == 0)
level_str = "process";
@@ -193,15 +192,139 @@ public:
virtual ~SendAppender() {
destructorImpl(); // 重要！释放 log4cplus 基类资源
}
};*/
class SendAppender : public Appender {
private:
struct RateState {
uint64_t hit_count = 0; // 同一条日志累计命中次数
std::chrono::steady_clock::time_point last_emit =
std::chrono::steady_clock::time_point::min();
};
static std::unordered_map<std::string, RateState> s_rate_map;
static std::mutex s_rate_mutex;
// 定义“同一条日志”的规则：logger + level + code + msg //原来只区分了日志登记名和等级，现在具体到每一条日志
static std::string make_key(const std::string& logger_name, int level, int code, const std::string& msg) {
std::ostringstream oss;
oss << logger_name << "|" << level << "|" << code << "|" << msg;
return oss.str();
}
// 前 5 次：1 秒一次；第 6 次起：300 秒一次
static bool should_emit(const std::string& key) {
using namespace std::chrono;
const auto now = steady_clock::now();
std::lock_guard<std::mutex> lk(s_rate_mutex);
RateState& st = s_rate_map[key];
st.hit_count++;
const int period_sec = (st.hit_count > 5) ? 300 : 1;
if (st.last_emit == steady_clock::time_point::min()) {
st.last_emit = now;
return true;
}
const auto elapsed = duration_cast<seconds>(now - st.last_emit).count();
if (elapsed >= period_sec) {
st.last_emit = now;
return true;
}
return false;
}
protected:
void append(const spi::InternalLoggingEvent& event) override {
std::string logger_name = event.getLoggerName();
int level = event.getLogLevel();
std::string msg = event.getMessage();
std::string level_str;
if (logger_name.find("process") == 0)
level_str = "process";
else if (logger_name.find("monitor") != std::string::npos)
level_str = "measurepoint";
else
level_str = "terminal";
// TLS code
int code = g_log_code_tls;
const int safe_logtype = 0;
if (!(level == ERROR_LOG_LEVEL ||
level == WARN_LOG_LEVEL ||
g_debug_switch.match(logger_name, level, safe_logtype))) {
return;
}
// ★新增：限频判断（同一条日志前 5 次 1 秒一次；之后 300 秒一次）
const std::string key = make_key(logger_name, level, code, msg);
if (!should_emit(key)) {
return;
}
std::ostringstream oss;
oss << "{\"processNo\":\"" << std::to_string(g_front_seg_index)
<< "\",\"nodeId\":\"" << FRONT_INST
<< "\",\"businessId\":\"" << extract_logger_id(logger_name)
<< "\",\"level\":\"" << level_str
<< "\",\"time\":\"" << now_yyyy_mm_dd_hh_mm_ss()
<< "\",\"grade\":\"" << get_level_str(level)
// ★建议：code 用数字（不是字符串）
<< "\",\"code\":" << code
<< ",\"log\":\"" << escape_json(msg) << "\"}";
queue_data_t connect_info;
connect_info.strTopic = G_LOG_TOPIC;
connect_info.strText = oss.str();
connect_info.tag = G_LOG_TAG;
connect_info.key = G_LOG_KEY;
std::lock_guard<std::mutex> lock(queue_data_list_mutex);
queue_data_list.push_back(connect_info);
}
std::string escape_json(const std::string& input) {
std::ostringstream ss;
for (size_t i = 0; i < input.size(); ++i) {
switch (input[i]) {
case '\\': ss << "\\\\"; break;
case '"': ss << "\\\""; break;
case '\n': ss << "\\n"; break;
case '\r': ss << "\\r"; break;
case '\t': ss << "\\t"; break;
default: ss << input[i]; break;
}
}
return ss.str();
}
void close() override {
// 可空实现
}
public:
SendAppender() {}
virtual ~SendAppender() {
destructorImpl();
}
};
//用来控制日志上送的静态变量定义
std::unordered_map<std::string, SendAppender::RateState> SendAppender::s_rate_map;
std::mutex SendAppender::s_rate_mutex;
//用来控制日志上送的结构
struct LOGEntry {
std::string id;
std::string level; // terminal / measurepoint
int logtype; // com / data
int min_grade;
int countdown;
std::string id; //测点和装置需要的id
std::string level; // terminal / measurepoint /process
int min_grade; // DEBUG / INFO / WARN / ERROR
int countdown; // 倒计时，单位秒
};
//日志上送map管理
@@ -209,8 +332,8 @@ std::map<std::string, LOGEntry> g_log_entries;
pthread_mutex_t g_log_mutex = PTHREAD_MUTEX_INITIALIZER;
// 生成唯一 key
std::string build_debug_key(const std::string& id, const std::string& level, int logtype) {
return id + "|" + level + "|" + (logtype == 1 ? "COM" : "DATA");
std::string build_debug_key(const std::string& id, const std::string& level) {
return id + "|" + level + "|";
}
// 外部线程中调用：每秒更新所有倒计时，0 则删除
@@ -234,17 +357,16 @@ void update_log_entries_countdown() {
void process_log_command(const std::string& id, const std::string& level, const std::string& grade, const std::string& logtype_str) {
if (level != "terminal" && level != "measurepoint") return;
int type = (logtype_str == "com") ? LOGTYPE_COM : LOGTYPE_DATA;
int grade_level = (grade == "DEBUG") ? DEBUG_LOG_LEVEL : INFO_LOG_LEVEL;
std::string key = build_debug_key(id, level, type);
std::string key = build_debug_key(id, level);
pthread_mutex_lock(&g_log_mutex);
LOGEntry& entry = g_log_entries[key]; // 会自动 insert 或取已有
entry.id = id;
entry.level = level;
entry.logtype = type;
entry.min_grade = grade_level;
entry.countdown = 60; // 重置倒计时
@@ -324,7 +446,7 @@ void init_loggers_bydevid(const std::string& dev_id)
Logger device_logger = init_logger(device_key, device_dir, dev_id, device_appender);
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
//DIY_WARNLOG(dev_id.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
}
// 初始化监测点日志，monitor.<mp_id>.COM / .DATA
@@ -349,7 +471,7 @@ void init_loggers_bydevid(const std::string& dev_id)
Logger mon_logger = init_logger(mon_key.str(), mon_path.str(), mon_name.str(), monitor_appender);
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
//DIY_WARNLOG(monitor.monitor_id.c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕", monitor.monitor_name.c_str(), monitor.logical_device_seq.c_str());
}
}
}
@@ -389,7 +511,7 @@ void init_loggers()
logger_map[device_key] = TypedLogger(device_logger, LOGTYPE_DATA);
//DIY_WARNLOG(term.terminal_id.c_str(), "【WARN】终端id:%s终端级日志初始化完毕", term.terminal_id.c_str());
// 初始化监测点日志
for (size_t i = 0; i < term.line.size(); ++i) {
@@ -412,8 +534,7 @@ void init_loggers()
logger_map[mon_key.str()] = TypedLogger(mon_logger, LOGTYPE_DATA);
//DIY_WARNLOG(mon_key.str().c_str(), "【WARN】监测点:%s - id:%s监测点级日志初始化完毕",
//monitor.monitor_name.c_str(), monitor.logical_device_seq.c_str());
}
}
}

9
LFtid1056/cloudfront/code/log4.h
View File

@@ -32,8 +32,6 @@ extern LOG_TLS_SPEC int g_log_code_tls;
#include "appender.h"
#define LOGTYPE_COM 1
#define LOGTYPE_DATA 2
/////////////////////////////////////////////入参验证
#if defined(__GNUC__) || defined(__clang__)
# define PRINTF_LIKE(fmt_index, first_arg) __attribute__((format(printf, fmt_index, first_arg)))
@@ -112,10 +110,13 @@ void format_log_msg(char* buf, size_t buf_size, const char* fmt, ...) PRINTF_LIK
// ====================== ★新增结束 ======================
typedef enum LogCode {
LOG_CODE_WIRETYPE = 96, /* 接线类型 */
LOG_CODE_CONFIG = 97, /* 配置相关 */
LOG_CODE_JSON = 98, /* JSON结构 */
LOG_CODE_OTHER = 99, /* 其他类型 */
LOG_CODE_LEDGER = 100, /* 台账类型 */
LOG_CODE_RPTINIT = 101, /* 报告初始化 */
LOG_CODE_ICD_AND_DOWNLOAD = 200, /* ICD 和文件下载类型 */
LOG_CODE_ICD_AND_DOWNLOAD = 200, /* ICD 类型 */
LOG_CODE_TRANSIENT = 300, /* 暂态发生 */
LOG_CODE_TRANSIENT_COMM = 301, /* 暂态接口 */
LOG_CODE_COMTRADE_FILE = 302, /* 录波文件（Comtrade） */
@@ -183,7 +184,7 @@ typedef enum LogCode {
#define DIY_DEBUGLOG(KEY, ...) DIY_LOG_CODE(log_debug, KEY, 0, LOG_CODE_OTHER,__VA_ARGS__) // ★修改：默认 code=0
// ★新增：显式传入 code 的便捷宏
// 用法示例：DIY_WARNLOG_CODE(full_key_m_c, warn_recallstart, "【WARN】监测点:%s ...", ...);
// 用法示例：DIY_WARNLOG_CODE(full_key_m_c, warn_recallstart, "监测点:%s ...", ...);
#define DIY_ERRORLOG_CODE(KEY, LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_error, KEY, LEVEL_INT, CODE_INT, __VA_ARGS__) // ★新增
#define DIY_WARNLOG_CODE(KEY, LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_warn, KEY, LEVEL_INT, CODE_INT, __VA_ARGS__) // ★新增
#define DIY_INFOLOG_CODE(KEY, LEVEL_INT,CODE_INT, ...) DIY_LOG_CODE(log_info, KEY, LEVEL_INT, CODE_INT, __VA_ARGS__) // ★新增

4
LFtid1056/cloudfront/code/main.cpp
View File

@@ -209,7 +209,7 @@ std::string get_parent_directory() {
//启动进程日志
init_logger_process();
DIY_WARNLOG("process","【WARN】前置的%d号进程 进程级日志初始化完毕", g_front_seg_index);
//读取台账
parse_device_cfg_web();
@@ -497,7 +497,7 @@ void Front::OnTimerThread()
while (!m_IsTimerCancel)
{
update_log_entries_countdown();
update_log_entries_countdown();//日志上送倒计时
//业务超时检查
check_device_busy_timeout();

55
LFtid1056/cloudfront/code/rocketmq.cpp
View File

@@ -219,11 +219,14 @@ void RocketMQProducer::sendMessage(const std::string& body,
<< std::endl;
} catch (const MQClientException& e) {
std::cerr << "[RocketMQProducer] 发送失败: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
// 根据需要进行重试或日志记录
} catch (const std::exception& e) {
std::cerr << "[RocketMQProducer] 异常: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
} catch (...) {
std::cerr << "[RocketMQProducer] 未知错误，消息发送失败。" << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
}
}
@@ -244,10 +247,13 @@ void RocketMQProducer::sendMessageOrderly(const std::string& body,
<< std::endl;
} catch (const MQClientException& e) {
std::cerr << "[RocketMQProducer] 顺序发送失败: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
} catch (const std::exception& e) {
std::cerr << "[RocketMQProducer] 异常: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
} catch (...) {
std::cerr << "[RocketMQProducer] 未知错误，顺序消息发送失败。" << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
}
}
@@ -304,7 +310,7 @@ void rocketmq_producer_send(rocketmq::RocketMQProducer* producer,
producer->sendMessage(body, topic, tags, keys);
} catch (const std::exception& e) {
std::cerr << "[rocketmq_producer_send] 发送失败: " << e.what() << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程 MQ发送失败", g_front_seg_index);
DIY_ERRORLOG_CODE("process",0,LOG_CODE_MQ, "MQ发送失败");
}
}
@@ -514,8 +520,8 @@ 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", "收到主题:%s - tag:%s的进程控制消息",
G_MQCONSUMER_TOPIC_SET.c_str(),FRONT_INST.c_str());
if (code_str == "set_process") {
@@ -547,8 +553,8 @@ 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_CODE("process",0,LOG_CODE_PROCESS_CONTROL, "执行指令:%s,reset表示重启所有进程,add表示添加进程",
fun.c_str());
send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到重置进程指令,重启所有进程!");
std::cout << "this msg should only execute once" << std::endl;
@@ -564,8 +570,7 @@ bool parseJsonMessageSET(const std::string& json_str) {
// 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());
DIY_WARNLOG_CODE("process",0,LOG_CODE_PROCESS_CONTROL, "执行指令:%s,即将重启", fun.c_str());
std::this_thread::sleep_for(std::chrono::seconds(3));
::_exit(-1039); // 进程退出
@@ -653,13 +658,14 @@ bool parseJsonMessageLOG(const std::string& json_str) {
return true;
}*/
DIY_INFOLOG("process", "【NORMAL】前置的%d号进程处理日志上送消息", g_front_seg_index);
DIY_INFOLOG("process", "收到主题:%s - tag:%s的日志控制消息",
G_MQCONSUMER_TOPIC_LOG.c_str(),FRONT_INST.c_str());
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_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到实时日志指令");
send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到日志控制指令");
if (code_str == "set_log") {
// 校验数据合法性
@@ -673,7 +679,7 @@ bool parseJsonMessageLOG(const std::string& json_str) {
process_log_command(id, level, grade, logtype);
} else {
std::cout << "type doesn't match" << std::endl;
DIY_WARNLOG("process", "【WARN】前置的%d号进程处理日志上送消息,格式不正确", g_front_seg_index);
DIY_WARNLOG_CODE("process",0,LOG_CODE_JSON, "日志控制指令失败");
}
std::cout << "this msg should only execute once" << std::endl;
@@ -727,8 +733,8 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d
std::cout << "msg index: " << process_No << " 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_UD.c_str());
DIY_INFOLOG("process", "收到主题:%s - tag:%s的台账更新消息",
G_MQCONSUMER_TOPIC_UD.c_str(), FRONT_INST.c_str());
//send_reply_to_queue(guid, static_cast<int>(ResponseCode::ACCEPTED), "收到台账更新指令");
std::vector<DeviceReply> reply_list;
@@ -881,7 +887,7 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d
/*send_reply_to_queue(json_data.guid, static_cast<int>(ResponseCode::NOT_FOUND),
"终端 id: " + tid + " 无法修改台账，未找到指定装置,改为添加这个装置");*/
DIY_WARNLOG("process", "【WARN】无法修改台账，未找到指定装置: %s ,改为添加这个装置", tid.c_str());
DIY_WARNLOG_CODE("process",0,LOG_CODE_LEDGER_UPDATE,"无法修改台账，未找到指定装置: %s ,改为添加这个装置的台账", json_data.terminal_name.c_str());
init_loggers_bydevid(json_data.terminal_id);
terminal_devlist.push_back(json_data);
@@ -1019,7 +1025,7 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
}
// 日志记录
DIY_INFOLOG("process", "【NORMAL】前置消费topic:%s_%s的实时触发消息",FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
DIY_INFOLOG("process", "收到主题:%s - tag:%s的实时触发消息",G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.c_str());
std::cout << "rtdata Callback received message: " << body << std::endl;
if (!key.empty()) {
@@ -1030,14 +1036,17 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
// 消息解析
std::string devid;
std::string devname;
ushort line;
bool realData = false, soeData = false;
int limit = 0;
int idx = 0;
get_terminal_name_by_terminal_id(devid, devname);
if (!parseJsonMessageRT(body, devid, line, realData, soeData, limit,idx)) {
std::cerr << "Failed to parse the JSON message." << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置消费topic:%s_%s的实时触发消息失败,消息的json格式不正确", FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "主题:%s - tag:%s的实时触发消息失败", G_MQCONSUMER_TOPIC_RT.c_str(), FRONT_INST.c_str());
return rocketmq::RECONSUME_LATER;
}
@@ -1051,7 +1060,7 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
if (ClientManager::instance().get_dev_status(devid) != 1) {
std::cout << "devid对应装置不在线: " << devid << std::endl;
// 记录日志不响应 web 端
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的实时数据触发消息失败,装置%s不在线", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str(),devid.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_COMM,"主题:%s - tag:%s的实时数据触发消息失败,装置%s不在线", G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.c_str(),devname.c_str());
return rocketmq::CONSUME_SUCCESS;
}
@@ -1062,7 +1071,7 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
}
else{
std::cerr << "rtdata is NULL." << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,消息的json结构不正确", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的补招触发消息失败",G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.c_str());
return rocketmq::RECONSUME_LATER;
}
@@ -1108,7 +1117,7 @@ rocketmq::ConsumeStatus myMessageCallbackupdate(const rocketmq::MQMessageExt& ms
// 调用业务逻辑处理函数
std::string updatefilepath = FRONT_PATH + "/etc/ledgerupdate";
if (!parseJsonMessageUD(body, updatefilepath)) {
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的台账更新消息失败,消息的json结构不正确", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_UD.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的台账更新消息失败",G_MQCONSUMER_TOPIC_UD.c_str(),FRONT_INST.c_str());
}
return rocketmq::CONSUME_SUCCESS;
@@ -1152,7 +1161,7 @@ rocketmq::ConsumeStatus myMessageCallbackset(const rocketmq::MQMessageExt& msg)
// 调用业务处理逻辑
if (!parseJsonMessageSET(body)) {
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s - tag:%s的进程控制消息失败,消息的json结构不正确", g_front_seg_index, G_MQCONSUMER_TOPIC_SET.c_str(), FRONT_INST.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的进程控制消息失败", G_MQCONSUMER_TOPIC_SET.c_str(), FRONT_INST.c_str());
}
return rocketmq::CONSUME_SUCCESS;
@@ -1196,7 +1205,7 @@ rocketmq::ConsumeStatus myMessageCallbacklog(const rocketmq::MQMessageExt& msg)
// 执行日志上送处理
if (!parseJsonMessageLOG(body)) {
DIY_ERRORLOG("process", "【ERROR】前置的%d号进程处理topic:%s_%s的日志上送消息失败,消息的json结构不正确", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_LOG.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的日志上送消息失败",G_MQCONSUMER_TOPIC_LOG.c_str(),FRONT_INST.c_str());
}
return rocketmq::CONSUME_SUCCESS;
@@ -2255,7 +2264,7 @@ rocketmq::ConsumeStatus cloudMessageCallback(const rocketmq::MQMessageExt& msg)
}
// 日志记录
DIY_INFOLOG("process", "【NORMAL】前置消费topic:%s_%s的云前置控制消息",FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_CLOUD.c_str());
DIY_INFOLOG("process", "收到主题:%s - tag:%s的云前置控制消息",G_MQCONSUMER_TOPIC_CLOUD.c_str(),FRONT_INST.c_str());
std::cout << "cloud Callback received message: " << body << std::endl;
if (!key.empty()) {
@@ -2273,7 +2282,7 @@ rocketmq::ConsumeStatus cloudMessageCallback(const rocketmq::MQMessageExt& msg)
if (!parseJsonMessageCLOUD(body, devid, guid, DetailObj,FrontId,Node)) {
std::cerr << "Failed to parse the JSON message." << std::endl;
DIY_ERRORLOG("process", "【ERROR】前置消费topic:%s_%s的云前置控制消息失败,消息的json格式不正确", FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "主题:%s - tag:%s的云前置控制消息失败", G_MQCONSUMER_TOPIC_CLOUD.c_str(), FRONT_INST.c_str());
return rocketmq::RECONSUME_LATER;
}
@@ -2292,7 +2301,7 @@ rocketmq::ConsumeStatus cloudMessageCallback(const rocketmq::MQMessageExt& msg)
if(!parsemsg(devid,guid,DetailObj)){
std::cerr << "clouddata is error." << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的云前置控制消息失败,消息无法解析", g_front_seg_index,FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RT.c_str());
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON,"主题:%s - tag:%s的云前置控制消息失败",G_MQCONSUMER_TOPIC_CLOUD.c_str(),FRONT_INST.c_str());
}
return rocketmq::CONSUME_SUCCESS;

4
LFtid1056/cloudfront/code/worker.cpp
View File

@@ -319,11 +319,11 @@ extern bool normalOutputEnabled;
showinshellflag = true;
handleViewLogCommand(cmd, clientFD);
} else if (cmd.find("G_TEST_NUM=") == 0) {
int num = std::atoi(cmd.substr(9).c_str());
int num = std::atoi(cmd.substr(11).c_str());
setTestNum(num);
sendStr(clientFD, "\r\x1B[KTEST_NUM updated\r\n");
} else if (cmd.find("G_TEST_TYPE=") == 0) {
int type = std::atoi(cmd.substr(10).c_str());
int type = std::atoi(cmd.substr(12).c_str());
setTestType(type);
sendStr(clientFD, "\r\x1B[KTEST_TYPE updated\r\n");
} else if (cmd.find("LOG=") == 0) {

5
LFtid1056/dealMsg.cpp
View File

@@ -124,12 +124,11 @@ void process_received_message(string mac, string id,const char* data, size_t len
//end_time.tm_min = 1;
//end_time.tm_sec = 1;
//ClientManager::instance().read_eventlog_action_to_device(id, start_time, end_time,2,1);
/*DIY_ERRORLOG_CODE("111", 0, static_cast<int>(LogCode::LOG_CODE_OTHER), "【ERROR】测试告警发送 前置");
DIY_ERRORLOG_CODE(id, 1, static_cast<int>(LogCode::LOG_CODE_OTHER), "【ERROR】测试告警发送 设备");
/*
std::string mpid;
get_monitor_id_by_dev_and_seq(id, 1, mpid);
if (!mpid.empty()) {
DIY_ERRORLOG_CODE(mpid, 2, static_cast<int>(LogCode::LOG_CODE_OTHER), "【ERROR】测试告警发送 测点");
}*/
}
if (udata[19] == 0x00) {

BIN
LFtid1056_20260106.rar Normal file
View File

Binary file not shown.