lnk/microser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: lnk:develop
Branches Tags
lnk:master lnk:develop lnk:redis lnk:zz lnk:arm_develop
lnk:mqC++ lnk:armtest1.0 lnk:v1.1.0 lnk:LN_1.0 lnk:DB_1.0 lnk:0731 lnk:v1.0.1
..
compare: lnk:64090e3f1ae0df972f22ca856fb741dd70d8a11f
Branches Tags
lnk:develop lnk:redis lnk:zz lnk:arm_develop lnk:master
lnk:mqC++ lnk:armtest1.0 lnk:v1.1.0 lnk:LN_1.0 lnk:DB_1.0 lnk:0731 lnk:v1.0.1

5 Commits
develop ... 64090e3f1a

Author SHA1 Message Date
lnk
64090e3f1a 最多只记录一次原始数据 2026-07-15 10:09:54 +08:00
lnk
a6e9934c88 数据追踪原始报告块 2026-07-15 10:09:54 +08:00
lnk
76e8fa683c 数据追踪增加接线方式和来源 2026-07-15 10:09:54 +08:00
lnk
d69c823575 完成redisstream前置生产和消费功能 2026-06-30 14:19:27 +08:00
lnk
ae911b46e5 redis引入未完成 2026-06-24 16:37:32 +08:00
19 changed files with 3602 additions and 833 deletions
Expand all files Collapse all files

137
cfg_parse/SimpleProducer.cpp
View File

