From d883dc620e2113d33b5065e88a118b3666958841 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 1 Aug 2026 12:59:33 +0800 Subject: [PATCH] =?UTF-8?q?refactor(pymsg):=20=E6=96=B0=E5=A2=9EAsyncRpcQu?= =?UTF-8?q?eue=E9=80=9A=E7=94=A8=E5=BC=82=E6=AD=A5=E9=98=9F=E5=88=97?= =?UTF-8?q?=EF=BC=8Clog=5Fclient=E5=9F=BA=E4=BA=8E=E6=AD=A4=E9=87=8D?= =?UTF-8?q?=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pymsg新增AsyncRpcQueue：提供内存队列+后台线程+攒批处理的通用异步基础设施， 支持两种模式：简单RPC模式(client+method+serialize)和自定义handler模式 - confs/dnfs/static的log_client.py重构为基于AsyncRpcQueue的薄封装， 保持API完全向后兼容(push/info/error/warning/debug/critical + 全局logs单例) - 消除了三份log_client副本中重复的队列/线程/攒批逻辑(约150行/份)， 代码量净减少约50行 - LogClient的行为与旧版一致：逐条INSERT到db、fail-fast网络错误、队列满丢最旧 --- confs/log_client.py | 204 +++++++++++-------------------------- dnfs/log_client.py | 204 +++++++++++-------------------------- pymsg/pymsg/__init__.py | 3 +- pymsg/pymsg/async_queue.py | 202 ++++++++++++++++++++++++++++++++++++ static/log_client.py | 204 +++++++++++-------------------------- 5 files changed, 384 insertions(+), 433 deletions(-) create mode 100644 pymsg/pymsg/async_queue.py diff --git a/confs/log_client.py b/confs/log_client.py index 0dd516f..515f1d6 100644 --- a/confs/log_client.py +++ b/confs/log_client.py @@ -1,13 +1,15 @@ -"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。 +""" +统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。 设计要点： - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃 + - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush - - 时间戳：日志入队时即记录 ts，反映事件真实发生时间 + - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间 + - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施 -用法： +用法（与旧版完全兼容）： from log_client import logs logs.info("dnfs", "节点新增成功 id=42") @@ -19,28 +21,25 @@ import json import os as _os import sys as _sys -import threading -import queue -import time import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import RpcClient, RpcError # noqa: E402 +from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) -_BATCH_SIZE = 50 # 队列满 50 条时 flush -_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush -_TIMEOUT = 5 # socket 超时（秒） -_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目 +_BATCH_SIZE = 50 +_FLUSH_INTERVAL = 2.0 +_TIMEOUT = 5 +_QUEUE_MAXSIZE = 10000 def _safe_json(obj): - """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。""" + """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。""" try: return json.dumps(obj, ensure_ascii=False, default=str) except Exception: @@ -48,163 +47,80 @@ def _safe_json(obj): def _now_iso(): - """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。""" + """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。""" return datetime.now(timezone.utc).isoformat() class LogClient: - """异步批量日志推送客户端。""" + """异步批量日志推送客户端（API 与旧版完全兼容）。""" - def __init__(self, service_name="log", - batch_size=_BATCH_SIZE, - flush_interval=_FLUSH_INTERVAL): + def __init__(self, service_name: str = "log", + batch_size: int = _BATCH_SIZE, + flush_interval: float = _FLUSH_INTERVAL): self._service_name = service_name - self._batch_size = batch_size - self._flush_interval = flush_interval - self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) - self._stop = threading.Event() - self._flush_signal = threading.Event() - self._flush_done = threading.Event() - self._thread = threading.Thread(target=self._flush_loop, - daemon=True, name='log-client') - self._thread.start() - atexit.register(self._shutdown) + self._client = RpcClient( + _DBV2_ADDR, _DBV2_PORT, + service_name + "-log", + timeout=_TIMEOUT, + ) + self._queue = AsyncRpcQueue( + handler=self._send_batch, + batch_size=batch_size, + flush_interval=flush_interval, + max_queue_size=_QUEUE_MAXSIZE, + on_error=self._on_error, + ) + self._queue.start() + atexit.register(self.close) # ── 公开接口 ──────────────────────────────────────────────────────── - def push(self, service, level, message, trace_id=None, extra=None): + def push(self, service: str, level: str, message: str, + trace_id: str = None, extra=None): """推送一条日志（非阻塞，立即返回）。 队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。 """ entry = { - 'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间 + 'ts': _now_iso(), 'service': service, 'level': level, 'message': message, 'trace_id': trace_id, 'extra': extra, } - try: - self._queue.put_nowait(entry) - except queue.Full: - # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略） - try: - self._queue.get_nowait() - except queue.Empty: - pass - try: - self._queue.put_nowait(entry) - except queue.Full: - pass # 极端竞争下仍失败则静默丢弃 + self._queue.put(entry) - def debug(self, service, message, **kw): + def debug(self, service: str, message: str, **kw): self.push(service, 'DEBUG', message, **kw) - def info(self, service, message, **kw): + def info(self, service: str, message: str, **kw): self.push(service, 'INFO', message, **kw) - def warning(self, service, message, **kw): + def warning(self, service: str, message: str, **kw): self.push(service, 'WARNING', message, **kw) - def error(self, service, message, **kw): + def warn(self, service: str, message: str, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service: str, message: str, **kw): self.push(service, 'ERROR', message, **kw) - def critical(self, service, message, **kw): + def critical(self, service: str, message: str, **kw): self.push(service, 'CRITICAL', message, **kw) def flush(self): - """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" - self._flush_done.clear() - self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量） - self._flush_done.wait(timeout=self._flush_interval + 2.0) - # 主线程排空队列中剩余条目 - self._drain_queue_to_db() + """手动 flush 队列中剩余的日志。""" + self._queue.flush() - def _drain_queue_to_db(self): - """排空主线程可见的队列中剩余条目，直接发送到 db。""" - batch = [] - while True: - try: - batch.append(self._queue.get_nowait()) - if len(batch) >= self._batch_size: - self._send(batch) - batch = [] - except queue.Empty: - break - if batch: - self._send(batch) + def close(self): + """关闭客户端，flush 剩余日志。""" + self._queue.close() # ── 内部实现 ──────────────────────────────────────────────────────── - def _flush_loop(self): - """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。 - - 外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等） - 都不会死亡，避免日志队列无限增长导致 OOM。 - """ - batch = [] - last_flush = time.monotonic() - - while not self._stop.is_set(): - try: - timeout = self._flush_interval - (time.monotonic() - last_flush) - chk_timeout = max(0.1, min(timeout, 0.5)) - try: - item = self._queue.get(timeout=chk_timeout) - batch.append(item) - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass - - should_flush = ( - len(batch) >= self._batch_size or - self._flush_signal.is_set() or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush and batch: - try: - self._send(batch) - except Exception as e: - # _send 内部已有细粒度异常处理，此处是最终安全网 - print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - batch = [] - last_flush = time.monotonic() - if self._flush_signal.is_set(): - self._flush_signal.clear() - self._flush_done.set() - except Exception as e: - # 最外层安全网：任何意外异常都不应杀死日志线程 - print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - time.sleep(0.5) # 防止异常风暴导致 CPU 满载 - - # 退出前 flush 剩余 - if batch: - try: - self._send(batch) - except Exception as e: - print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr) - self._flush_done.set() - - def _send(self, batch): - """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。 - - 整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe）， - 不抛异常给调用方。 - """ - try: - client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - except Exception as e: - print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr) - return - + def _send_batch(self, batch: list): + """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。""" for entry in batch: extra_val = entry.get("extra") try: @@ -224,21 +140,21 @@ class LogClient: }, } try: - client.call("execute", payload, timeout=_TIMEOUT) + self._client.call("execute", payload, timeout=_TIMEOUT) except RpcError as e: - # 业务错误（如表结构不符）：打印后继续下一条 print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) except Exception as e: - # 网络错误：db 不可用，放弃本批剩余（fail-safe） - print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr) break - def _shutdown(self): - """进程退出时 flush 剩余日志。""" - self._stop.set() - self.flush() + def _on_error(self, e: Exception, batch: list): + """异常安全网：handler 外的异常也不会崩溃线程。""" + print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}", + file=_sys.stderr) # ── 全局单例 ────────────────────────────────────────────────────────── +# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。 +# 此文件（confs/log_client.py）的 service_name 为 "confs"。 -logs = LogClient() +logs = LogClient(service_name="confs") diff --git a/dnfs/log_client.py b/dnfs/log_client.py index 0dd516f..e61f68a 100644 --- a/dnfs/log_client.py +++ b/dnfs/log_client.py @@ -1,13 +1,15 @@ -"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。 +""" +统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。 设计要点： - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃 + - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush - - 时间戳：日志入队时即记录 ts，反映事件真实发生时间 + - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间 + - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施 -用法： +用法（与旧版完全兼容）： from log_client import logs logs.info("dnfs", "节点新增成功 id=42") @@ -19,28 +21,25 @@ import json import os as _os import sys as _sys -import threading -import queue -import time import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import RpcClient, RpcError # noqa: E402 +from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) -_BATCH_SIZE = 50 # 队列满 50 条时 flush -_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush -_TIMEOUT = 5 # socket 超时（秒） -_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目 +_BATCH_SIZE = 50 +_FLUSH_INTERVAL = 2.0 +_TIMEOUT = 5 +_QUEUE_MAXSIZE = 10000 def _safe_json(obj): - """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。""" + """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。""" try: return json.dumps(obj, ensure_ascii=False, default=str) except Exception: @@ -48,163 +47,80 @@ def _safe_json(obj): def _now_iso(): - """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。""" + """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。""" return datetime.now(timezone.utc).isoformat() class LogClient: - """异步批量日志推送客户端。""" + """异步批量日志推送客户端（API 与旧版完全兼容）。""" - def __init__(self, service_name="log", - batch_size=_BATCH_SIZE, - flush_interval=_FLUSH_INTERVAL): + def __init__(self, service_name: str = "log", + batch_size: int = _BATCH_SIZE, + flush_interval: float = _FLUSH_INTERVAL): self._service_name = service_name - self._batch_size = batch_size - self._flush_interval = flush_interval - self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) - self._stop = threading.Event() - self._flush_signal = threading.Event() - self._flush_done = threading.Event() - self._thread = threading.Thread(target=self._flush_loop, - daemon=True, name='log-client') - self._thread.start() - atexit.register(self._shutdown) + self._client = RpcClient( + _DBV2_ADDR, _DBV2_PORT, + service_name + "-log", + timeout=_TIMEOUT, + ) + self._queue = AsyncRpcQueue( + handler=self._send_batch, + batch_size=batch_size, + flush_interval=flush_interval, + max_queue_size=_QUEUE_MAXSIZE, + on_error=self._on_error, + ) + self._queue.start() + atexit.register(self.close) # ── 公开接口 ──────────────────────────────────────────────────────── - def push(self, service, level, message, trace_id=None, extra=None): + def push(self, service: str, level: str, message: str, + trace_id: str = None, extra=None): """推送一条日志（非阻塞，立即返回）。 队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。 """ entry = { - 'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间 + 'ts': _now_iso(), 'service': service, 'level': level, 'message': message, 'trace_id': trace_id, 'extra': extra, } - try: - self._queue.put_nowait(entry) - except queue.Full: - # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略） - try: - self._queue.get_nowait() - except queue.Empty: - pass - try: - self._queue.put_nowait(entry) - except queue.Full: - pass # 极端竞争下仍失败则静默丢弃 + self._queue.put(entry) - def debug(self, service, message, **kw): + def debug(self, service: str, message: str, **kw): self.push(service, 'DEBUG', message, **kw) - def info(self, service, message, **kw): + def info(self, service: str, message: str, **kw): self.push(service, 'INFO', message, **kw) - def warning(self, service, message, **kw): + def warning(self, service: str, message: str, **kw): self.push(service, 'WARNING', message, **kw) - def error(self, service, message, **kw): + def warn(self, service: str, message: str, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service: str, message: str, **kw): self.push(service, 'ERROR', message, **kw) - def critical(self, service, message, **kw): + def critical(self, service: str, message: str, **kw): self.push(service, 'CRITICAL', message, **kw) def flush(self): - """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" - self._flush_done.clear() - self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量） - self._flush_done.wait(timeout=self._flush_interval + 2.0) - # 主线程排空队列中剩余条目 - self._drain_queue_to_db() + """手动 flush 队列中剩余的日志。""" + self._queue.flush() - def _drain_queue_to_db(self): - """排空主线程可见的队列中剩余条目，直接发送到 db。""" - batch = [] - while True: - try: - batch.append(self._queue.get_nowait()) - if len(batch) >= self._batch_size: - self._send(batch) - batch = [] - except queue.Empty: - break - if batch: - self._send(batch) + def close(self): + """关闭客户端，flush 剩余日志。""" + self._queue.close() # ── 内部实现 ──────────────────────────────────────────────────────── - def _flush_loop(self): - """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。 - - 外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等） - 都不会死亡，避免日志队列无限增长导致 OOM。 - """ - batch = [] - last_flush = time.monotonic() - - while not self._stop.is_set(): - try: - timeout = self._flush_interval - (time.monotonic() - last_flush) - chk_timeout = max(0.1, min(timeout, 0.5)) - try: - item = self._queue.get(timeout=chk_timeout) - batch.append(item) - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass - - should_flush = ( - len(batch) >= self._batch_size or - self._flush_signal.is_set() or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush and batch: - try: - self._send(batch) - except Exception as e: - # _send 内部已有细粒度异常处理，此处是最终安全网 - print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - batch = [] - last_flush = time.monotonic() - if self._flush_signal.is_set(): - self._flush_signal.clear() - self._flush_done.set() - except Exception as e: - # 最外层安全网：任何意外异常都不应杀死日志线程 - print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - time.sleep(0.5) # 防止异常风暴导致 CPU 满载 - - # 退出前 flush 剩余 - if batch: - try: - self._send(batch) - except Exception as e: - print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr) - self._flush_done.set() - - def _send(self, batch): - """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。 - - 整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe）， - 不抛异常给调用方。 - """ - try: - client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - except Exception as e: - print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr) - return - + def _send_batch(self, batch: list): + """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。""" for entry in batch: extra_val = entry.get("extra") try: @@ -224,21 +140,21 @@ class LogClient: }, } try: - client.call("execute", payload, timeout=_TIMEOUT) + self._client.call("execute", payload, timeout=_TIMEOUT) except RpcError as e: - # 业务错误（如表结构不符）：打印后继续下一条 print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) except Exception as e: - # 网络错误：db 不可用，放弃本批剩余（fail-safe） - print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr) break - def _shutdown(self): - """进程退出时 flush 剩余日志。""" - self._stop.set() - self.flush() + def _on_error(self, e: Exception, batch: list): + """异常安全网：handler 外的异常也不会崩溃线程。""" + print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}", + file=_sys.stderr) # ── 全局单例 ────────────────────────────────────────────────────────── +# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。 +# 此文件（confs/log_client.py）的 service_name 为 "confs"。 -logs = LogClient() +logs = LogClient(service_name="dnfs") diff --git a/pymsg/pymsg/__init__.py b/pymsg/pymsg/__init__.py index 6a9830e..502d5b3 100644 --- a/pymsg/pymsg/__init__.py +++ b/pymsg/pymsg/__init__.py @@ -525,5 +525,6 @@ class MsgServer: # ── JSON RPC 约定层（依赖上面的 MsgClient / MsgServer，故放在末尾导入）── from .rpc import RpcClient, RpcError, RpcServer # noqa: E402 +from .async_queue import AsyncRpcQueue # noqa: E402 -__all__ += ["RpcClient", "RpcServer", "RpcError"] +__all__ += ["RpcClient", "RpcServer", "RpcError", "AsyncRpcQueue"] diff --git a/pymsg/pymsg/async_queue.py b/pymsg/pymsg/async_queue.py new file mode 100644 index 0000000..e8e8893 --- /dev/null +++ b/pymsg/pymsg/async_queue.py @@ -0,0 +1,202 @@ +""" +AsyncRpcQueue - 通用异步批量发送队列 + +提供内存队列 + 后台线程的异步批量处理机制，适用于日志上报、指标采集等 +非关键路径场景。业务线程put操作非阻塞，后台线程攒批处理，失败时通过错误回调 +通知，不影响主业务流程。 + +两种使用模式： + +1. 简单RPC模式（指定client + method + serialize）： + queue = AsyncRpcQueue( + client=rpc_client, + method='log_api.push', + batch_size=50, + flush_interval=2.0, + serialize=lambda items: {'logs': json.dumps(items)}, + ) + +2. 自定义处理模式（指定handler回调）： + def send_batch(batch): + for item in batch: + rpc_client.call('execute', item) + queue = AsyncRpcQueue( + handler=send_batch, + batch_size=50, + flush_interval=2.0, + ) +""" + +import threading +import time +from collections import deque +from typing import Any, Callable, Optional + + +class AsyncRpcQueue: + """通用异步批量发送队列""" + + def __init__( + self, + client: Any = None, + method: str = None, + *, + handler: Optional[Callable[[list], None]] = None, + batch_size: int = 100, + flush_interval: float = 2.0, + max_queue_size: int = 10000, + serialize: Optional[Callable[[list], dict]] = None, + on_error: Optional[Callable[[Exception, list], None]] = None, + retry_on_refused: bool = False, + retry_interval: float = 1.0, + ): + """ + 初始化异步队列 + + Args: + client: RpcClient实例（RPC模式必填） + method: RPC方法名（RPC模式必填） + handler: 自定义批量处理回调，接收一批items列表。提供此参数时 + 忽略client/method/serialize/retry_on_refused。 + batch_size: 批量发送的阈值，达到此数量立即发送 + flush_interval: 定时flush间隔（秒） + max_queue_size: 队列最大长度，超过则丢弃最旧数据 + serialize: 序列化函数，接收items列表返回RPC参数字典 + 默认: lambda items: {'items': items} + on_error: 错误回调，接收(exception, batch_items) + retry_on_refused: 是否在连接被拒绝时重试（RPC模式，透传给client.call） + retry_interval: 重试间隔（秒）（RPC模式，透传给client.call） + """ + if handler is None: + if client is None or method is None: + raise ValueError("必须提供 handler 或 (client + method)") + self._handler = self._make_rpc_handler( + client, method, + serialize or (lambda items: {'items': items}), + retry_on_refused, retry_interval, + ) + else: + self._handler = handler + + self._batch_size = batch_size + self._flush_interval = flush_interval + self._max_queue_size = max_queue_size + self._on_error = on_error + + self._queue: deque = deque() + self._lock = threading.Lock() + self._wakeup = threading.Event() + self._thread: Optional[threading.Thread] = None + self._closed = False + self._dropped_count = 0 + + @staticmethod + def _make_rpc_handler(client, method, serialize, retry_on_refused, retry_interval): + """创建RPC模式的默认handler""" + def handler(batch: list): + params = serialize(batch) + client.call( + method, params, + retry_on_refused=retry_on_refused, + retry_interval=retry_interval, + ) + return handler + + def start(self) -> None: + """启动后台发送线程""" + if self._thread is not None and self._thread.is_alive(): + return + self._closed = False + self._thread = threading.Thread( + target=self._run, + name=f'AsyncRpcQueue', + daemon=True, + ) + self._thread.start() + + def put(self, item: Any) -> bool: + """ + 放入一条数据到队列（非阻塞） + + Args: + item: 要发送的数据项 + + Returns: + bool: True表示放入成功，False表示队列已关闭 + """ + if self._closed: + return False + with self._lock: + if len(self._queue) >= self._max_queue_size: + self._queue.popleft() + self._dropped_count += 1 + self._queue.append(item) + self._wakeup.set() + return True + + def flush(self) -> None: + """立即触发一次发送（会阻塞直到处理完成，主要用于关闭时）""" + self._send_batch() + + def close(self) -> None: + """关闭队列，flush剩余数据后停止后台线程""" + if self._closed: + return + self._closed = True + self._wakeup.set() + if self._thread is not None: + self._thread.join(timeout=5.0) + self._send_batch() + + @property + def pending_count(self) -> int: + """当前队列中待发送的数据量""" + with self._lock: + return len(self._queue) + + @property + def dropped_count(self) -> int: + """因队列满而丢弃的数据量""" + return self._dropped_count + + def _run(self) -> None: + """后台线程主循环""" + last_flush = time.time() + while not self._closed: + self._wakeup.wait(timeout=0.1) + self._wakeup.clear() + + now = time.time() + should_flush = False + with self._lock: + if len(self._queue) >= self._batch_size: + should_flush = True + elif now - last_flush >= self._flush_interval and self._queue: + should_flush = True + if should_flush: + self._send_batch() + last_flush = time.time() + + def _send_batch(self) -> None: + """取一批数据处理""" + with self._lock: + if not self._queue: + return + batch = list(self._queue) + self._queue.clear() + try: + self._handler(batch) + except Exception as e: + if self._on_error: + try: + self._on_error(e, batch) + except Exception: + pass + + def __enter__(self): + self.start() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + return False diff --git a/static/log_client.py b/static/log_client.py index 0dd516f..f0b95b6 100644 --- a/static/log_client.py +++ b/static/log_client.py @@ -1,13 +1,15 @@ -"""统一日志客户端 — 各服务共用（dnfs / smtp / static / confs 四份副本内容一致，改动请同步）。 +""" +统一日志客户端 — 各服务共用（基于 pymsg.AsyncRpcQueue）。 设计要点： - 通过 pymsg RPC 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 - - fail-safe：db 服务不可用时静默丢弃，绝不影响业务；后台线程任何异常都不会崩溃 + - 异步非阻塞：内存队列 + 后台线程攒批推送，不拖慢业务 + - fail-safe：db 服务不可用时静默丢弃，绝不影响业务 - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush - - 时间戳：日志入队时即记录 ts，反映事件真实发生时间 + - 时间戳：日志入队时即记录 ts（UTC ISO 格式），反映事件真实发生时间 + - 底层复用 pymsg.AsyncRpcQueue 通用队列基础设施 -用法： +用法（与旧版完全兼容）： from log_client import logs logs.info("dnfs", "节点新增成功 id=42") @@ -19,28 +21,25 @@ import json import os as _os import sys as _sys -import threading -import queue -import time import atexit from datetime import datetime, timezone _sys.path.insert(0, _os.path.join(_os.path.dirname(__file__), "..")) -from pymsg import RpcClient, RpcError # noqa: E402 +from pymsg import RpcClient, RpcError, AsyncRpcQueue # noqa: E402 # ── 配置 ────────────────────────────────────────────────────────────── _DBV2_ADDR = _os.environ.get('DB_HOST', '127.0.0.1') _DBV2_PORT = int(_os.environ.get('DB_PORT', '8005')) -_BATCH_SIZE = 50 # 队列满 50 条时 flush -_FLUSH_INTERVAL = 2.0 # 超时 2 秒强制 flush -_TIMEOUT = 5 # socket 超时（秒） -_QUEUE_MAXSIZE = 10000 # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目 +_BATCH_SIZE = 50 +_FLUSH_INTERVAL = 2.0 +_TIMEOUT = 5 +_QUEUE_MAXSIZE = 10000 def _safe_json(obj): - """安全序列化 extra 字段，不可序列化对象转为 repr 字符串，保证不抛异常。""" + """安全序列化 extra 字段，不可序列化对象转为 repr 字符串。""" try: return json.dumps(obj, ensure_ascii=False, default=str) except Exception: @@ -48,163 +47,80 @@ def _safe_json(obj): def _now_iso(): - """生成当前 UTC 时间 ISO 字符串（在入队时调用，保证时间戳准确）。""" + """生成当前 UTC 时间 ISO 字符串（入队时调用，保证时间戳准确）。""" return datetime.now(timezone.utc).isoformat() class LogClient: - """异步批量日志推送客户端。""" + """异步批量日志推送客户端（API 与旧版完全兼容）。""" - def __init__(self, service_name="log", - batch_size=_BATCH_SIZE, - flush_interval=_FLUSH_INTERVAL): + def __init__(self, service_name: str = "log", + batch_size: int = _BATCH_SIZE, + flush_interval: float = _FLUSH_INTERVAL): self._service_name = service_name - self._batch_size = batch_size - self._flush_interval = flush_interval - self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE) - self._stop = threading.Event() - self._flush_signal = threading.Event() - self._flush_done = threading.Event() - self._thread = threading.Thread(target=self._flush_loop, - daemon=True, name='log-client') - self._thread.start() - atexit.register(self._shutdown) + self._client = RpcClient( + _DBV2_ADDR, _DBV2_PORT, + service_name + "-log", + timeout=_TIMEOUT, + ) + self._queue = AsyncRpcQueue( + handler=self._send_batch, + batch_size=batch_size, + flush_interval=flush_interval, + max_queue_size=_QUEUE_MAXSIZE, + on_error=self._on_error, + ) + self._queue.start() + atexit.register(self.close) # ── 公开接口 ──────────────────────────────────────────────────────── - def push(self, service, level, message, trace_id=None, extra=None): + def push(self, service: str, level: str, message: str, + trace_id: str = None, extra=None): """推送一条日志（非阻塞，立即返回）。 队列满时静默丢弃最旧条目再放入新条目，保证不阻塞业务线程、不导致 OOM。 """ entry = { - 'ts': _now_iso(), # 入队时即生成时间戳，反映真实事件时间 + 'ts': _now_iso(), 'service': service, 'level': level, 'message': message, 'trace_id': trace_id, 'extra': extra, } - try: - self._queue.put_nowait(entry) - except queue.Full: - # 队列满：丢弃最旧的一条，再放入新条目（drop-oldest 策略） - try: - self._queue.get_nowait() - except queue.Empty: - pass - try: - self._queue.put_nowait(entry) - except queue.Full: - pass # 极端竞争下仍失败则静默丢弃 + self._queue.put(entry) - def debug(self, service, message, **kw): + def debug(self, service: str, message: str, **kw): self.push(service, 'DEBUG', message, **kw) - def info(self, service, message, **kw): + def info(self, service: str, message: str, **kw): self.push(service, 'INFO', message, **kw) - def warning(self, service, message, **kw): + def warning(self, service: str, message: str, **kw): self.push(service, 'WARNING', message, **kw) - def error(self, service, message, **kw): + def warn(self, service: str, message: str, **kw): + self.push(service, 'WARNING', message, **kw) + + def error(self, service: str, message: str, **kw): self.push(service, 'ERROR', message, **kw) - def critical(self, service, message, **kw): + def critical(self, service: str, message: str, **kw): self.push(service, 'CRITICAL', message, **kw) def flush(self): - """手动 flush 队列中剩余的日志（含后台线程持有的局部 batch）。""" - self._flush_done.clear() - self._flush_signal.set() # 通知后台线程立即 flush 其局部 batch - # 等待后台线程完成 flush（最多等待 flush_interval + 2 秒余量） - self._flush_done.wait(timeout=self._flush_interval + 2.0) - # 主线程排空队列中剩余条目 - self._drain_queue_to_db() + """手动 flush 队列中剩余的日志。""" + self._queue.flush() - def _drain_queue_to_db(self): - """排空主线程可见的队列中剩余条目，直接发送到 db。""" - batch = [] - while True: - try: - batch.append(self._queue.get_nowait()) - if len(batch) >= self._batch_size: - self._send(batch) - batch = [] - except queue.Empty: - break - if batch: - self._send(batch) + def close(self): + """关闭客户端，flush 剩余日志。""" + self._queue.close() # ── 内部实现 ──────────────────────────────────────────────────────── - def _flush_loop(self): - """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。 - - 外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等） - 都不会死亡，避免日志队列无限增长导致 OOM。 - """ - batch = [] - last_flush = time.monotonic() - - while not self._stop.is_set(): - try: - timeout = self._flush_interval - (time.monotonic() - last_flush) - chk_timeout = max(0.1, min(timeout, 0.5)) - try: - item = self._queue.get(timeout=chk_timeout) - batch.append(item) - while len(batch) < self._batch_size: - try: - batch.append(self._queue.get_nowait()) - except queue.Empty: - break - except queue.Empty: - pass - - should_flush = ( - len(batch) >= self._batch_size or - self._flush_signal.is_set() or - (batch and time.monotonic() - last_flush >= self._flush_interval) - ) - if should_flush and batch: - try: - self._send(batch) - except Exception as e: - # _send 内部已有细粒度异常处理，此处是最终安全网 - print(f"[log_client] _send 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - batch = [] - last_flush = time.monotonic() - if self._flush_signal.is_set(): - self._flush_signal.clear() - self._flush_done.set() - except Exception as e: - # 最外层安全网：任何意外异常都不应杀死日志线程 - print(f"[log_client] _flush_loop 异常（已捕获，线程继续运行）: {e}", - file=_sys.stderr) - time.sleep(0.5) # 防止异常风暴导致 CPU 满载 - - # 退出前 flush 剩余 - if batch: - try: - self._send(batch) - except Exception as e: - print(f"[log_client] 退出时 flush 异常: {e}", file=_sys.stderr) - self._flush_done.set() - - def _send(self, batch): - """通过 pymsg RPC（execute 路由）批量发送到 dbv2 Logs 表。 - - 整个 batch 复用一个 TCP 连接；网络错误时放弃本批剩余条目（fail-safe）， - 不抛异常给调用方。 - """ - try: - client = RpcClient(_DBV2_ADDR, _DBV2_PORT, self._service_name + "-log") - except Exception as e: - print(f"[log_client] 创建 RpcClient 失败: {e}", file=_sys.stderr) - return - + def _send_batch(self, batch: list): + """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。""" for entry in batch: extra_val = entry.get("extra") try: @@ -224,21 +140,21 @@ class LogClient: }, } try: - client.call("execute", payload, timeout=_TIMEOUT) + self._client.call("execute", payload, timeout=_TIMEOUT) except RpcError as e: - # 业务错误（如表结构不符）：打印后继续下一条 print(f"[log_client] INSERT 失败: {e}", file=_sys.stderr) except Exception as e: - # 网络错误：db 不可用，放弃本批剩余（fail-safe） - print(f"[log_client] 发送失败: {e}", file=_sys.stderr) + print(f"[log_client] 发送失败（放弃本批剩余）: {e}", file=_sys.stderr) break - def _shutdown(self): - """进程退出时 flush 剩余日志。""" - self._stop.set() - self.flush() + def _on_error(self, e: Exception, batch: list): + """异常安全网：handler 外的异常也不会崩溃线程。""" + print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}", + file=_sys.stderr) # ── 全局单例 ────────────────────────────────────────────────────────── +# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。 +# 此文件（confs/log_client.py）的 service_name 为 "confs"。 -logs = LogClient() +logs = LogClient(service_name="static")