From 02127869c0cdde18c5186ce7c8789f6628f9c4ca Mon Sep 17 00:00:00 2001 From: lnk Date: Wed, 7 Jan 2026 11:02:03 +0800 Subject: [PATCH] add fun in log4 --- LFtid1056/cloudfront/code/log4.cpp | 146 +++++++++++++++++++++++-- LFtid1056/cloudfront/code/log4.h | 2 - LFtid1056/cloudfront/code/main.cpp | 2 +- LFtid1056/cloudfront/code/rocketmq.cpp | 5 +- 4 files changed, 139 insertions(+), 16 deletions(-) diff --git a/LFtid1056/cloudfront/code/log4.cpp b/LFtid1056/cloudfront/code/log4.cpp index 11cfa47..4027a36 100644 --- a/LFtid1056/cloudfront/code/log4.cpp +++ b/LFtid1056/cloudfront/code/log4.cpp @@ -125,14 +125,13 @@ bool DebugSwitch::match(const std::string& logger_name, int level, int logtype) std::map 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 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 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(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 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 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 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; // 重置倒计时 diff --git a/LFtid1056/cloudfront/code/log4.h b/LFtid1056/cloudfront/code/log4.h index 56e2e40..a74db7e 100644 --- a/LFtid1056/cloudfront/code/log4.h +++ b/LFtid1056/cloudfront/code/log4.h @@ -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))) diff --git a/LFtid1056/cloudfront/code/main.cpp b/LFtid1056/cloudfront/code/main.cpp index 35cb48d..389bb58 100644 --- a/LFtid1056/cloudfront/code/main.cpp +++ b/LFtid1056/cloudfront/code/main.cpp @@ -497,7 +497,7 @@ void Front::OnTimerThread() while (!m_IsTimerCancel) { - update_log_entries_countdown(); + update_log_entries_countdown();//日志上送倒计时 //业务超时检查 check_device_busy_timeout(); diff --git a/LFtid1056/cloudfront/code/rocketmq.cpp b/LFtid1056/cloudfront/code/rocketmq.cpp index 5e860a9..131e13b 100644 --- a/LFtid1056/cloudfront/code/rocketmq.cpp +++ b/LFtid1056/cloudfront/code/rocketmq.cpp @@ -1036,11 +1036,14 @@ 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_CODE("process",0,LOG_CODE_JSON, "主题:%s - tag:%s的实时触发消息失败", G_MQCONSUMER_TOPIC_RT.c_str(), FRONT_INST.c_str()); @@ -1057,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_CODE("process",0,LOG_CODE_COMM,"主题:%s - tag:%s的实时数据触发消息失败,装置%s不在线", G_MQCONSUMER_TOPIC_RT.c_str(),FRONT_INST.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; }