Python中捕获自定义RPC异常需区分gRPC与HTTP协议:gRPC须显式捕获grpc.RpcError并检查code()和details(),HTTP RPC需校验status_code且解析响应体code字段;重试时应按错误类型决策,网络类可重试,客户端错误绝不重试,服务端错误视业务而定,并必须携带idempotency_key。

Python中捕获自定义RPC异常的常见错误现象
直接用 except Exception: 试图兜底,结果发现 RPC 失败时抛出的是 RPCError 或 GrpcError 这类非标准异常,根本 catch 不到;或者用了 except grpc.RpcError: 却漏掉了服务端返回的业务级错误码(比如 status_code=16 表示权限不足),误判为网络失败。
区分 gRPC 与自定义协议的异常处理路径
gRPC 场景下,必须显式 import 并捕获 grpc.RpcError,它不是 Exception 的子类;而基于 HTTP+JSON 的自定义 RPC(如用 requests.post() 调用),错误通常藏在响应体里——比如返回 {"code": 4001, "message": "token expired"},需要手动解析 response.json() 并检查 code 字段。
- gRPC:优先检查
exception.code()和exception.details(),不要依赖str(exception) - HTTP RPC:必须校验
response.status_code(如 500/400)且解析 body 中的code字段,二者缺一不可 - 若用
aiohttp或httpx,注意异步异常仍需try/except包裹await行为,不能只包外层协程
从错误信息中提取可操作的业务状态码
很多自定义 RPC 把真实错误码塞在响应头(如 X-Error-Code: 2001)、或响应体的嵌套字段(如 data.error.code),而不是顶层 code。硬编码解析逻辑容易崩,建议封装一个统一的 parse_rpc_error() 函数:
def parse_rpc_error(resp):
if hasattr(resp, 'code') and callable(getattr(resp, 'code')):
# gRPC case
return resp.code(), resp.details()
elif hasattr(resp, 'json'):
# requests.Response
try:
body = resp.json()
return body.get('code', resp.status_code), body.get('message', '')
except:
return resp.status_code, 'invalid json response'
return 500, 'unknown error source'
关键点:不假设结构,用 hasattr 和 getattr 做类型探测,避免 AttributeError 二次崩溃。
立即学习“Python免费学习笔记(深入)”;
重试逻辑中如何避免重复提交副作用
对幂等性没保障的 RPC(如 CreateOrder),在捕获 grpc.StatusCode.UNAVAILABLE 后盲目重试,可能造成订单创建两次。必须结合错误类型做决策:
- 网络类错误(
UNAVAILABLE,DEADLINE_EXCEEDED)可重试 - 客户端错误(
INVALID_ARGUMENT,NOT_FOUND, 或 HTTP 4xx)绝不重试 - 服务端错误(
INTERNAL,UNKNOWN, 或 HTTP 5xx)视业务容忍度决定是否重试 - 所有重试必须带唯一
request_id或idempotency_key请求头,由服务端做去重
真正麻烦的是那些没文档说明错误码语义的私有 RPC,这时候只能靠抓包看实际返回、或翻服务端日志反推——别信接口文档里写的“code=0 表示成功”,生产环境往往早被改过三次了。


















