refactor(pymsg): 新增AsyncRpcQueue通用异步队列，log_client基于此重构

- 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

@@ -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
+_BATCH_SIZE = 50
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
+_FLUSH_INTERVAL = 2.0
-_TIMEOUT = 5             # socket 超时（秒）
+_TIMEOUT = 5
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_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",
+    def __init__(self, service_name: str = "log",
-                 batch_size=_BATCH_SIZE,
+                 batch_size: int = _BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+                 flush_interval: float = _FLUSH_INTERVAL):
         self._service_name = service_name
-        self._batch_size = batch_size
+        self._client = RpcClient(
-        self._flush_interval = flush_interval
+            _DBV2_ADDR, _DBV2_PORT,
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
+            service_name + "-log",
-        self._stop = threading.Event()
+            timeout=_TIMEOUT,
-        self._flush_signal = threading.Event()
+        )
-        self._flush_done = threading.Event()
+        self._queue = AsyncRpcQueue(
-        self._thread = threading.Thread(target=self._flush_loop,
+            handler=self._send_batch,
-                                        daemon=True, name='log-client')
+            batch_size=batch_size,
-        self._thread.start()
+            flush_interval=flush_interval,
-        atexit.register(self._shutdown)
+            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(entry)
-            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  # 极端竞争下仍失败则静默丢弃
 
-    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）。"""
+        """手动 flush 队列中剩余的日志。"""
-        self._flush_done.clear()
+        self._queue.flush()
-        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()
 
-    def _drain_queue_to_db(self):
+    def close(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
+        """关闭客户端，flush 剩余日志。"""
-        batch = []
+        self._queue.close()
-        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 _flush_loop(self):
+    def _send_batch(self, batch: list):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 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
-
         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):
+    def _on_error(self, e: Exception, batch: list):
-        """进程退出时 flush 剩余日志。"""
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
-        self._stop.set()
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
-        self.flush()
+              file=_sys.stderr)
 
 
 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。
 
-logs = LogClient()
+logs = LogClient(service_name="confs")

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
+_BATCH_SIZE = 50
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
+_FLUSH_INTERVAL = 2.0
-_TIMEOUT = 5             # socket 超时（秒）
+_TIMEOUT = 5
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_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",
+    def __init__(self, service_name: str = "log",
-                 batch_size=_BATCH_SIZE,
+                 batch_size: int = _BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+                 flush_interval: float = _FLUSH_INTERVAL):
         self._service_name = service_name
-        self._batch_size = batch_size
+        self._client = RpcClient(
-        self._flush_interval = flush_interval
+            _DBV2_ADDR, _DBV2_PORT,
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
+            service_name + "-log",
-        self._stop = threading.Event()
+            timeout=_TIMEOUT,
-        self._flush_signal = threading.Event()
+        )
-        self._flush_done = threading.Event()
+        self._queue = AsyncRpcQueue(
-        self._thread = threading.Thread(target=self._flush_loop,
+            handler=self._send_batch,
-                                        daemon=True, name='log-client')
+            batch_size=batch_size,
-        self._thread.start()
+            flush_interval=flush_interval,
-        atexit.register(self._shutdown)
+            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(entry)
-            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  # 极端竞争下仍失败则静默丢弃
 
-    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）。"""
+        """手动 flush 队列中剩余的日志。"""
-        self._flush_done.clear()
+        self._queue.flush()
-        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()
 
-    def _drain_queue_to_db(self):
+    def close(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
+        """关闭客户端，flush 剩余日志。"""
-        batch = []
+        self._queue.close()
-        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 _flush_loop(self):
+    def _send_batch(self, batch: list):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 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
-
         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):
+    def _on_error(self, e: Exception, batch: list):
-        """进程退出时 flush 剩余日志。"""
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
-        self._stop.set()
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
-        self.flush()
+              file=_sys.stderr)
 
 
 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。
 
-logs = LogClient()
+logs = LogClient(service_name="dnfs")

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"]

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

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
+_BATCH_SIZE = 50
-_FLUSH_INTERVAL = 2.0    # 超时 2 秒强制 flush
+_FLUSH_INTERVAL = 2.0
-_TIMEOUT = 5             # socket 超时（秒）
+_TIMEOUT = 5
-_QUEUE_MAXSIZE = 10000   # 队列上限，防止后台线程异常时内存无限增长；满则丢弃最旧条目
+_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",
+    def __init__(self, service_name: str = "log",
-                 batch_size=_BATCH_SIZE,
+                 batch_size: int = _BATCH_SIZE,
-                 flush_interval=_FLUSH_INTERVAL):
+                 flush_interval: float = _FLUSH_INTERVAL):
         self._service_name = service_name
-        self._batch_size = batch_size
+        self._client = RpcClient(
-        self._flush_interval = flush_interval
+            _DBV2_ADDR, _DBV2_PORT,
-        self._queue = queue.Queue(maxsize=_QUEUE_MAXSIZE)
+            service_name + "-log",
-        self._stop = threading.Event()
+            timeout=_TIMEOUT,
-        self._flush_signal = threading.Event()
+        )
-        self._flush_done = threading.Event()
+        self._queue = AsyncRpcQueue(
-        self._thread = threading.Thread(target=self._flush_loop,
+            handler=self._send_batch,
-                                        daemon=True, name='log-client')
+            batch_size=batch_size,
-        self._thread.start()
+            flush_interval=flush_interval,
-        atexit.register(self._shutdown)
+            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(entry)
-            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  # 极端竞争下仍失败则静默丢弃
 
-    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）。"""
+        """手动 flush 队列中剩余的日志。"""
-        self._flush_done.clear()
+        self._queue.flush()
-        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()
 
-    def _drain_queue_to_db(self):
+    def close(self):
-        """排空主线程可见的队列中剩余条目，直接发送到 db。"""
+        """关闭客户端，flush 剩余日志。"""
-        batch = []
+        self._queue.close()
-        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 _flush_loop(self):
+    def _send_batch(self, batch: list):
-        """后台线程：积攒到 batch_size 或超时 flush_interval 时发送。
+        """后台线程回调：逐条 INSERT 到 db（与旧版行为一致：网络错误放弃本批剩余）。"""
-
-        外层捕获所有异常，保证后台线程在任何情况下（网络故障、序列化错误等）
-        都不会死亡，避免日志队列无限增长导致 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
-
         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):
+    def _on_error(self, e: Exception, batch: list):
-        """进程退出时 flush 剩余日志。"""
+        """异常安全网：handler 外的异常也不会崩溃线程。"""
-        self._stop.set()
+        print(f"[log_client] _send_batch 异常（已捕获，线程继续运行）: {e}",
-        self.flush()
+              file=_sys.stderr)
 
 
 # ── 全局单例 ──────────────────────────────────────────────────────────
+# 各服务的 log_client.py 副本需将 service_name 改为自己的服务名。
+# 此文件（confs/log_client.py）的 service_name 为 "confs"。
 
-logs = LogClient()
+logs = LogClient(service_name="static")