@@ -43,6 +43,10 @@
#include "../log4cplus/log4.h"//lnk添加log4
//添加redis stream相关头文件
#include "../redisstream/RedisStreamMQ.h"
#include <QSettings>
using namespace std;
static std::once_flag g_consumer_once;
@@ -63,6 +67,11 @@ extern std::string G_MQCONSUMER_TOPIC_RT;
extern std::string G_MQCONSUMER_TOPIC_TEST;
//消费者连接秘钥
extern std::string G_MQCONSUMER_ACCESSKEY;
extern std::string G_MQCONSUMER_SECRETKEY;
extern std::string G_MQCONSUMER_CHANNEL;
extern std::string FRONT_INST;
extern bool DEBUGOPEN;
@@ -80,10 +89,9 @@ extern int g_front_seg_index;
extern char subdir[128];
////////////////////////////////////////////////////////////////////////////////////////////////////////////
//消费者连接秘钥
extern std::string G_MQCONSUMER_ACCESSKEY;
extern std::string G_MQCONSUMER_SECRETKEY;
extern std::string G_MQCONSUMER_CHANNEL;
extern int G_MQ_BACKEND;
////////////////////////////////////////////////////////////////////////////////////////////////////////
// 前向声明 RocketMQConsumer 类
class RocketMQConsumer;
@@ -102,11 +110,23 @@ int64_t G_APP_START_MS = []() -> int64_t {
int64_t G_START_SKEW_MS = 1000;
bool should_process_after_start(const rocketmq::MQMessageExt& msg)
bool RocketMQ_should_process_after_start(const rocketmq::MQMessageExt& msg)
{
const int64_t born_ts = static_cast<int64_t>(msg.getBornTimestamp());
return born_ts >= (G_APP_START_MS - G_START_SKEW_MS);
}
bool should_process_after_start(const rocketmq::MQMessageExt& msg)
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
return RedisStream_should_process_after_start(msg);
}
else
{
return RocketMQ_should_process_after_start(msg);
}
}
///////////////////////////////
class InternalListener;
@@ -419,7 +439,7 @@ void RocketMQConsumer::setConsumerMessageModel(const std::string& topic)
RocketMQConsumer* g_consumer = NULL;
// 初始化消费者函数
void InitializeConsumer(
void RocketMQ_InitializeConsumer(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions) // 接收多个订阅
@@ -446,7 +466,7 @@ void InitializeConsumer(
}
// 关闭并销毁消费者函数
void ShutdownAndDestroyConsumer()
void RocketMQ_ShutdownAndDestroyConsumer()
{
if (g_consumer != NULL) {
delete g_consumer;
@@ -455,7 +475,7 @@ void ShutdownAndDestroyConsumer()
}
// 消费消息的接口函数
void rocketmq_consumer_receive(
void RocketMQ_rocketmq_consumer_receive(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions) // 接收多个订阅
@@ -844,7 +864,7 @@ private:
RocketMQProducer* g_producer = NULL;
// 初始化生产者（在程序启动时调用一次）
void InitializeProducer()
void RocketMQ_InitializeProducer()
{
if (g_producer == NULL) {
try {
@@ -859,7 +879,7 @@ void InitializeProducer()
}
// 关闭并销毁生产者（在程序结束时调用一次）
void ShutdownAndDestroyProducer()
void RocketMQ_ShutdownAndDestroyProducer()
{
if (g_producer != NULL) {
delete g_producer;
@@ -868,7 +888,7 @@ void ShutdownAndDestroyProducer()
}
// 发送消息的接口函数
void rocketmq_producer_send(const std::string& body,
void RocketMQ_rocketmq_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys)
@@ -887,7 +907,102 @@ void rocketmq_producer_send(const std::string& body,
}
}
#endif
/////////////////////////////////////////////////////////////////////////////////////////////////////////生产者消费者切换
void InitializeProducer()
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_InitializeProducer();
return;
}
RocketMQ_InitializeProducer();
}
void ShutdownAndDestroyProducer()
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_ShutdownAndDestroyProducer();
return;
}
RocketMQ_ShutdownAndDestroyProducer();
}
void rocketmq_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys)
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_producer_send(
body,
topic,
tags,
keys
);
return;
}
RocketMQ_rocketmq_producer_send(
body,
topic,
tags,
keys
);
}
void InitializeConsumer(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions)
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_InitializeConsumer(
consumerName,
nameServer,
subscriptions
);
return;
}
RocketMQ_InitializeConsumer(
consumerName,
nameServer,
subscriptions
);
}
void ShutdownAndDestroyConsumer()
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_ShutdownAndDestroyConsumer();
return;
}
RocketMQ_ShutdownAndDestroyConsumer();
}
void rocketmq_consumer_receive(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions)
{
if (G_MQ_BACKEND == MQ_BACKEND_REDIS_STREAM)
{
RedisStream_consumer_receive(
consumerName,
nameServer,
subscriptions
);
return;
}
RocketMQ_rocketmq_consumer_receive(
consumerName,
nameServer,
subscriptions
);
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////
/*
// producer_send0测试用

130
cfg_parse/cfg_parser.cpp
View File

@@ -156,7 +156,6 @@ public:
char dev_type[64];
char dev_key[255];
char dev_series[255];
char com_type[32];
char addr_str[64];
char port[64];
char timestamp[64];
@@ -377,6 +376,9 @@ int G_LOG_RATE_KEEP_BURST_MS = 1000; // 间隔 >= 1000ms，按二级策略
int G_LOG_RATE_KEEP_BURST_COUNT = 60; // 二级保留前60条
int G_LOG_RATE_KEEP_HIGHFREQ_COUNT = 10; // 高频保留前10条
//添加底层队列开关
int G_MQ_BACKEND = 0; // 0-默认RocketMQ，1-Redis Stream
//lnk20250115添加台账锁
extern pthread_mutex_t mtx;
@@ -650,6 +652,20 @@ void init_config() {
qDebug() << "Read UDS_DOWNLOAD_URL:" << UDS_DOWNLOAD_URL << endl;
/*添加rocketmq的解析10-9 */////////////////////////////////////////////////////////////////
ba = settings.value("RocketMq/Backend", "rocketmq").toString().toLatin1();
std::string mq_backend = ba.data();
if (mq_backend == "redis" || mq_backend == "redisstream" || mq_backend == "1") {
G_MQ_BACKEND = MQ_BACKEND_REDIS_STREAM;
} else {
G_MQ_BACKEND = MQ_BACKEND_ROCKETMQ;
}
std::cout << "Read RocketMq/Backend:" << mq_backend
<< ", G_MQ_BACKEND:" << G_MQ_BACKEND
<< std::endl;
//生产者
ba = settings.value("RocketMq/producer", "").toString().toLatin1();
G_ROCKETMQ_PRODUCER = strdup(ba.data());
@@ -861,9 +877,6 @@ void init_config() {
else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
TEST_PORT = TEST_PORT + SOE_COMTRADE_BASE_NODE_ID + g_front_seg_index;
}
else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
TEST_PORT = TEST_PORT + PQDIF_DATA_BASE_NODE_ID + g_front_seg_index;
}
}
@@ -1704,8 +1717,6 @@ void add_terminal_to_trigger_update(trigger_update_xml_t* trigger_update_xml, co
}
}
static bool terminal_update_should_bind_to_current_process(const terminal* dev);
// 解析 XML 数据并提取 terminal 信息
void parse_terminal_from_data(trigger_update_xml_t* trigger_update_xml, const std::string& str_tag, const std::string& data,const std::string& guid_value) {
terminal work_terminal = {}; // 创建新的 terminal 对象
@@ -1721,13 +1732,6 @@ void parse_terminal_from_data(trigger_update_xml_t* trigger_update_xml, const st
strcpy(work_terminal.dev_type, extract_value(data, "devType").c_str());
strcpy(work_terminal.dev_key, extract_value(data, "devKey").c_str());
strcpy(work_terminal.dev_series, extract_value(data, "series").c_str());
std::string com_type = extract_value(data, "comType");
if (com_type.empty())
com_type = extract_value(data, "com_type");
if (com_type.empty())
com_type = "MMS";
strncpy(work_terminal.com_type, com_type.c_str(), sizeof(work_terminal.com_type) - 1);
work_terminal.com_type[sizeof(work_terminal.com_type) - 1] = '\0';
strcpy(work_terminal.processNo, extract_value(data, "processNo").c_str());
strcpy(work_terminal.addr_str, extract_value(data, "ip").c_str());
strcpy(work_terminal.port, extract_value(data, "port").c_str());
@@ -1814,13 +1818,6 @@ void parse_terminal_from_data(trigger_update_xml_t* trigger_update_xml, const st
// 根据 str_tag 将 terminal 添加到相应的数组
if (!terminal_update_should_bind_to_current_process(&work_terminal)) {
std::cout << "skip ledger update xml by comType:" << work_terminal.com_type
<< " terminal:" << work_terminal.terminal_id
<< " process:" << subdir << g_front_seg_index << std::endl;
return;
}
add_terminal_to_trigger_update(trigger_update_xml, str_tag, work_terminal);
}
@@ -1841,25 +1838,11 @@ void parse_ledger_update(trigger_update_xml_t* trigger_update_xml, const std::st
if (strTag == "delete") {
// 填充终端信息
strcpy(delete_terminal.terminal_id, extract_value(data, "id").c_str());
std::string com_type = extract_value(data, "comType");
if (com_type.empty())
com_type = extract_value(data, "com_type");
if (com_type.empty())
com_type = "MMS";
strncpy(delete_terminal.com_type, com_type.c_str(), sizeof(delete_terminal.com_type) - 1);
delete_terminal.com_type[sizeof(delete_terminal.com_type) - 1] = '\0';
//添加guid20250506
strncpy(delete_terminal.guid, guid_value.c_str(), sizeof(delete_terminal.guid) - 1);
delete_terminal.guid[sizeof(delete_terminal.guid) - 1] = '\0';
if (!terminal_update_should_bind_to_current_process(&delete_terminal)) {
std::cout << "skip delete ledger update xml by comType:" << delete_terminal.com_type
<< " terminal:" << delete_terminal.terminal_id
<< " process:" << subdir << g_front_seg_index << std::endl;
return;
}
// 将删除信息写入 delete_updates
std::cout << "delete ledger!!!!!"<<std::endl;
if (trigger_update_xml->delete_update_num < MAX_UPDATEA_NUM) {
@@ -3422,45 +3405,6 @@ void SendJsonAPI_web(const std::string& strUrl, const char* code, const std::str
}
// 打印 terminal_dev_map 中所有内容的函数
static bool is_cfg_pqdif_process()
{
return std::strstr(subdir, "cfg_pqdif_data") != NULL;
}
static bool is_pqdif_com_type(const char* com_type)
{
QString type = com_type ? QString::fromLocal8Bit(com_type).trimmed() : QString();
return type.compare("PQDIF", Qt::CaseInsensitive) == 0;
}
static bool terminal_should_bind_to_current_process(const terminal_dev* dev)
{
if (dev == NULL)
return false;
if (is_pqdif_com_type(dev->com_type))
return is_cfg_pqdif_process();
if (is_cfg_pqdif_process())
return false;
return (atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0);
}
static bool terminal_update_should_bind_to_current_process(const terminal* dev)
{
if (dev == NULL)
return false;
if (is_pqdif_com_type(dev->com_type))
return is_cfg_pqdif_process();
if (is_cfg_pqdif_process())
return false;
return true;
}
void printTerminalDevMap(const QMap<QString, terminal_dev*>& terminal_dev_map) {
QMap<QString, terminal_dev*>::const_iterator it;
for (it = terminal_dev_map.constBegin(); it != terminal_dev_map.constEnd(); ++it) {
@@ -3479,7 +3423,6 @@ void printTerminalDevMap(const QMap<QString, terminal_dev*>& terminal_dev_map) {
<< ", Device Type:" << QString(dev->dev_type)
<< ", Device Key:" << QString(dev->dev_key)
<< ", Device Series:" << QString(dev->dev_series)
<< ", Device comType:" << QString(dev->com_type)
<< ", Device processNo:" << QString(dev->processNo)
<< ", Device maxProcessNum:" << QString(dev->maxProcessNum)
<< ", Address:" << QString(dev->addr_str)
@@ -3514,7 +3457,6 @@ void printLedger(const ied_usr_t& ied_usr) {
std::cout << "|-- dev_type: " << ied_usr.dev_type << std::endl;
std::cout << "|-- dev_key: " << ied_usr.dev_key << std::endl;
std::cout << "|-- dev_series: " << ied_usr.dev_series << std::endl;
std::cout << "|-- com_type: " << ied_usr.com_type << std::endl;
std::cout << "|-- dev_flag: " << ied_usr.dev_flag << std::endl;
std::cout << "|-- last_call_wavelist_time: " << ied_usr.last_call_wavelist_time << std::endl;
@@ -3609,7 +3551,6 @@ void printLedgerinshell(const ied_usr_t& ied_usr, QIODevice* outputDevice) {
outputDevice->write("\r\x1B[K");outputDevice->write("|-- dev_type: " + QByteArray(ied_usr.dev_type) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- dev_key: " + QByteArray(ied_usr.dev_key) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- dev_series: " + QByteArray(ied_usr.dev_series) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- com_type: " + QByteArray(ied_usr.com_type) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- dev_processNo: " + QByteArray(ied_usr.processNo) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- dev_flag: " + QByteArray::number(ied_usr.dev_flag) + "\n");
outputDevice->write("\r\x1B[K");outputDevice->write("|-- last_call_wavelist_time: " + QByteArray::number(ied_usr.last_call_wavelist_time) + "\n");
@@ -4161,13 +4102,6 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
else strncpy(dev->dev_series, "N/A", sizeof(dev->dev_series) - 1);
dev->dev_series[sizeof(dev->dev_series) - 1] = '\0';
cJSON* comType = cJSON_GetObjectItem(item, "comType");
if (comType == NULL)
comType = cJSON_GetObjectItem(item, "com_type");
if (comType && comType->type == cJSON_String) strncpy(dev->com_type, comType->valuestring, sizeof(dev->com_type) - 1);
else strncpy(dev->com_type, "MMS", sizeof(dev->com_type) - 1);
dev->com_type[sizeof(dev->com_type) - 1] = '\0';
//lnk20250210台账进程号
cJSON* processNo = cJSON_GetObjectItem(item, "processNo"); // processNo转为字符串
if (processNo && processNo->type == cJSON_Number) snprintf(dev->processNo, sizeof(dev->processNo), "%d", processNo->valueint);
@@ -4273,20 +4207,19 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
//lnk20260312防止内存泄漏
bool inserted = false;
bool should_bind = terminal_should_bind_to_current_process(dev);
// 检查是否存在重复键
if (terminal_dev_map->contains(key)) {
std::cerr << "Duplicate terminal_code found: " << key.toStdString() << std::endl;
// 删除旧的 terminal_dev 对象以避免内存泄漏
if (should_bind) delete terminal_dev_map->value(key);
delete terminal_dev_map->value(key);
// 移除旧的键值对
if (should_bind) terminal_dev_map->remove(key);
terminal_dev_map->remove(key);
// 插入新的 terminal_dev 对象
if(should_bind){//lnk20250210匹配进程号
if(atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0){//lnk20250210匹配进程号
//调试用
std::cout<< "process num match" << std::endl;
terminal_dev_map->insert(key, dev);
@@ -4294,7 +4227,7 @@ int terminal_ledger_web(QMap<QString, terminal_dev*>* terminal_dev_map,
}//后续修改为只有进程号匹配上index才录入当前进程
} else {
// 插入新的 terminal_dev 对象
if(should_bind){//lnk20250210匹配进程号
if(atoi(dev->processNo) == g_front_seg_index || g_front_seg_index == 0){//lnk20250210匹配进程号
//调试用
std::cout<< "process num match" << std::endl;
terminal_dev_map->insert(key, dev);
@@ -4455,7 +4388,6 @@ int parse_device_cfg_web()
char dev_type[64];
char dev_key[255];
char dev_series[255];
char com_type[32];
char addr_str[64];
char port_char[64];
@@ -4485,8 +4417,6 @@ int parse_device_cfg_web()
strncpy(dev_type, value->dev_type, sizeof(dev_type) - 1);
strncpy(dev_key, value->dev_key, sizeof(dev_key) - 1);
strncpy(dev_series, value->dev_series, sizeof(dev_series) - 1);
strncpy(com_type, value->com_type, sizeof(com_type) - 1);
com_type[sizeof(com_type) - 1] = '\0';
strncpy(addr_str, value->addr_str, sizeof(addr_str) - 1);
strncpy(port_char, value->port, sizeof(port_char) - 1);
strncpy(processNo, value->processNo, sizeof(processNo) - 1);//进程号
@@ -4552,14 +4482,6 @@ int parse_device_cfg_web()
cout << "ied_usr->dev_type:" << ied_usr->dev_type << endl;
}
//lnk20250210台账进程号
if (com_type != NULL && com_type[0] != '\0') {
apr_snprintf(ied_usr->com_type, sizeof(ied_usr->com_type), "%s", com_type);//comType
cout << "ied_usr->com_type:" << ied_usr->com_type << endl;
}
else {
apr_snprintf(ied_usr->com_type, sizeof(ied_usr->com_type), "%s", "MMS");
cout << "ied_usr->com_type(default):" << ied_usr->com_type << endl;
}
if (processNo != NULL) {
apr_snprintf(ied_usr->processNo, sizeof(ied_usr->processNo), "%s", processNo);//processNo
cout << "ied_usr->processNo:" << ied_usr->processNo << endl;
@@ -5829,13 +5751,6 @@ int update_one_terminal_ledger(terminal* update, int i,ied_t* ied,int terminal_i
apr_snprintf(ied_usr->dev_type, sizeof(ied_usr->dev_type), "%s", update[i].dev_type);
printf("ied_usr->dev_type: %s\n", ied_usr->dev_type);
}
if (update[i].com_type != NULL && update[i].com_type[0] != '\0') {
apr_snprintf(ied_usr->com_type, sizeof(ied_usr->com_type), "%s", update[i].com_type);
printf("ied_usr->com_type: %s\n", ied_usr->com_type);
} else {
apr_snprintf(ied_usr->com_type, sizeof(ied_usr->com_type), "%s", "MMS");
printf("ied_usr->com_type (default): %s\n", ied_usr->com_type);
}
if (update[i].processNo != NULL) {
apr_snprintf(ied_usr->processNo, sizeof(ied_usr->processNo), "%s", update[i].processNo);
printf("ied_usr->processNo: %s\n", ied_usr->processNo);
@@ -6666,7 +6581,6 @@ void clearIedUsr(ied_usr_t *ied_usr) {
memset(ied_usr->dev_type, 0, sizeof(ied_usr->dev_type));
memset(ied_usr->dev_key, 0, sizeof(ied_usr->dev_key));
memset(ied_usr->dev_series, 0, sizeof(ied_usr->dev_series));
memset(ied_usr->com_type, 0, sizeof(ied_usr->com_type));
memset(ied_usr->terminal_id, 0, sizeof(ied_usr->terminal_id));
memset(ied_usr->org_name, 0, sizeof(ied_usr->org_name));
memset(ied_usr->maint_name, 0, sizeof(ied_usr->maint_name));
@@ -7393,8 +7307,6 @@ const char* get_front_msg_from_subdir() {
return "实时数据进程";
else if (std::strstr(subdir, "cfg_soe_comtrade") != NULL)
return "暂态和告警进程";
else if (std::strstr(subdir, "cfg_pqdif_data") != NULL)
return "pqdif处理进程";
else if (std::strstr(subdir, "cfg_recallhis_data") != NULL)
return "稳态补招进程";
else if (std::strstr(subdir, "cfg_stat_data") != NULL)

2
cfg_parse/log4.cpp
View File

@@ -229,8 +229,6 @@ std::string get_front_type_from_subdir() {
return "realTime";
else if (std::strstr(subdir, "cfg_soe_comtrade") != NULL)
return "comtrade";
else if (std::strstr(subdir, "cfg_pqdif_data") != NULL)
return "PQDIF";
else if (std::strstr(subdir, "cfg_recallhis_data") != NULL)
return "recall";
else if (std::strstr(subdir, "cfg_stat_data") != NULL)

16
include/rocketmq/SimpleProducer.h
View File

@@ -23,9 +23,9 @@ using namespace rocketmq;
//void producer_send0();
//void StartSendMessage(CProducer* producer,const char* strbody);
//void producer_send(const char* strbody);
//void rocketmq_producer_send(const char* strbody,const char* topic);
//void RocketMQ_rocketmq_producer_send(const char* strbody,const char* topic);
//void rocketmq_StartSendMessage(CProducer* producer,const char* strbody,const char* topic);
void rocketmq_producer_send(const std::string& body,
void RocketMQ_rocketmq_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys);
@@ -42,8 +42,8 @@ extern void my_rocketmq_send(Ckafka_data_t& data);
///////////////////////////////////////////////////////生产者
void InitializeProducer();
void ShutdownAndDestroyProducer();
void RocketMQ_InitializeProducer();
void RocketMQ_ShutdownAndDestroyProducer();
//////////////////////////////////////////////////////消费者
typedef ConsumeStatus (*MessageCallBack)(
const MQMessageExt& msg
@@ -59,13 +59,13 @@ struct Subscription {
MessageCallBack cb)
: topic(t), tag(tg), callback(cb) {}
};
//void InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key);
void InitializeConsumer(const std::string& consumerName,
//void RocketMQ_InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key);
void RocketMQ_InitializeConsumer(const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
void ShutdownAndDestroyConsumer();
void RocketMQ_ShutdownAndDestroyConsumer();
void rocketmq_consumer_receive(
void RocketMQ_rocketmq_consumer_receive(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);

1
json/mms_json_inter.h
View File

@@ -26,7 +26,6 @@
#define THREE_SECS_DATA_BASE_NODE_ID 200 // 三秒数据、实时数据
#define SOE_COMTRADE_BASE_NODE_ID 300 // soe 录波
#define HIS_DATA_BASE_NODE_ID 400 //历史、补招
#define PQDIF_DATA_BASE_NODE_ID 800 // pqdif数据
class json_block_data //json拼接参数类
{

73
json/save2json.cpp
View File

@@ -271,8 +271,6 @@ void add_comm_log(char* log_str)
com_log_fn += "comm_300_comtrade.txt";
else if (g_node_id == HIS_DATA_BASE_NODE_ID)
com_log_fn += "comm_400_his.txt";
else if (g_node_id == PQDIF_DATA_BASE_NODE_ID)
com_log_fn += "comm_800_pqdif.txt";
else if (g_node_id == NEW_HIS_DATA_BASE_NODE_ID) {
com_log_fn.append(QString("comm_400_his_%1.txt").arg(g_front_seg_index));
}
@@ -1196,48 +1194,6 @@ void add_indent(std::stringstream& stream, int level) {
}
}
static bool is_cfg_pqdif_process()
{
return std::strstr(subdir, "cfg_pqdif_data") != NULL;
}
static bool is_pqdif_com_type(const char* com_type)
{
QString type = com_type ? QString::fromLocal8Bit(com_type).trimmed() : QString();
return type.compare("PQDIF", Qt::CaseInsensitive) == 0;
}
static void fill_terminal_com_type(terminal* tmnl, cJSON* item)
{
if (tmnl == NULL)
return;
cJSON* comType = cJSON_GetObjectItem(item, "comType");
if (comType == NULL)
comType = cJSON_GetObjectItem(item, "com_type");
if (comType && comType->type == cJSON_String)
std::strncpy(tmnl->com_type, comType->valuestring, sizeof(tmnl->com_type) - 1);
else
std::strncpy(tmnl->com_type, "MMS", sizeof(tmnl->com_type) - 1);
tmnl->com_type[sizeof(tmnl->com_type) - 1] = '\0';
}
static bool terminal_update_should_write_file(const terminal* tmnl)
{
if (tmnl == NULL)
return false;
if (is_pqdif_com_type(tmnl->com_type))
return is_cfg_pqdif_process();
if (is_cfg_pqdif_process())
return false;
return true;
}
std::string prepare_update(const std::string& code_str, const terminal& json_data,const std::string& guid) //添加guid
{
std::cout << "prepare update" << std::endl;
@@ -1309,9 +1265,6 @@ std::string prepare_update(const std::string& code_str, const terminal& json_dat
add_indent(xmlStream, indentLevel);
xmlStream << "<series>" << json_data.dev_series << "</series>" << std::endl;
add_indent(xmlStream, indentLevel);
xmlStream << "<comType>" << (json_data.com_type[0] ? json_data.com_type : "MMS") << "</comType>" << std::endl;
//lnk20250210
add_indent(xmlStream, indentLevel);
xmlStream << "<processNo>" << json_data.processNo << "</processNo>" << std::endl;
@@ -1394,9 +1347,6 @@ std::string prepare_update(const std::string& code_str, const terminal& json_dat
add_indent(xmlStream, indentLevel);
xmlStream << "<id>" << json_data.terminal_id << "</id>" << std::endl;
add_indent(xmlStream, indentLevel);
xmlStream << "<comType>" << (json_data.com_type[0] ? json_data.com_type : "MMS") << "</comType>" << std::endl;
indentLevel--;
add_indent(xmlStream, indentLevel);
xmlStream << "</terminalData>" << std::endl;
@@ -1675,7 +1625,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
std::string guid = guidstr->valuestring;
//进程号为0的进程处理所有台账更新消息
if (!is_cfg_pqdif_process() && process_No != g_front_seg_index && g_front_seg_index !=0) {
if (process_No != g_front_seg_index && g_front_seg_index !=0) {
std::cout << "msg index:"<< process_No <<"doesnt match self index:" << g_front_seg_index << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
@@ -1704,7 +1654,6 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
cJSON* item = cJSON_GetArrayItem(data, i);
terminal json_data;
memset(&json_data, 0, sizeof(json_data));
// 填充 terminal_dev 的数据
cJSON* id = cJSON_GetObjectItem(item, "id"); // terminal_id
if (id && id->type == cJSON_String)
@@ -1766,8 +1715,6 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
else
std::strncpy(json_data.dev_series, "N/A", sizeof(json_data.dev_series) - 1);
fill_terminal_com_type(&json_data, item);
//lnk20250210台账进程号
cJSON* processNo = cJSON_GetObjectItem(item, "processNo"); // processNo转为字符串
if (processNo && processNo->type == cJSON_Number) snprintf(json_data.processNo, sizeof(json_data.processNo), "%d", processNo->valueint);
@@ -1818,7 +1765,6 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
for (int j = 0; j < monitorData_size && j < 10; j++) { // 最多 10 个监测点
cJSON* monitor_item = cJSON_GetArrayItem(monitorData, j);
monitor monitor_data;
memset(&monitor_data, 0, sizeof(monitor_data));
cJSON* monitor_id = cJSON_GetObjectItem(monitor_item, "id"); // monitor_id
if (monitor_id && monitor_id->type == cJSON_String)
@@ -1889,13 +1835,6 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
print_terminal(&json_data);
if (!terminal_update_should_write_file(&json_data)) {
std::cout << "skip ledger update by comType:" << json_data.com_type
<< " terminal:" << json_data.terminal_id
<< " process:" << subdir << g_front_seg_index << std::endl;
continue;
}
// 准备 XML 内容并写入文件
std::string xmlContent = prepare_update(code_str, json_data,guid);//添加guid20250506
if (xmlContent != "") {
@@ -1929,17 +1868,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
cJSON* id = cJSON_GetObjectItem(item, "id");
if (id != nullptr) {
terminal json_data;
memset(&json_data, 0, sizeof(json_data));
std::strncpy(json_data.terminal_id, cJSON_GetObjectItem(item, "id")->valuestring, sizeof(json_data.terminal_id) - 1);
fill_terminal_com_type(&json_data, item);
if (!terminal_update_should_write_file(&json_data)) {
std::cout << "skip delete ledger update by comType:" << json_data.com_type
<< " terminal:" << json_data.terminal_id
<< " process:" << subdir << g_front_seg_index << std::endl;
continue;
}
// 准备 XML 内容并写入文件
std::string xmlContent = prepare_update(code_str, json_data,guid);//添加guid20250506
if(xmlContent != ""){

1
json/save2json.h
View File

@@ -688,7 +688,6 @@ struct terminal // 终端台账
char dev_type[64];
char dev_key[255];
char dev_series[255];
char com_type[32];
char processNo[64]; //lnk20250210进程号
char addr_str[64];
char port[64];

6
mms/db_interface.h
View File

@@ -31,7 +31,6 @@
#define NEW_HIS_DATA_BASE_NODE_ID 500
#define RECALL_HIS_DATA_BASE_NODE_ID 600
#define RECALL_ALL_DATA_BASE_NODE_ID 700
#define PQDIF_DATA_BASE_NODE_ID 800
#define REPORT_TYPE_STAT 1
#define REPORT_TYPE_REAL 2
@@ -43,6 +42,11 @@
#define SHOULD_UNREGISTER 2
///////////////////////////////////////////////////////////////////////////////
#define MQ_BACKEND_ROCKETMQ 0
#define MQ_BACKEND_REDIS_STREAM 1
///////////////////////////////////////////////////////////////////////////////
#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */

2
mms/main.c
View File

@@ -81,8 +81,6 @@ void init_global_function_enable()
three_secs_enabled = 1;
}else if (strcmp(subdir,"cfg_soe_comtrade")==0) { //告警和录波和暂态
g_node_id = SOE_COMTRADE_BASE_NODE_ID;
}else if (strcmp(subdir,"cfg_pqdif_data")==0) {
g_node_id = PQDIF_DATA_BASE_NODE_ID;
}else if (strcmp(subdir,"cfg_his_data")==0) { //不使用
g_node_id = RECALL_ALL_DATA_BASE_NODE_ID;
g_node_id = HIS_DATA_BASE_NODE_ID;

569
mms/mms_process.c
View File

@@ -44,13 +44,7 @@ SD_CONST static ST_CHAR *SD_CONST thisFileName = __FILE__;
#ifdef _OS_UNIX_
#include <sys/vfs.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#endif
#include <ctype.h>
#include <errno.h>
#include <time.h>
extern uint32_t g_dead_lock_counter;
extern uint32_t g_thread_blocked_times;
@@ -101,9 +95,6 @@ static void format_time_ms(long long ms, char* buf, size_t buf_len){
tm_time.tm_sec,
milli);
}
static apr_status_t ChannelCheckPQDIFFiles(chnl_usr_t *chnl_usr);
static apr_status_t ChannelRecallPQDIFFiles(chnl_usr_t *chnl_usr, LD_info_t *LD_info, long long start_sec, long long end_sec);
//lnk20250122start
apr_status_t init_rem_dib_table()
{
@@ -581,28 +572,21 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
LD_info_t *LD_info;
loginfo_t *loginfo = NULL;
int cpuno;
int is_pqdif_process;
double now;
static double last_check_recall_config_time = 0.0;
ied = chnl_usr->chnl->ied;
ied_usr = GET_IEDEXT_ADDR(ied);
is_pqdif_process = (g_node_id == PQDIF_DATA_BASE_NODE_ID);
for(cpuno=0 ; cpuno<ied->cpucount; cpuno++) {
LD_info = &(ied_usr->LD_info[cpuno]);
//添加保护，防止台账中测点号不连续导致的崩溃
if (LD_info->cpuno == 0 ||
LD_info->LD_name == NULL)
{
continue;
}
if (!is_pqdif_process &&
(LD_info->logcount <= 0 ||
LD_info->loginfo == NULL))
LD_info->LD_name == NULL ||
LD_info->logcount <= 0 ||
LD_info->loginfo == NULL)
{
continue;
}
@@ -628,13 +612,11 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
continue;
}*/
if (!is_pqdif_process) {
loginfo = LD_info->loginfo[0] ;
//添加保护，防止台账中测点号不连续导致的崩溃
if (loginfo == NULL || loginfo->LD_info == NULL)
continue;
}
apr_sleep(apr_time_from_sec(1) / 10);
@@ -646,17 +628,14 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
int failed_count = 0;
/////////////////////////////////////////////////////根据配置文件控制下发补招时间为北京时间还是utc时间
long long utc_or_beijing = 0;
if (!is_pqdif_process) {
printf("~~~~~~~this dev type is %s~~~~~~~",ied_usr->dev_type);
XmlConfigC cfg1;
memset(&cfg1,0,sizeof(cfg1));
if (get_xml_config_by_dev_type(ied_usr->dev_type, &cfg1)) {
printf("ValueOfTimeUnit = %s\n", cfg1.ValueOfTimeUnit);
} else {
printf("读取失败，未找到 dev_type\n");
}
utc_or_beijing = 0;
long long utc_or_beijing;
if(strcmp(cfg1.ValueOfTimeUnit, "utc") == 0){//装置时间是utc还是北京
utc_or_beijing = 28800;//秒
@@ -667,11 +646,6 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
DIY_WARNLOG_CODE(LD_info->mp_id,2,LOG_CODE_RECALL,"【WARN】监测点:%s - id:%s开始补招数据,注意!下发补招时间为beijing时间,监测点对应装置型号:%s", LD_info->name,LD_info->mp_id,ied_usr->dev_type);
}
}
else {
DIY_WARNLOG_CODE(LD_info->mp_id,2,LOG_CODE_RECALL,"[PQDIF][RECALL] monitor:%s - id:%s start recall PQDIF files, dev_type:%s", LD_info->name,LD_info->mp_id,ied_usr->dev_type);
}
//////////////////////////////////////////////////////////////
//记录本次补招的时间范围
long long min_start_sec = LLONG_MAX;
@@ -692,9 +666,6 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
LD_info->autorecallflag = 1;//正在补招
//当前不区分稳态和暂态lnk20241030，如果做区分修改：Check_Recall_Config从xml文件获取数据后，赋值给了LD_info
if (is_pqdif_process)
continue;
loginfo->need_steady = LD_info->autorecall[i]->need_steady; loginfo->need_voltage = LD_info->autorecall[i]->need_voltage;
loginfo->start_time = apr_time_from_sec(LD_info->autorecall[i]->start - 5);//保证时间开头
@@ -747,24 +718,6 @@ void ChannelCheckIECLogs(chnl_usr_t *chnl_usr)
//不管是否成功，这个补招文件必须删除，可能出现一直失败，循环读取文件和循环补招导致程序崩溃202050724lnk
//if (failed_count==0) {//成功
if (is_pqdif_process) {
if (min_start_sec == LLONG_MAX || max_end_sec <= min_start_sec) {
failed_count++;
printf("[PQDIF][RECALL] invalid recall range mp_id=%s min=%lld max=%lld\n",
LD_info->mp_id, min_start_sec, max_end_sec);
}
else {
ret = ChannelRecallPQDIFFiles(chnl_usr, LD_info, min_start_sec, max_end_sec);
if (ret != APR_SUCCESS)
failed_count++;
}
g_dead_lock_counter = 0;
g_thread_blocked_times = 0;
now = sGetMsTime();
last_check_recall_config_time = now;
}
Delete_recall_Xml(LD_info->mp_id);
// ===== [新增] 组装上送的补招开始/结束时间字符串（本地时区，直接“秒转字符串”）=====
@@ -1176,7 +1129,6 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
//3-写入台账内容///////////////////////////////////
//4-配置映射文件//////////////////////////////
if (g_node_id != PQDIF_DATA_BASE_NODE_ID) {
char model[64] = {0};
// 获取模型ID，检查是否返回 NULL
parse_model_cfg_web_one(ied,&model);//存储在/FeProject/dat/
@@ -1211,11 +1163,6 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
//5-报告块初始化///////////////////////////////////
//调试
}
else {
printf("[PQDIF] skip model xml and report init terminal=%s\n", ied_usr->terminal_id);
}
printf("ledger id: %s\n", ((ied_usr_t*)ied->channel[0].ied->usr_ext)->terminal_id);
//6-init_rem_dib_table//////////////////////////////
@@ -1285,7 +1232,6 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
//3-写入台账内容////////////////////////////////////////////
//4-配置映射文件///////////////////////////////////////////
if (g_node_id != PQDIF_DATA_BASE_NODE_ID) {
char model[64] = {0};
// 获取模型ID，检查是否返回 NULL
parse_model_cfg_web_one(ied,&model);//存储在/FeProject/dat/
@@ -1320,11 +1266,6 @@ void process_ledger_update(trigger_update_xml_t *ledger_update_xml)
parse_rpt_log_ini_one(ied);
//5-报告块初始化///////////////////////////////////
}
else {
printf("[PQDIF] skip model xml and report init terminal=%s\n", ied_usr->terminal_id);
}
//6-init_rem_dib_table//////////////////////////////
init_rem_dib_table_one(ied);
//6-init_rem_dib_table///////////////////////////////////
@@ -1482,7 +1423,6 @@ void print_terminal(const terminal* tmnl) {
printf("Device Type: %s\n", is_empty(tmnl->dev_type) ? "N/A" : tmnl->dev_type);
printf("Device Key: %s\n", is_empty(tmnl->dev_key) ? "N/A" : tmnl->dev_key);
printf("Device Series: %s\n", is_empty(tmnl->dev_series) ? "N/A" : tmnl->dev_series);
printf("Com Type: %s\n", is_empty(tmnl->com_type) ? "MMS" : tmnl->com_type);
printf("Address: %s\n", is_empty(tmnl->addr_str) ? "N/A" : tmnl->addr_str);
printf("Port: %s\n", is_empty(tmnl->port) ? "N/A" : tmnl->port);
printf("Timestamp: %s\n", is_empty(tmnl->timestamp) ? "N/A" : tmnl->timestamp);
@@ -1694,19 +1634,12 @@ void CheckAllConnectedChannel()
if(DEBUGOPEN)printf("[FILEDIR] enter HandleFileDirReqForChannel");
HandleFileDirReqForChannel(chnl_usr);//文件目录请求
}
if(g_node_id == PQDIF_DATA_BASE_NODE_ID){
ChannelCheckPQDIFFiles(chnl_usr);
}
else{
ChannelCheckIECReports(chnl_usr);//报告
}
ChannelCheckIECReports(chnl_usr);//报告
if ( (g_node_id == SOE_COMTRADE_BASE_NODE_ID) || (g_node_id == HIS_DATA_BASE_NODE_ID) || (g_node_id == NEW_HIS_DATA_BASE_NODE_ID) || (g_node_id == RECALL_HIS_DATA_BASE_NODE_ID) || (g_node_id == RECALL_ALL_DATA_BASE_NODE_ID))
ChannelCheckWaveFiles(chnl_usr);//录波文件
if(g_node_id == HIS_DATA_BASE_NODE_ID || g_node_id == NEW_HIS_DATA_BASE_NODE_ID || g_node_id == RECALL_HIS_DATA_BASE_NODE_ID || (g_node_id == RECALL_ALL_DATA_BASE_NODE_ID)
|| g_node_id == PQDIF_DATA_BASE_NODE_ID)
if(g_node_id == HIS_DATA_BASE_NODE_ID || g_node_id == NEW_HIS_DATA_BASE_NODE_ID || g_node_id == RECALL_HIS_DATA_BASE_NODE_ID || (g_node_id == RECALL_ALL_DATA_BASE_NODE_ID))
ChannelCheckIECLogs(chnl_usr);//补招文件
if ( (sGetMsTime() - chnl_usr->m_LastPosRespTime) > 15*1000 ) //wait 15 secs，隔15秒获取一次响应，两次没响应后关闭
{
char** varnames ;
@@ -2049,361 +1982,6 @@ static ST_RET Write_Named_Var(LD_info_t *LD_info,chnl_usr_t *chnl_usr,char* VarN
#define TICKS_PER_MIN 60LL*1000LL*1000LL
/* PQDIF 文件召唤参数。
* C# 读取程序的默认周期是 2 小时，文件名形如：
* PQMonitor_PQM1_20190401_1400_02.pqd
* 其中 “1_20190401_1400_” 用于匹配测点序号和时间片。
* 当前进程没有单独的手动时间窗输入，所以这里实现自动模式：
* 1. 每个终端按固定扫描间隔读取一次 /PQDIF/ 目录；
* 2. 只下载最近一个完整周期内、且测点序号匹配的 .pqd 文件；
* 3. 下载前检查 /FeProject/data/pqdif/ 下是否已有同名文件，避免重复下载。
*/
#define PQDIF_REMOTE_DIR "/PQDIF/"
#define PQDIF_LOCAL_DIR "/FeProject/data/pqdif"
#define PQDIF_AUTO_INTERVAL_HOURS 2
#define PQDIF_SCAN_INTERVAL_MS (60 * 1000)
#define PQDIF_MAX_DEVICE_INDEX 4096
#define PQDIF_MAX_FETCH_PER_SCAN 32
static const char* pqdif_basename(const char* filename)
{
const char* base = filename;
const char* p;
if (filename == NULL)
return "";
for (p = filename; *p != '\0'; ++p) {
if (*p == '/' || *p == '\\')
base = p + 1;
}
return base;
}
static int pqdif_has_pqd_suffix(const char* filename)
{
size_t len;
if (filename == NULL)
return FALSE;
len = strlen(filename);
if (len < 4)
return FALSE;
return (filename[len - 4] == '.'
&& tolower((unsigned char)filename[len - 3]) == 'p'
&& tolower((unsigned char)filename[len - 2]) == 'q'
&& tolower((unsigned char)filename[len - 1]) == 'd');
}
static void pqdif_make_time_match(int cpu_no, time_t tm_value, char* out, size_t out_len)
{
struct tm tm_local;
memset(&tm_local, 0, sizeof(tm_local));
localtime_r(&tm_value, &tm_local);
/* C# 目录匹配逻辑使用 “测点号_yyyyMMdd_HH00_” 作为关键片段。
* 例如测点号 1、2026-07-17 10 点，对应 “1_20260717_1000_”。
*/
snprintf(out, out_len, "%d_%04d%02d%02d_%02d00_",
cpu_no % 10,
tm_local.tm_year + 1900,
tm_local.tm_mon + 1,
tm_local.tm_mday,
tm_local.tm_hour);
}
static int pqdif_use_interval_end_time(const char* dev_type)
{
return (dev_type != NULL && strcmp(dev_type, "PS_NET_PQDIF") == 0);
}
static time_t pqdif_align_interval_start(long long sec)
{
time_t hour_floor;
struct tm tm_local;
int hour_mod;
memset(&tm_local, 0, sizeof(tm_local));
hour_floor = (time_t)sec;
localtime_r(&hour_floor, &tm_local);
tm_local.tm_min = 0;
tm_local.tm_sec = 0;
hour_floor = mktime(&tm_local);
hour_mod = tm_local.tm_hour % PQDIF_AUTO_INTERVAL_HOURS;
return hour_floor - hour_mod * 3600;
}
static int pqdif_match_interval(const char* filename, int cpu_no, time_t interval_start, const char* dev_type)
{
time_t name_time;
char time_match[64];
if (!pqdif_has_pqd_suffix(filename))
return FALSE;
if (cpu_no <= 0)
return FALSE;
/* C# 原逻辑按设备类型决定文件名时间：
* - 普通设备用周期开始时间；
* - PS_NET_PQDIF 用周期结束时间。
* 这里保持同样规则，避免把普通设备下一周期的开始文件误认为本周期结束文件。
*/
name_time = interval_start;
if (pqdif_use_interval_end_time(dev_type))
name_time += PQDIF_AUTO_INTERVAL_HOURS * 3600;
pqdif_make_time_match(cpu_no, name_time, time_match, sizeof(time_match));
return strstr(filename, time_match) != NULL;
}
static int pqdif_match_recent_interval(const char* filename, int cpu_no, const char* dev_type)
{
time_t now_time;
time_t interval_start;
now_time = time(NULL);
interval_start = pqdif_align_interval_start((long long)now_time) - PQDIF_AUTO_INTERVAL_HOURS * 3600;
return pqdif_match_interval(filename, cpu_no, interval_start, dev_type);
}
static int pqdif_ensure_local_dir()
{
#ifdef _OS_UNIX_
if (mkdir("/FeProject/data", 0777) != 0 && errno != EEXIST)
return SD_FAILURE;
if (mkdir(PQDIF_LOCAL_DIR, 0777) != 0 && errno != EEXIST)
return SD_FAILURE;
#endif
return SD_SUCCESS;
}
static int pqdif_local_file_exists(const char* filename)
{
char local_file[512];
const char* base = pqdif_basename(filename);
if (base[0] == '\0')
return FALSE;
snprintf(local_file, sizeof(local_file), "%s/%s", PQDIF_LOCAL_DIR, base);
#ifdef _OS_UNIX_
return access(local_file, F_OK) == 0;
#else
return FALSE;
#endif
}
static void pqdif_build_remote_filename(const char* dir_filename, char* out, size_t out_len)
{
if (dir_filename == NULL || dir_filename[0] == '\0') {
out[0] = '\0';
return;
}
/* mms_mvla_fdir 的返回值有两种可能：
* 1. 带目录：/PQDIF/xxx.pqd
* 2. 仅文件名：xxx.pqd
* PQDIF 必须固定从 /PQDIF/ 取，所以仅返回文件名时要补齐远端目录。
*/
if (strchr(dir_filename, '/') != NULL || strchr(dir_filename, '\\') != NULL)
snprintf(out, out_len, "%s", dir_filename);
else
snprintf(out, out_len, "%s%s", PQDIF_REMOTE_DIR, dir_filename);
}
static int pqdif_get_file_from_device(chnl_usr_t *chnl_usr, const char *rem_filename, char *only_filename_ret, size_t only_filename_ret_len)
{
int ret;
char loc_file_fullname[512];
char rem_file_fullname[512];
const char *only_filename;
if (chnl_usr == NULL || chnl_usr->net_info == NULL || rem_filename == NULL || rem_filename[0] == '\0')
return SD_FAILURE;
only_filename = pqdif_basename(rem_filename);
if (only_filename[0] == '\0')
return SD_FAILURE;
if (pqdif_ensure_local_dir() != SD_SUCCESS) {
echo_warn1("[PQDIF] create local dir failed: %s\n", PQDIF_LOCAL_DIR);
return SD_FAILURE;
}
memset(loc_file_fullname,0,sizeof(loc_file_fullname));
memset(rem_file_fullname,0,sizeof(rem_file_fullname));
apr_snprintf(loc_file_fullname,sizeof(loc_file_fullname),"%s/%s",PQDIF_LOCAL_DIR,only_filename);
apr_snprintf(rem_file_fullname,sizeof(rem_file_fullname),"%s",rem_filename);
if (only_filename_ret != NULL && only_filename_ret_len > 0)
apr_snprintf(only_filename_ret,only_filename_ret_len,"%s",only_filename);
printf("[PQDIF] mms_getFile local=%s remote=%s\n", loc_file_fullname, rem_file_fullname);
ret = mms_getFile(chnl_usr->net_info,loc_file_fullname,rem_file_fullname,g_pt61850app->mmsOpTimeout);
if (ret != SD_SUCCESS) {
echo_warn3("[PQDIF] mms_getFile failed IP=%s local=%s remote=%s\n",
chnl_usr->ip_str, loc_file_fullname, rem_file_fullname);
return ret;
}
#ifdef _OS_UNIX_
{
struct stat st;
if (stat(loc_file_fullname, &st) != 0 || st.st_size <= 0) {
unlink(loc_file_fullname);
echo_warn2("[PQDIF] file is empty or invalid: %s from %s\n",
loc_file_fullname, rem_file_fullname);
return SD_FAILURE;
}
}
#endif
return SD_SUCCESS;
}
static apr_status_t ChannelRecallPQDIFFiles(chnl_usr_t *chnl_usr, LD_info_t *LD_info, long long start_sec, long long end_sec)
{
ied_t *ied;
ied_usr_t *ied_usr;
char **filenames = NULL;
int filenum = 0;
int ret;
int file_idx;
int fetched = 0;
int existed = 0;
int missing = 0;
int failed = 0;
time_t interval_start;
time_t interval_step;
if (chnl_usr == NULL || chnl_usr->chnl == NULL || chnl_usr->chnl->ied == NULL ||
chnl_usr->chnl->ied->usr_ext == NULL || chnl_usr->net_info == NULL ||
LD_info == NULL) {
printf("[PQDIF][RECALL] invalid context, skip\n");
return APR_EGENERAL;
}
if (LD_info->cpuno == 0 || LD_info->LD_name == NULL || LD_info->mp_id[0] == '\0') {
printf("[PQDIF][RECALL] invalid LD_info, skip\n");
return APR_EGENERAL;
}
if (end_sec <= start_sec) {
printf("[PQDIF][RECALL] invalid time range mp_id=%s start=%lld end=%lld\n",
LD_info->mp_id, start_sec, end_sec);
return APR_EGENERAL;
}
ied = chnl_usr->chnl->ied;
ied_usr = (ied_usr_t*)ied->usr_ext;
if (pqdif_ensure_local_dir() != SD_SUCCESS) {
echo_warn1("[PQDIF][RECALL] create local dir failed: %s\n", PQDIF_LOCAL_DIR);
return APR_EGENERAL;
}
/* 单点补招与 C# 手动/补招逻辑保持一致：
* 1. 先读取装置 /PQDIF/ 目录；
* 2. 按 PQDIF_AUTO_INTERVAL_HOURS 将补招时间范围切成多个文件周期；
* 3. 普通设备用周期开始时间匹配文件名，PS_NET_PQDIF 用周期结束时间匹配文件名；
* 4. 每个周期只取一份匹配文件，下载接口直接调用 mms_getFile。
*/
ret = mms_mvla_fdir(chnl_usr->net_info,(ST_CHAR*)PQDIF_REMOTE_DIR,
3*g_pt61850app->mmsOpTimeout,&filenames,&filenum,g_pt61850app->tmp_pool);
if (ret != SD_SUCCESS) {
echo_warn2("[PQDIF][RECALL] mms_mvla_fdir failed, IED=%d IP=%s\n", ied->id, chnl_usr->ip_str);
return APR_EGENERAL;
}
if (filenames == NULL || filenum <= 0) {
printf("[PQDIF][RECALL] no file in remote dir %s, mp_id=%s IP=%s\n",
PQDIF_REMOTE_DIR, LD_info->mp_id, chnl_usr->ip_str);
return APR_EGENERAL;
}
interval_step = PQDIF_AUTO_INTERVAL_HOURS * 3600;
interval_start = pqdif_align_interval_start(start_sec);
printf("[PQDIF][RECALL] begin mp_id=%s cpuno=%d range=%lld~%lld interval_start=%lld files=%d dev_type=%s\n",
LD_info->mp_id, LD_info->cpuno, start_sec, end_sec, (long long)interval_start,
filenum, ied_usr->dev_type);
while ((long long)interval_start < end_sec) {
int found_this_interval = FALSE;
for (file_idx = 0; file_idx < filenum; file_idx++) {
const char *base_name;
char remote_filename[256];
char only_filename_ret[256];
int ret_get;
if (filenames[file_idx] == NULL)
continue;
base_name = pqdif_basename(filenames[file_idx]);
if (base_name[0] == '\0')
continue;
if (!pqdif_match_interval(base_name, LD_info->cpuno, interval_start, ied_usr->dev_type))
continue;
found_this_interval = TRUE;
if (pqdif_local_file_exists(base_name)) {
existed++;
printf("[PQDIF][RECALL] local file exists, skip mp_id=%s interval=%lld file=%s\n",
LD_info->mp_id, (long long)interval_start, base_name);
break;
}
memset(remote_filename,0,sizeof(remote_filename));
memset(only_filename_ret,0,sizeof(only_filename_ret));
pqdif_build_remote_filename(filenames[file_idx], remote_filename, sizeof(remote_filename));
if (remote_filename[0] == '\0') {
failed++;
break;
}
ret_get = pqdif_get_file_from_device(chnl_usr, remote_filename, only_filename_ret, sizeof(only_filename_ret));
if (ret_get == SD_SUCCESS) {
fetched++;
chnl_usr->m_LastPosRespTime = sGetMsTime();
printf("[PQDIF][RECALL] fetch success mp_id=%s interval=%lld remote=%s local=%s/%s\n",
LD_info->mp_id, (long long)interval_start, remote_filename, PQDIF_LOCAL_DIR, only_filename_ret);
}
else {
failed++;
echo_warn3("[PQDIF][RECALL] fetch failed mp_id=%s remote=%s IP=%s\n",
LD_info->mp_id, remote_filename, chnl_usr->ip_str);
}
break;
}
if (!found_this_interval) {
missing++;
printf("[PQDIF][RECALL] no matched file mp_id=%s interval=%lld dev_type=%s\n",
LD_info->mp_id, (long long)interval_start, ied_usr->dev_type);
}
interval_start += interval_step;
}
printf("[PQDIF][RECALL] finish mp_id=%s fetched=%d existed=%d missing=%d failed=%d\n",
LD_info->mp_id, fetched, existed, missing, failed);
if (missing > 0 || failed > 0)
return APR_EGENERAL;
return APR_SUCCESS;
}
int pt61850_write_cn_file(chnl_usr_t *chnl_usr, ied_t *ied, char *rem_filename, char *only_filename_ret)
{
@@ -2449,140 +2027,6 @@ int pt61850_write_cn_file(chnl_usr_t *chnl_usr, ied_t *ied, char *rem_filename,
return ret;
}
static apr_status_t ChannelCheckPQDIFFiles(chnl_usr_t *chnl_usr)
{
ied_t *ied;
ied_usr_t *ied_usr;
LD_info_t *LD_info;
char **filenames = NULL;
int filenum = 0;
int ret;
int cpuno;
int file_idx;
int fetched = 0;
int matched = 0;
int scan_idx;
double now_ms;
static double s_last_scan_ms[PQDIF_MAX_DEVICE_INDEX] = {0};
if (chnl_usr == NULL || chnl_usr->chnl == NULL || chnl_usr->chnl->ied == NULL ||
chnl_usr->chnl->ied->usr_ext == NULL || chnl_usr->net_info == NULL) {
printf("[PQDIF] invalid channel context, skip scan\n");
return APR_EGENERAL;
}
ied = chnl_usr->chnl->ied;
ied_usr = (ied_usr_t*)ied->usr_ext;
/* CheckAllConnectedChannel 会轮询通道；这里按终端做节流，避免每次循环都读一次
* /PQDIF/ 目录。目录读和文件下载都走 MMS，同一个终端一分钟扫描一次即可。
*/
scan_idx = ied_usr->dev_idx;
if (scan_idx < 0 || scan_idx >= PQDIF_MAX_DEVICE_INDEX)
scan_idx = (int)chnl_usr->chnl_id;
if (scan_idx < 0 || scan_idx >= PQDIF_MAX_DEVICE_INDEX)
scan_idx = 0;
now_ms = sGetMsTime();
if (s_last_scan_ms[scan_idx] > 0 &&
(now_ms - s_last_scan_ms[scan_idx]) < PQDIF_SCAN_INTERVAL_MS) {
return APR_SUCCESS;
}
s_last_scan_ms[scan_idx] = now_ms;
/* 本函数只做 PQDIF 文件获取：
* 1. 从装置固定目录 /PQDIF/ 取文件列表；
* 2. 用台账里的 LD_info->cpuno 匹配文件名里的 “测点号_yyyyMMdd_HH00_”；
* 3. 只下载 .pqd 文件，且本地 /FeProject/data/pqdif/ 已存在同名文件时跳过；
* 4. 真正下载直接调用 mms_getFile，不改动原录波文件下载函数。
*/
if (pqdif_ensure_local_dir() != SD_SUCCESS) {
echo_warn1("[PQDIF] create local dir failed: %s\n", PQDIF_LOCAL_DIR);
return APR_EGENERAL;
}
ret = mms_mvla_fdir(chnl_usr->net_info,(ST_CHAR*)PQDIF_REMOTE_DIR,
3*g_pt61850app->mmsOpTimeout,&filenames,&filenum,g_pt61850app->tmp_pool);
if (ret != SD_SUCCESS) {
echo_warn2("[PQDIF] mms_mvla_fdir failed, IED=%d IP=%s\n", ied->id, chnl_usr->ip_str);
return APR_EGENERAL;
}
if (filenames == NULL || filenum <= 0) {
printf("[PQDIF] no file in remote dir %s, IED=%d IP=%s\n", PQDIF_REMOTE_DIR, ied->id, chnl_usr->ip_str);
return APR_SUCCESS;
}
printf("[PQDIF] scan remote dir %s success, IED=%d IP=%s file_count=%d\n",
PQDIF_REMOTE_DIR, ied->id, chnl_usr->ip_str, filenum);
for (cpuno = 0; cpuno < ied->cpucount; cpuno++) {
LD_info = &(ied_usr->LD_info[cpuno]);
if (LD_info->cpuno == 0 ||
LD_info->LD_name == NULL ||
LD_info->mp_id[0] == '\0') {
continue;
}
for (file_idx = 0; file_idx < filenum; file_idx++) {
const char *base_name;
char remote_filename[256];
char only_filename_ret[256];
int ret_get;
if (filenames[file_idx] == NULL)
continue;
base_name = pqdif_basename(filenames[file_idx]);
if (base_name[0] == '\0')
continue;
/* C# 读取程序的兼容逻辑是读目录后按片段匹配：
* cpuNo%10_yyyyMMdd_HH00_
* 这里保持同样的匹配方式，不依赖厂家前缀，也兼容普通设备用
* 周期开始时间命名、振兴类设备用周期结束时间命名的差异。
*/
if (!pqdif_match_recent_interval(base_name, LD_info->cpuno, ied_usr->dev_type))
continue;
matched++;
if (pqdif_local_file_exists(base_name)) {
printf("[PQDIF] local file exists, skip mp_id=%s file=%s\n", LD_info->mp_id, base_name);
continue;
}
memset(remote_filename,0,sizeof(remote_filename));
memset(only_filename_ret,0,sizeof(only_filename_ret));
pqdif_build_remote_filename(filenames[file_idx], remote_filename, sizeof(remote_filename));
if (remote_filename[0] == '\0')
continue;
ret_get = pqdif_get_file_from_device(chnl_usr, remote_filename, only_filename_ret, sizeof(only_filename_ret));
if (ret_get == SD_SUCCESS) {
fetched++;
chnl_usr->m_LastPosRespTime = sGetMsTime();
printf("[PQDIF] fetch success mp_id=%s name=%s remote=%s local=%s/%s\n",
LD_info->mp_id, LD_info->name, remote_filename, PQDIF_LOCAL_DIR, only_filename_ret);
}
else {
echo_warn3("[PQDIF] fetch failed mp_id=%s remote=%s IP=%s\n",
LD_info->mp_id, remote_filename, chnl_usr->ip_str);
}
if (fetched >= PQDIF_MAX_FETCH_PER_SCAN) {
printf("[PQDIF] reach fetch limit per scan=%d, IED=%d IP=%s\n",
PQDIF_MAX_FETCH_PER_SCAN, ied->id, chnl_usr->ip_str);
return APR_SUCCESS;
}
}
}
printf("[PQDIF] scan finish IED=%d IP=%s matched=%d fetched=%d\n",
ied->id, chnl_usr->ip_str, matched, fetched);
return APR_SUCCESS;
}
apr_status_t prepare_call_cn_wavelist(LD_info_t *LD_info, int FltNum)
{
int i;
@@ -3023,3 +2467,4 @@ char* convertMsToDateTimeString(int64_t usTime) {
return buffer;
}

11
mms/rdb_client.c
View File

@@ -205,7 +205,6 @@ apr_status_t init_rdb()
//台账读取过后初始化各级的日志
init_loggers();
if (g_node_id != PQDIF_DATA_BASE_NODE_ID) {
rv = parse_model_cfg_web();
if (rv != APR_SUCCESS) {//不可能
echo_errg("Parsed model with error,try to run! \n");
@@ -229,10 +228,6 @@ apr_status_t init_rdb()
return rv;
}
}
else {
printf("[PQDIF] skip model xml and report init in init_rdb\n");
}
}
if (app_get_private_config(g_my_conf_fname) != APR_SUCCESS) {
echo_errg("Failed when processing private configuration\n");
@@ -316,9 +311,6 @@ apr_status_t run_protocol()
else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
ServerPort = SOCKET_PORT + SOE_COMTRADE_BASE_NODE_ID + g_front_seg_index;
}
else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
ServerPort = SOCKET_PORT + PQDIF_DATA_BASE_NODE_ID + g_front_seg_index;
}
struct sockaddr_in server_sockaddr;
memset(&server_sockaddr, 0, sizeof(server_sockaddr));
@@ -359,9 +351,6 @@ apr_status_t run_protocol()
else if (g_node_id == SOE_COMTRADE_BASE_NODE_ID) {//暂态录波
HTTP_PORT = HTTP_PORT + SOE_COMTRADE_BASE_NODE_ID + g_front_seg_index;
}
else if (g_node_id == PQDIF_DATA_BASE_NODE_ID) {
HTTP_PORT = HTTP_PORT + PQDIF_DATA_BASE_NODE_ID + g_front_seg_index;
}
printf("try_start_web_http_thread \n");
try_start_web_http_thread();
printf("try_start_http_thread \n");

1
mms/rdb_client.h
View File

@@ -337,7 +337,6 @@ struct ied_usr_t{
char dev_type[256]; /**< 设备类型 */
char dev_key[256]; /**< 设备秘钥 */
char dev_series[256]; /**< 设备识别码 */
char com_type[32]; /**< 通讯类型 mms/pqdif */
int dev_flag; /**< 设备标志 */
void *cookie;
double last_call_wavelist_time ; //上次召录波列表时间

19
pt61850netd_pqfe.pro
View File

@@ -2,7 +2,13 @@ TEMPLATE = app
TARGET = pt61850netd_pqfe
DEPENDPATH += .
INCLUDEPATH += . ./source/include ./source/include/mmslite ./source/include/pg_inst ./source/include/curl ./source/include/oss_sdk ./source/include/roketmq
INCLUDEPATH += . ./source/include \
./source/include/mmslite \
./source/include/pg_inst \
./source/include/curl \
./source/include/oss_sdk \
./source/include/roketmq \
./source/redisstream
QMAKE_ORIG_TARGET = $(TARGET)
@@ -14,6 +20,8 @@ DEFINES += _CRT_SECURE_NO_WARNINGS
DEFINES += MMS_LITE LINUX=2 MOSI LEAN_T TP0_ENABLED
DEFINES += CLIENT _DEBUG _REENTRANT _GNU_SOURCE _LARGEFILE64_SOURCE
DEFINES += USE_MQ_MULTI_BACKEND
# 添加 debug/release 编译选项配置
CONFIG(debug, debug|release) {
message("Building debug version with debug symbols")
@@ -66,6 +74,7 @@ unix {
include(fe_common.pri)
SOURCES += source/mms/event2.c
INCLUDEPATH += ./source/include/apr-linux
INCLUDEPATH += /FeProject/include
LIBS += -L/FeProject/lib
LIBS += -L/FeProject/lib/pgodbc
DEFINES += DEBUG_SISCO
@@ -98,6 +107,8 @@ unix {
LIBS += -lrdkafka++
LIBS += -lhttprun
LIBS += -llog4cplus
LIBS += -lhiredis
LIBS += /FeProject/lib/libz.a
}
#install
@@ -127,7 +138,8 @@ HEADERS += source/mms/db_interface.h \
source/json/cjson.h \
source/rocketmq/SimpleProducer.h \
source/cfg_parse/custom_printf.h \
source/log4cplus/log4.h
source/log4cplus/log4.h \
source/redisstream/RedisStreamMQ.h
SOURCES += source/mms/main.c \
source/mms/clntobj.c \
@@ -163,4 +175,5 @@ SOURCES += source/mms/main.c \
source/cfg_parse/base64.cpp \
source/cfg_parse/uds_huaweiyun.cpp \
source/cfg_parse/SimpleProducer.cpp \
source/cfg_parse/log4.cpp
source/cfg_parse/log4.cpp \
source/redisstream/RedisStreamMQ.cpp

1262
redisstream/RedisStreamMQ.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

46
redisstream/RedisStreamMQ.h Normal file
View File

@@ -0,0 +1,46 @@
#ifndef REDIS_STREAM_MQ_H
#define REDIS_STREAM_MQ_H
/*
* lnk20260622 新增：Redis Streams 后端声明（运行时切换版）。
*
* 注意：
* 1. 本文件不定义 InitializeProducer / rocketmq_producer_send 等统一入口。
* 2. 只定义 RedisStream_xxx 后端函数，统一入口放在 SimpleProducer.cpp 中按配置转发。
* 3. 这样 RocketMQ 原源码和 RedisStream 源码可以同时编译进程序。
*/
//#include "mq_compat.h"
#include <string>
#include <vector>
// lnk20260626 修改：RedisStreamMQ.h 只需要知道 MQMessageExt 是一个类型即可
// 这里用前向声明，避免 RedisStreamMQ.h 直接依赖 RocketMQ 头文件
namespace rocketmq {
class MQMessageExt;
}
struct Subscription;
void RedisStream_InitializeProducer();
void RedisStream_ShutdownAndDestroyProducer();
void RedisStream_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys);
void RedisStream_InitializeConsumer(const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
void RedisStream_ShutdownAndDestroyConsumer();
void RedisStream_consumer_receive(const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
bool RedisStream_should_process_after_start(const rocketmq::MQMessageExt& msg);
#endif // REDIS_STREAM_MQ_H

82
redisstream/mq_compat.h Normal file
View File

@@ -0,0 +1,82 @@
#ifndef MQ_COMPAT_H
#define MQ_COMPAT_H
#include <string>
#include <stdint.h>
namespace mqcompat {
enum ConsumeStatus {
CONSUME_SUCCESS = 0,
RECONSUME_LATER = 1
};
class MQMessageExt {
public:
MQMessageExt()
: bornTimestamp_(0),
queueId_(0),
queueOffset_(0) {
}
MQMessageExt(const std::string& topic,
const std::string& tags,
const std::string& keys,
const std::string& body,
int64_t bornTs,
const std::string& msgId)
: topic_(topic),
tags_(tags),
keys_(keys),
body_(body),
bornTimestamp_(bornTs),
queueId_(0),
queueOffset_(0),
msgId_(msgId),
streamId_(msgId) {
}
const std::string& getTopic() const { return topic_; }
const std::string& getTags() const { return tags_; }
const std::string& getKeys() const { return keys_; }
const std::string& getBody() const { return body_; }
int64_t getBornTimestamp() const { return bornTimestamp_; }
int getQueueId() const { return queueId_; }
int64_t getQueueOffset() const { return queueOffset_; }
const std::string& getMsgId() const { return msgId_; }
const std::string& getStreamId() const { return streamId_; }
void setTopic(const std::string& v) { topic_ = v; }
void setTags(const std::string& v) { tags_ = v; }
void setKeys(const std::string& v) { keys_ = v; }
void setBody(const std::string& v) { body_ = v; }
void setBornTimestamp(int64_t v) { bornTimestamp_ = v; }
void setQueueId(int v) { queueId_ = v; }
void setQueueOffset(int64_t v) { queueOffset_ = v; }
void setMsgId(const std::string& v) { msgId_ = v; }
void setStreamId(const std::string& v) {
streamId_ = v;
if (msgId_.empty()) {
msgId_ = v;
}
}
private:
std::string topic_;
std::string tags_;
std::string keys_;
std::string body_;
int64_t bornTimestamp_;
int queueId_;
int64_t queueOffset_;
std::string msgId_;
std::string streamId_;
};
} // namespace mqcompat
#endif

85
rocketmq/SimpleProducer.h
View File

@@ -1,10 +1,10 @@
#ifndef SIMPLEPRODUCER_H
#define SIMPLEPRODUCER_H
#ifdef __cplusplus
#include "../json/mms_json_inter.h"
//#include "../rocketmq/CProducer.h"
//#include "../rocketmq/CMessage.h"
//#include "../rocketmq/CSendResult.h"
//#include "../rocketmq/CPushConsumer.h"
#include "../rocketmq/DefaultMQProducer.h"
#include "../rocketmq/MQMessage.h"
#include "../rocketmq/SendResult.h"
@@ -17,18 +17,44 @@
#include <iostream>
#include <string>
/*
* lnk20260622 修改：
* 支持 RocketMQ / Redis Streams 双后端。
*
* 编译时：两套源码都编译进去。
* 运行时：通过配置文件 [MQ] Backend=rocketmq 或 redisstream 选择。
*
* 业务层仍然使用：
* rocketmq::MQMessageExt
* rocketmq::ConsumeStatus
* Subscription
* InitializeProducer()
* rocketmq_producer_send()
* InitializeConsumer()
*
* 不需要业务层判断当前走 RocketMQ 还是 RedisStream。
*/
/*
* 关键：
* mq_compat.h 里必须定义兼容的：
* namespace rocketmq {
* enum ConsumeStatus;
* class MQMessageExt;
* }
*
* 如果实际编译 RocketMQ，也可以让 mq_compat.h 内部 include RocketMQ 原头。
*/
//#include "../mq/mq_compat.h"
#include "../rocketmq/SimpleProducer.h"
using namespace rocketmq;
/*添加测试函数lnk10-10*/
//void producer_send0();
//void StartSendMessage(CProducer* producer,const char* strbody);
//void producer_send(const char* strbody);
//void rocketmq_producer_send(const char* strbody,const char* topic);
//void rocketmq_StartSendMessage(CProducer* producer,const char* strbody,const char* topic);
void rocketmq_producer_send(const std::string& body,
void RocketMQ_rocketmq_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys);
extern "C" {
void rocketmq_test_rt();
void rocketmq_test_ud();
@@ -38,13 +64,16 @@ void rocketmq_test_set();
void rocketmq_test_only();
void rocketmq_test_300(int mpnum, int front_index, int type);
}
extern void my_rocketmq_send(Ckafka_data_t& data);
///////////////////////////////////////////////////////
// 生产者统一入口
void RocketMQ_InitializeProducer();
void RocketMQ_ShutdownAndDestroyProducer();
///////////////////////////////////////////////////////生产者
void InitializeProducer();
void ShutdownAndDestroyProducer();
//////////////////////////////////////////////////////消费者
///////////////////////////////////////////////////////
// 消费者统一入口
typedef ConsumeStatus (*MessageCallBack)(
const MQMessageExt& msg
);
@@ -59,7 +88,24 @@ struct Subscription {
MessageCallBack cb)
: topic(t), tag(tg), callback(cb) {}
};
//void InitializeConsumer(const std::string& consumerName, const std::string& nameServer, const char* topic, const char* tag, const std::string& key);
void RocketMQ_InitializeConsumer(const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
void RocketMQ_ShutdownAndDestroyConsumer();
void RocketMQ_rocketmq_consumer_receive(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
void rocketmq_producer_send(const std::string& body,
const std::string& topic,
const std::string& tags,
const std::string& keys);
void InitializeProducer();
void ShutdownAndDestroyProducer();
void InitializeConsumer(const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
@@ -69,7 +115,8 @@ void rocketmq_consumer_receive(
const std::string& consumerName,
const std::string& nameServer,
const std::vector<Subscription>& subscriptions);
//////////////////////////////////////////////////////
//////////////////////////////////////////////////////
#endif
//////////////////////////////////////////////////////
#endif // __cplusplus
#endif // SIMPLEPRODUCER_H

3
set_process.sh
View File

@@ -329,7 +329,6 @@ handle_reset() {
sed -i '/cfg_recallhis_data/d' /FeProject/etc/runtime.cf
sed -i '/cfg_3s_data/d' /FeProject/etc/runtime.cf
sed -i '/cfg_soe_comtrade/d' /FeProject/etc/runtime.cf
sed -i '/cfg_pqdif_data/d' /FeProject/etc/runtime.cf
sed -i '/fe_watchdog/d' /FeProject/etc/runtime.cf
# 根据进程号添加对应进程配置
@@ -341,7 +340,6 @@ handle_reset() {
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_recallhis_data -s 1_1^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_3s_data^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_soe_comtrade^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_pqdif_data^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
else
#看门狗固定放在第一个，防止stop时会把要杀死的进程重启
@@ -355,7 +353,6 @@ handle_reset() {
#以下部分没有多进程
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_3s_data^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_soe_comtrade^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
sed -i "2a\\$(printf '/FeProject/bin/ ^ pt61850netd_pqfe -d cfg_pqdif_data^ ^ ^ 1 ^ ^\n')" /FeProject/etc/runtime.cf
fi

1935
zlib.h Normal file
View File

File diff suppressed because it is too large Load Diff