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

finish dz function

Browse Source
This commit is contained in:
lnk
2025-09-05 16:28:26 +08:00
parent 869189d18e
commit 9dc138a617
6 changed files with 431 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -429,7 +429,7 @@ std::string parseJsonMessageRC(const std::string& inputJson) {
}
}
bool parseJsonMessageRT(const std::string& body,std::string& devSeries,std::string& line,bool& realData,bool& soeData,int& limit){
bool parseJsonMessageRT(const std::string& body,std::string& devSeries,ushort& line,bool& realData,bool& soeData,int& limit){
json root;
try {
root = json::parse(body);
@@ -472,7 +472,7 @@ bool parseJsonMessageRT(const std::string& body,std::string& devSeries,std::stri
try {
devSeries = messageBody["devSeries"].get<std::string>();
line = messageBody["line"].get<std::string>();
line = messageBody["line"].get<ushort>();
realData = messageBody["realData"].get<bool>();
soeData = messageBody["soeData"].get<bool>();
limit = messageBody["limit"].get<int>();
@@ -859,7 +859,8 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
}
// 消息解析
std::string devid, line;
std::string devid;
ushort line;
bool realData = false, soeData = false;
int limit = 0;
@@ -870,33 +871,16 @@ rocketmq::ConsumeStatus myMessageCallbackrtdata(const rocketmq::MQMessageExt& ms
}
// 加锁访问台账
int dev_index;
int mp_index;
if( !devid.empty() && !line.empty()){
std::lock_guard<std::mutex> lock(ledgermtx);
dev_index = find_dev_index_from_dev_id(devid);
mp_index = find_mp_index_from_mp_id(line);
if( !devid.empty() && line > 0){
//不再使用文件触发方式，直接调用接口向终端发起请求
ClientManager::instance().set_real_state_count(devid, 60, line);//一秒询问一次，询问60次,下一次同一个测点调用的话就会刷新
}
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());
}
if (dev_index == -1 || mp_index == -1) {
std::cerr << "dev index or mp index is not found" << std::endl;
return rocketmq::RECONSUME_LATER;
}
// 写入 XML
/*if (!createXmlFile(dev_index, mp_index, realData, soeData, limit, "new")) {
DIY_ERRORLOG("process", "【ERROR】前置无法创建实时数据触发文件");
std::cerr << "Failed to create the XML file." << std::endl;
return rocketmq::RECONSUME_LATER;
}*/
//不再使用文件触发方式，直接调用接口向终端发起请求
ClientManager::instance().set_real_state_count(devid, 60,mp_index);//一秒询问一次，询问60次
return rocketmq::CONSUME_SUCCESS;
}