add test fun

LFtid1056/cloudfront/code/rocketmq.cpp

@@ -1825,6 +1825,141 @@ void rocketmq_test_300(int mpnum, int front_index, int type, Front* front) {
     }
 }
 
+///////////////////////////////////////////////////////////////////////////////////////////////////////////批量台账更新函数
+// 生成无分隔符 MAC：00B88D00BB01 这种
+static std::string MakeMacNoDash(int nIndex, int nType)
+{
+    long long TestIndex = 0x0000bb00LL + (long long)nIndex + (long long)nType * 0x00010000LL;
+
+    unsigned int A = (unsigned int)(TestIndex / 65536LL);
+    unsigned int B = (unsigned int)((TestIndex & 0x0000ff00LL) >> 8);
+    unsigned int C = (unsigned int)(TestIndex & 0x000000ffLL);
+
+    std::ostringstream oss;
+    oss << "00B88D"
+        << std::uppercase << std::hex << std::setw(2) << std::setfill('0') << A
+        << std::uppercase << std::hex << std::setw(2) << std::setfill('0') << B
+        << std::uppercase << std::hex << std::setw(2) << std::setfill('0') << C;
+    return oss.str();
+}
+
+// 读取 ud.txt 模板
+static bool ReadWholeFile(const std::string& path, std::string& out)
+{
+    std::ifstream file(path.c_str(), std::ios::in);
+    if (!file.is_open()) return false;
+    std::stringstream buffer;
+    buffer << file.rdbuf();
+    out = buffer.str();
+    return true;
+}
+
+// 批量发送台账更新（入参：进程号、起始数字、生成个数）
+void rocketmq_test_ud_batch(Front* front, int processNo, int start, int count)
+{
+    if (!front || !front->m_producer) {
+        std::cerr << "front 或 producer 无效\n";
+        return;
+    }
+    if (processNo <= 0 || start <= 0 || count <= 0) {
+        std::cerr << "参数非法: processNo/start/count 必须 > 0\n";
+        return;
+    }
+
+    // 1) 读模板
+    std::string raw;
+    if (!ReadWholeFile("mult.txt", raw)) {
+        std::cerr << "读取 mult.txt 失败\n";
+        return;
+    }
+
+    // 2) 解析外层 JSON
+    nlohmann::json outer;
+    try {
+        outer = nlohmann::json::parse(raw);
+    } catch (const std::exception& e) {
+        std::cerr << "外层 JSON 解析失败: " << e.what() << "\n";
+        return;
+    }
+
+    // 3) 解析 messageBody（内层 JSON 字符串）
+    if (!outer.contains("messageBody") || !outer["messageBody"].is_string()) {
+        std::cerr << "ud.txt 缺少 messageBody 或类型不是字符串\n";
+        return;
+    }
+
+    nlohmann::json innerTpl;
+    try {
+        innerTpl = nlohmann::json::parse(outer["messageBody"].get<std::string>());
+    } catch (const std::exception& e) {
+        std::cerr << "messageBody 内层 JSON 解析失败: " << e.what() << "\n";
+        return;
+    }
+
+    // 你的 baseId（从模板里取更稳妥；这里按你给的例子“id”字段提取）
+    // innerTpl["data"][0]["id"] = "7e4a...7c96"
+    std::string baseId;
+    try {
+        baseId = innerTpl["data"][0]["id"].get<std::string>();
+    } catch (...) {
+        std::cerr << "无法从 innerTpl.data[0].id 获取 baseId\n";
+        return;
+    }
+
+    // 4) 循环构造并发送
+    for (int k = 0; k < count; ++k) {
+        int i = start + k;
+
+        // 拷贝模板，每条消息独立修改
+        nlohmann::json inner = innerTpl;
+        nlohmann::json outOuter = outer;
+
+        // 修改 processNo / maxProcessNum
+        inner["processNo"] = processNo;
+        inner["data"][0]["maxProcessNum"] = processNo;
+
+        // 生成字段
+        std::string newId = baseId + std::to_string(i);               // 设备 id
+        std::string newIp = MakeMacNoDash(i, /*nType*/0);             // 00B88D00BB01..32
+        std::string newName = std::string("压测测试lnk虚拟") + std::to_string(10 + i); // 11..(10+start+count-1)
+        std::string newMonId = std::string("00B88D00BB1") + std::to_string(i); // 101..150
+
+        // 写入 data[0]
+        inner["data"][0]["id"] = newId;
+        inner["data"][0]["ip"] = newIp;
+        inner["data"][0]["name"] = newName;
+
+        // 写入 monitorData[0]
+        if (inner["data"][0].contains("monitorData") && inner["data"][0]["monitorData"].is_array() &&
+            !inner["data"][0]["monitorData"].empty())
+        {
+            inner["data"][0]["monitorData"][0]["id"] = newMonId;
+            inner["data"][0]["monitorData"][0]["name"] = newName;
+            inner["data"][0]["monitorData"][0]["deviceId"] = newId;
+        }
+
+        // 把内层 JSON dump 回 messageBody 字符串（外层 dump 会自动转义）
+        outOuter["messageBody"] = inner.dump();
+
+        // 组装 queue_data_t 并入队
+        queue_data_t data;
+        data.monitor_no = 123; // 你原来的测试值
+        data.strTopic = G_MQCONSUMER_TOPIC_UD;
+        data.strText = outOuter.dump();
+        data.mp_id = "123123";
+        data.tag = FRONT_INST;
+        data.key = G_ROCKETMQ_KEY_TEST;
+
+        {
+            std::lock_guard<std::mutex> lock(queue_data_list_mutex);
+            queue_data_list.push_back(data);
+        }
+
+        // 间隔 1 秒
+        std::this_thread::sleep_for(std::chrono::seconds(1));
+    }
+}
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////其他测试函数
 
 void rocketmq_test_rt(Front* front)//用来测试实时数据

