feat(pymsg): RpcClient 新增 retry_on_refused 阻塞等待模式，解决微服务启动顺序依赖
pymsg 客户端: - MsgClient.call() 新增 retry_on_refused/retry_interval 参数 - retry_on_refused=True 时，ConnectionRefusedError 不抛出，按 retry_interval(默认1s)阻塞重试 - 重试期间每5秒向 stderr 打印等待提示，避免日志刷屏 - 仅对 ConnectionRefusedError 重试；TimeoutError/ConnectionError/RpcError 仍立即抛出 - 默认 retry_on_refused=False，完全向后兼容 - RpcClient 构造函数新增同名关键字参数，call() 支持单次调用覆盖 - 新增 time 模块导入 - 更新 README 文档（新增"启动等待模式"章节和参数表） 各服务（db/dnfs/static）: - _fetch_from_confs() 改用 retry_on_refused=True，移除手动 ConnectionRefusedError 捕获和 sys.exit - 保留 RpcError（如配置文件不存在）的友好错误处理和 sys.exit(1) - 启动时不再因 confs 未就绪而退出：pymsg 自动阻塞等待 confs 启动后自动连接成功 - 启动顺序从"刚性要求 confs 先启动"变为"不严格要求启动顺序" - log_client 保持默认 retry_on_refused=False，后台线程不阻塞（fail-safe 丢弃日志） - 更新各服务 README 和 confs README 的启动顺序描述 测试验证: - 基本 call() 正常工作 - retry_on_refused=False 连接拒绝立即抛 ConnectionRefusedError - retry_on_refused=True 阻塞等待（延迟1.5s启动服务后成功连接，重试日志正常输出）
This commit is contained in:
qing committed
1 parent
0ae107f953
commit
101aed48be
10 files changed
+141
-84
No files matched your search
+31
-1
@@ -110,6 +110,20 @@ resp = client.call('{"action": "select", "table": "Users"}', timeout=10.0)
|
||||
|
||||
目标不可达时抛出 `ConnectionRefusedError`，超时无回复时抛出 `TimeoutError`。
|
||||
|
||||
**启动等待模式（retry_on_refused）**：
|
||||
|
||||
```python
|
||||
# 服务启动时阻塞等待下游服务就绪，连接被拒绝时自动重试
|
||||
resp = client.call('{"_route": "get", "name": "db.ini"}',
|
||||
timeout=5.0,
|
||||
retry_on_refused=True, # 连接拒绝时阻塞重试
|
||||
retry_interval=1.0) # 重试间隔 1 秒（默认）
|
||||
```
|
||||
|
||||
`retry_on_refused=True` 时，`ConnectionRefusedError` 不会抛出，而是按 `retry_interval`（默认 1.0 秒）间隔不断重试，直到连接成功。重试期间每 5 秒向 stderr 打印一次等待提示（如 `[pymsg] 等待服务 127.0.0.1:8010 就绪... (已重试 N 次)`），避免日志刷屏。仅对 `ConnectionRefusedError` 重试；`TimeoutError`、`ConnectionError`、业务错误等仍立即抛出。
|
||||
|
||||
适用场景：微服务启动顺序不确定时，服务进程可在不依赖外部进程管理器的情况下自动等待下游依赖就绪，无需手动 sys.exit + 外部重启。`log_client` 等 fail-safe 场景应保持默认 `retry_on_refused=False`，避免后台线程阻塞。
|
||||
|
||||
**3. 长连接监听（回调模式）**
|
||||
|
||||
```python
|
||||
@@ -226,7 +240,23 @@ except RpcError as e:
|
||||
... # 服务端业务错误（success=false）
|
||||
```
|
||||
|
||||
`call(route, payload=None, *, timeout=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。
|
||||
| 构造参数 | 含义 |
|
||||
|----------|------|
|
||||
| `addr` / `port` | 服务端地址与端口 |
|
||||
| `name` | 客户端标识（透传 `MsgClient`） |
|
||||
| `timeout` | 默认调用超时秒数（默认 `10.0`） |
|
||||
| `retry_on_refused` | 连接被拒绝时是否阻塞重试（关键字参数，默认 `False`） |
|
||||
| `retry_interval` | 重试间隔秒数（关键字参数，默认 `1.0`） |
|
||||
|
||||
`call(route, payload=None, *, timeout=None, retry_on_refused=None, retry_interval=None)`：自动注入 `_route`、JSON 编解码、检查 `success`，返回 `data` 字段；业务错误抛 `RpcError`，网络故障仍抛 `ConnectionRefusedError` / `TimeoutError`（来自 `MsgClient`）。`retry_on_refused` 和 `retry_interval` 参数可在单次调用时覆盖构造时的设置。
|
||||
|
||||
**启动等待示例**（解决微服务启动顺序问题）：
|
||||
|
||||
```python
|
||||
# confs 客户端：启动时阻塞等待 confs 就绪，不依赖启动顺序
|
||||
rpc = RpcClient("127.0.0.1", 8010, "db", timeout=5, retry_on_refused=True)
|
||||
data = rpc.call("get", {"name": "db.ini"}) # confs 未启动时自动阻塞等待
|
||||
```
|
||||
|
||||
## 实际用例
|
||||
|
||||
|
||||
+37
-14
@@ -22,6 +22,7 @@ import select
|
||||
import socket as _socket
|
||||
import struct
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
@@ -200,34 +201,56 @@ class MsgClient:
|
||||
)
|
||||
|
||||
# ---- 请求-应答（短连接）------------------------------------------------
|
||||
def call(self, msg: str, *, timeout: "float | None" = None) -> str:
|
||||
def call(self, msg: str, *, timeout: "float | None" = None,
|
||||
retry_on_refused: bool = False,
|
||||
retry_interval: float = 1.0,
|
||||
_log_every: float = 5.0) -> str:
|
||||
"""发送一条消息并阻塞等待单条回复（短连接：connect → send → recv → close）。
|
||||
|
||||
适合请求-应答模式，如 RPC 调用。不会给消息添加 name 前缀。
|
||||
|
||||
Args:
|
||||
msg: 要发送的原始字符串。
|
||||
timeout: 超时秒数，默认 10 秒。
|
||||
msg: 要发送的原始字符串。
|
||||
timeout: 超时秒数，默认 10 秒。
|
||||
retry_on_refused: 目标服务未启动（ConnectionRefusedError）时是否阻塞重试，
|
||||
直到连接成功。适用于启动顺序不确定的服务发现场景。
|
||||
默认 False（连接拒绝立即抛异常）。
|
||||
retry_interval: 重试间隔秒数，默认 1.0。
|
||||
_log_every: 重试期间打印等待提示的间隔（秒），默认每 5 秒打印一次，避免刷屏。
|
||||
|
||||
Returns:
|
||||
str: 服务器返回的字符串。
|
||||
|
||||
Raises:
|
||||
ConnectionRefusedError: 目标服务不可达。
|
||||
ConnectionRefusedError: 目标服务不可达且 ``retry_on_refused=False``。
|
||||
TimeoutError: 等待响应超时。
|
||||
ConnectionError: 连接在收全数据前断开。
|
||||
"""
|
||||
_tout = timeout if timeout is not None else 10.0
|
||||
try:
|
||||
with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as s:
|
||||
s.settimeout(_tout)
|
||||
s.connect((self.addr, self.port))
|
||||
send(s, msg)
|
||||
return recv(s, timeout=_tout)
|
||||
except ConnectionRefusedError:
|
||||
raise ConnectionRefusedError(
|
||||
f"Could not connect to {self.addr}:{self.port}"
|
||||
)
|
||||
_last_log = 0.0
|
||||
_attempt = 0
|
||||
while True:
|
||||
try:
|
||||
with _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) as s:
|
||||
s.settimeout(_tout)
|
||||
s.connect((self.addr, self.port))
|
||||
send(s, msg)
|
||||
return recv(s, timeout=_tout)
|
||||
except ConnectionRefusedError:
|
||||
if not retry_on_refused:
|
||||
raise ConnectionRefusedError(
|
||||
f"Could not connect to {self.addr}:{self.port}"
|
||||
)
|
||||
_attempt += 1
|
||||
now = time.monotonic()
|
||||
if now - _last_log >= _log_every or _attempt == 1:
|
||||
print(
|
||||
f"[pymsg] 等待服务 {self.addr}:{self.port} 就绪... "
|
||||
f"(已重试 {_attempt} 次)",
|
||||
file=__import__('sys').stderr,
|
||||
)
|
||||
_last_log = now
|
||||
time.sleep(retry_interval)
|
||||
|
||||
# ---- 持续监听（阻塞当前线程，回调模式）-------------------------------
|
||||
def listen(self, on_message: "callable") -> None:
|
||||
|
||||
+19
-3
@@ -36,21 +36,33 @@ class RpcClient:
|
||||
|
||||
rpc = RpcClient("127.0.0.1", 8005, "dnfs")
|
||||
data = rpc.call("execute", {"action": "select", "table": "Users"})
|
||||
|
||||
# 启动时阻塞等待服务就绪（连接被拒绝时自动重试）
|
||||
rpc = RpcClient("127.0.0.1", 8010, "static", retry_on_refused=True)
|
||||
"""
|
||||
|
||||
def __init__(self, addr: str, port: int, name: str = "rpc",
|
||||
timeout: float = 10.0) -> None:
|
||||
timeout: float = 10.0,
|
||||
*, retry_on_refused: bool = False,
|
||||
retry_interval: float = 1.0) -> None:
|
||||
self._client = MsgClient(addr, port, name)
|
||||
self._timeout = timeout
|
||||
self._retry_on_refused = retry_on_refused
|
||||
self._retry_interval = retry_interval
|
||||
|
||||
def call(self, route: str, payload: "dict | None" = None, *,
|
||||
timeout: "float | None" = None):
|
||||
timeout: "float | None" = None,
|
||||
retry_on_refused: "bool | None" = None,
|
||||
retry_interval: "float | None" = None):
|
||||
"""调用指定路由，返回响应中的 ``data`` 字段。
|
||||
|
||||
Args:
|
||||
route: 路由名（注入为 ``_route`` 字段）。
|
||||
payload: 业务字段 dict（``_route`` 键会被覆盖）。
|
||||
timeout: 本次调用的超时秒数，缺省用构造时的值。
|
||||
retry_on_refused: 本次调用是否在连接被拒绝时阻塞重试，
|
||||
None 表示使用构造时的值（默认 False）。
|
||||
retry_interval: 本次调用的重试间隔，None 表示使用构造时的值。
|
||||
|
||||
Returns:
|
||||
服务端返回的 ``data``（可能为 None）。
|
||||
@@ -58,13 +70,17 @@ class RpcClient:
|
||||
Raises:
|
||||
RpcError: 服务端返回 success=false，或响应非法。
|
||||
pymsg.TimeoutError: 等待响应超时。
|
||||
ConnectionRefusedError: 目标服务不可达。
|
||||
ConnectionRefusedError: 目标服务不可达且未开启重试。
|
||||
"""
|
||||
req = dict(payload or {})
|
||||
req["_route"] = route
|
||||
_retry = self._retry_on_refused if retry_on_refused is None else retry_on_refused
|
||||
_interval = self._retry_interval if retry_interval is None else retry_interval
|
||||
resp = self._client.call(
|
||||
json.dumps(req, ensure_ascii=False),
|
||||
timeout=self._timeout if timeout is None else timeout,
|
||||
retry_on_refused=_retry,
|
||||
retry_interval=_interval,
|
||||
)
|
||||
try:
|
||||
obj = json.loads(resp)
|
||||
|
||||
Reference in new issue
Block a user