LFtid1056/cloudfront/code/rocketmq.h

@@ -347,6 +347,8 @@ void rocketmq_test_rt(Front* front);
 void rocketmq_test_getdir(Front* front);
 void InitializeProducer(rocketmq::RocketMQProducer*& producer);
 
+void rocketmq_test_ud_batch(Front* front, int processNo, int start, int count);
+
 #endif  // _ROCKETMQ_CLIENT_WRAPPER_H_
 
 

LFtid1056/cloudfront/code/worker.cpp

@@ -299,6 +299,7 @@ extern bool normalOutputEnabled;
                 "Available commands:\r\n"
                 "G_TEST_NUM=<num>   - Set the G_TEST_NUM\r\n"
                 "G_TEST_TYPE=<num>   - Set the G_TEST_TYPE 0:use ledger,1:use number\r\n"
+                "TESTLEDGER <processNo>,<start>,<count> - Batch send UD ledger updates (e.g. TESTLEDGER 3,1,50)\r\n"
                 "LOG=<bool>       - Set the LOG\r\n"
                 "MAX=<int>       - Set the MAX_ITEMS\r\n"
                 "dir               - Execute rocketmq_test_getdir\r\n"
@@ -315,6 +316,43 @@ extern bool normalOutputEnabled;
                 "exit             - Exit the shell\r\n"
                 "help             - Show this help message\r\n";
             sendStr(clientFD, "\r\x1B[K" + helpText);
+        } else if (cmd.find("TESTLEDGER") == 0) {
+            // 支持：TESTLEDGER 3,1,50（中间允许空格）
+            size_t pos = cmd.find(' ');
+            if (pos == std::string::npos || pos + 1 >= cmd.size()) {
+                sendStr(clientFD, "\r\x1B[KUsage: TESTLEDGER <processNo>,<start>,<count>\r\n");
+            } else {
+                std::string args = cmd.substr(pos + 1);
+
+                // 去掉可能的空格
+                for (size_t i = 0; i < args.size(); ) {
+                    if (args[i] == ' ' || args[i] == '\t' || args[i] == '\r' || args[i] == '\n')
+                        args.erase(i, 1);
+                    else
+                        ++i;
+                }
+
+                // 解析三个整数
+                int processNo = 0, start = 0, count = 0;
+                size_t c1 = args.find(',');
+                size_t c2 = (c1 == std::string::npos) ? std::string::npos : args.find(',', c1 + 1);
+
+                if (c1 == std::string::npos || c2 == std::string::npos) {
+                    sendStr(clientFD, "\r\x1B[KUsage: TESTLEDGER <processNo>,<start>,<count> (e.g. TESTLEDGER 3,1,50)\r\n");
+                } else {
+                    processNo = std::atoi(args.substr(0, c1).c_str());
+                    start     = std::atoi(args.substr(c1 + 1, c2 - (c1 + 1)).c_str());
+                    count     = std::atoi(args.substr(c2 + 1).c_str());
+
+                    if (processNo <= 0 || start <= 0 || count <= 0) {
+                        sendStr(clientFD, "\r\x1B[KInvalid args. Need >0. Example: TESTLEDGER 3,1,50\r\n");
+                    } else {
+                        // 调用批量发送（每条间隔1秒的逻辑在函数内部）
+                        rocketmq_test_ud_batch(m_front, processNo, start, count);
+                        sendStr(clientFD, "\r\x1B[KExecuted TESTLEDGER batch send\r\n");
+                    }
+                }
+            }
         } else if (cmd.find("viewlog") == 0) {
             showinshellflag = true;
             handleViewLogCommand(cmd, clientFD);