RPC 模块面试讲解稿
基于 skynet 自研的分布式 RPC 框架，支持同步调用、异步通知、超时控制、熔断保护。
一、一句话定位（15 秒开场）
我基于 skynet actor 模型自研了一套分布式 RPC 框架，支持同步调用、异步通知、超时控制、熔断保护。业务代码只需
rpc.call("svc", "method", args...)，框架自动完成跨节点路由和 relay 中继。核心优化包括零 fork 协程开销、自定义协议通道、二进制安全的 body 序列化。
二、架构设计（1 分钟）
2.1 三层结构
┌─────────────────────────────────────────────────────────┐
│ 业务层 │
│ rpc.call("match_mgr", "create_battle", players...) │
└───────────────────────┬─────────────────────────────────┘
│
┌───────────────┴───────────────┐
│ │
目标在本节点 目标在其他节点
local_call() remote_call()
│ │
▼ ▼
┌───────────────┐ ┌──────────────────┐
│ relay_agent │ │ relaysvr │
│ (节点代理) │ │ (集群路由) │
│ │ │ │
│ 分派到 handler │ │ 查路由表 → 转发 │
└───────┬───────┘ └────────┬─────────┘
│ │
└───────────┬───────────────────┘
▼
┌──────────┐
│ handler │
└──────────┘
2.2 各层职责
| 层 | 文件 | 职责 |
|---|---|---|
| rpc.lua | lualib/rpc.lua | 业务入口，路由决策，Token 生命周期管理，熔断 |
| relaysvr | service/relaysvr.lua | 集群路由表，service_name → node_name 查找 |
| relay_agent | service/relay_agent.lua | 节点代理，service_name → handle 转发，响应回传 |
2.3 调用路径（完整链路）
本地调用（同节点）：
rpc.call("chat_mgr", "broadcast", msg)
→ resolve("chat_mgr") → 查到本地 handle
→ local_call(handle, "broadcast", timeout, args)
→ 有超时: skynet.send(relay_agent, "dispatch_call_async", token, packstring(args))
→ relay_agent → skynet.unpack → skynet.call(handler, "broadcast", ...)
→ 结果通过 skynet.send("snrpc", token, packed) 回传
→ snrpc dispatch → rpc_mgr:Response(token, packed)
→ skynet.wakeup → WaitResponse 返回
→ 无超时: skynet.call(handler, "lua", "broadcast", args...) 直连，零中转
远程调用（跨节点，零 fork 路径）：
rpc.call("match_mgr", "create_battle", players)
→ resolve("match_mgr") → 未命中（不在本节点）
→ remote_call("match_mgr", "create_battle", timeout, args)
→ skynet.send(relaysvr, "route_call_async", node_name, self(), ...)
→ relaysvr 查路由表: "match_mgr" → "battle" 节点
→ cluster.send("battle", "@relay_agent", "dispatch_call_async_remote", ...)
→ relay_agent(battle) → skynet.unpack → skynet.call(match_mgr, ...)
→ 结果: cluster.send(源节点, "@relay_agent", "__rpc_ret__", caller_addr, token, packed)
→ relay_agent(源节点) → skynet.send(caller_addr, "snrpc", token, packed)
→ snrpc dispatch → rpc_mgr:Response(token, packed)
→ skynet.wakeup → WaitResponse 返回
降级路径（纯调用节点，无 relay_agent）：
rpc.call(...)
→ node_name == nil → 回退 fork 路径
→ skynet.fork → skynet.call(relaysvr, "route_call", svc, method, header, body_str)
→ relaysvr 同步转发 → 等待结果返回
三、核心设计 & 为什么这样做
3.1 Token 生命周期管理（RpcManager）
问题： skynet 原生 skynet.call 的超时是全协程级别的。一旦超时，整个协程被销毁，无法区分"哪个 RPC 调用超时了"，更无法做精细化的超时告警和统计。
方案： 自研 RpcManager，为每个调用分配一个唯一 token。
Token 编码： (seq << 3) | rpc_type（自增序列号 << 3 + 调用类型低 3 位），支持 4 种 RPC 类型。
NewRpc(rpc_type, timeout)
├─ 分配 token: (seq << 3) | rpc_type
├─ timer.once(timeout, cb) ← 超时回调
└─ 返回 token
Response(token, packed_reply)
├─ 查 tokens[token] 状态
├─ 未超时 (true) → 存储 rsps[token] + skynet.wakeup
└─ 已超时 (false) → 丢弃（迟到响应）
WaitResponse(token)
├─ skynet.wait("rpc_wait_"..token) ← 阻塞当前协程
├─ 被 skynet.wakeup 唤醒
└─ RemoveRpc(token) → 返回 (Error.OK, result) 或 (Error.Timeout)
关键洞察：超时 vs 正常响应是竞速关系
时间线:
0ms ── 发送请求 ── 注册超时回调
...
500ms ── 超时触发 → tokens[token] = false → skynet.wakeup → WaitResponse 返回 Timeout
500ms ── 响应到达 → 发现 tokens[token] == false → 丢弃 ← "迟到响应被优雅处理"
或者:
300ms ── 响应到达 → tokens[token] = true → rsps[token] = reply → skynet.wakeup
500ms ── 超时触发 → tokens[token] == nil (已被 RemoveRpc) → 无事发生
两种竞速结果都是正确的，没有协程泄漏，没有内存泄漏。
3.2 零 fork 优化
问题： 初版实现每次 RPC 调用都 skynet.fork(function() skynet.call(...) end)，高并发下协程创建开销不能忽略。假设 RPC 超时 5 秒，每秒 1000 个调用 → 峰值 5000 个协程。skynet 的协程有独立的 Lua stack，每个至少几个 KB。
方案： 不创建协程。把"调用 handler → 回传结果"的职责下放给 relay_agent（独立 skynet 服务）。
改前（每调用多一个 fork 协程）:
┌─ 主协程 ──────────────────────────┐
│ local_call(...) │
│ skynet.fork(function() │ ← 协程 A (额外开销)
│ skynet.call(handler, ...) │
│ rpc_mgr:Response(token, ...) │
│ end) │
│ return rpc_mgr:WaitResponse() │ ← 主协程阻塞等待
└───────────────────────────────────┘
改后（零额外协程）:
┌─ 主协程 ───────────────────────────────────────┐ ┌─ relay_agent 服务 ─┐
│ local_call(...) │ │ │
│ skynet.send(relay_agent, "dispatch_call_async")│ │ skynet.call(handler)│
│ return rpc_mgr:WaitResponse() │ │ skynet.send(snrpc) │
└────────────────────────────────────────────────┘ └────────────────────┘
↑ 主协程阻塞，零额外协程创建 ↑ relay_agent 用自己的协程
收益： 整个 RPC 调用只消耗调用方自己一个协程。relay_agent 是独立的 skynet 服务，它的协程是所有调用方共享的，复用率极高。
3.3 snrpc 自定义协议
问题： 老方案用 skynet.send(addr, "lua", "__rpc_resp__", token, data)。skynet 的 “lua” 协议是共享通道——所有 lua 消息走同一个 dispatch handler。收到消息后要解析 cmd 字符串做比对：
-- 老方案
skynet.dispatch("lua", function(session, source, cmd, ...)
if cmd == "__rpc_resp__" then -- 每个消息都要比对
rpc_mgr:Response(...)
elseif cmd == "do_match" then
...
end
end)
方案： 利用 skynet 的内核能力，注册独立协议类型。
skynet.register_protocol {
name = "snrpc",
id = 13,
pack = skynet.pack,
unpack = skynet.unpack,
dispatch = function(session, source, token, packed_reply)
rpc_mgr:Response(token, packed_reply)
end,
}
-- 发送
skynet.send(addr, "snrpc", token, packed)
收益：
- skynet 内核按协议号 13 直接路由，不需要字符串比对，O(1)
- RPC 响应和业务消息完全隔离，互不干扰
- 框架自动注册在
rpc.init()中，业务代码无感
3.4 packstring 序列化 body
问题： table.unpack(args) 把 body 拆成 N 个独立参数穿过 relay 链。二进制数据（protobuf/sproto 编码的 string）作为独立参数传递时，经过多次 pack/unpack 可能存在隐含问题。
改前:
发端: table.unpack(args) → N 个独立参数
relaysvr: ... → 透传 N 个参数
relay_agent: ..., arg1, arg2, ... → 接收 N 个参数
handler: 收到 N 个参数
改后:
发端: skynet.packstring(table.unpack(args)) → 1 个不透明 blob
relaysvr: ... → 透传 1 个参数（零改动！）
relay_agent: body_str → 接收 1 个参数
handler: skynet.unpack(body_str) → 还原 N 个参数
为什么 relaysvr 不需要改动？ 因为它的签名是 function CMD.route_call(svc, method, ...)，... 自动适配参数数量的变化。
3.5 三态熔断器
问题： 当远程服务宕机或持续超时时，调用方不应无限重试，否则会浪费协程、内存和网络资源。
方案： 实现标准的三态熔断器。
CLOSED ──连续失败 N 次──▶ OPEN ──冷却 T 秒──▶ HALF_OPEN
▲ │
└──────────── 探测成功 ────────────────────────┘
探测失败 → 回到 OPEN
- CLOSED： 正常放行，记录成败
- OPEN： 直接拒绝（
allow() → false），不发起调用 - HALF_OPEN： 放行一次探测请求，成功则 CLOSED，失败则 OPEN
仅对远程调用生效，本地调用不经过熔断器。
3.6 本地优先路由
function M.call(...)
local handle = resolve(service_name) -- 先查本地
if handle then
return local_call(...) -- 同节点直连
end
return remote_call(...) -- 走 relay 中继
end
同节点 skynet.call 是纯内存消息传递，延迟远低于跨节点 relay。
四、使用的 Skynet 核心接口（附说明）
消息通信
| 接口 | 作用 |
|---|---|
skynet.call(addr, "lua", cmd, ...) | 同步阻塞调用，等 skynet.ret() 返回。用于 relay_agent → handler |
skynet.send(addr, type, ...) | 异步发送，不等待返回。用于零 fork 发包 + snrpc 响应回传 |
skynet.ret(pack(...)) | 给 caller 返回结果。用于 relay_agent 返回 handler 的返回值 |
cluster.proxy("node@svc") | 创建跨节点远程代理地址 |
cluster.send(node, "@svc", ...) | 跨节点异步发送。用于 relaysvr → relay_agent 转发 |
cluster.register(name, addr) | 注册服务到 skynet 集群 |
cluster.call(node, addr, ...) | 跨节点同步调用（降级路径使用） |
协程控制
| 接口 | 作用 |
|---|---|
skynet.wait(key) | 挂起当前协程，直到 skynet.wakeup(key) 被调用。用于 WaitResponse 阻塞 |
skynet.wakeup(key) | 唤醒挂起在对应 key 上的协程。响应到达 / 超时时调用 |
skynet.fork(func) | 创建新协程。仅在降级路径使用（纯调用节点无 relay_agent） |
skynet.self() | 获取当前服务地址。告诉 relay_agent “响应回传给我” |
定时器
| 接口 | 作用 |
|---|---|
skynet.timeout(cs, cb) | 延迟 cs 个 10ms 单位后执行回调。timer 模块的底层依赖 |
timer.once(cs, cb) | 一次性定时器。为每个 RPC token 注册超时回调 |
timer.every(cs, cb) | 周期定时器。周期上报路由表 |
序列化
| 接口 | 作用 |
|---|---|
skynet.pack(...) | 多值序列化为字符串。响应打包：pack(ok, result) |
skynet.unpack(str) | 反序列化。回包解包 + relay_agent 解包 body |
skynet.packstring(...) | 序列化为纯字符串（与 pack 同类）。body 打包，不拆散穿过 relay 链 |
协议 & 服务管理
| 接口 | 作用 |
|---|---|
skynet.register_protocol{name,id,...} | 注册自定义协议。snrpc 协议 id=13 |
skynet.dispatch(type, handler) | 注册协议处理器。snrpc 消息直连 rpc_mgr:Response |
skynet.uniqueservice(name) | 获取或创建全局唯一服务。启动 relay_agent |
skynet.localname(name) | 查找本地已注册服务 |
skynet.register(name) | 注册当前服务名称 |
skynet.now() | 获取当前时间（10ms 精度）。熔断器冷却计时 |
五、面试问答 Q&A
Q1: RPC 超时怎么实现的？
用 skynet 原生的协程挂起/唤醒机制。每个 RPC 调用分配一个 token，启动时注册
timer.once超时回调。调用方通过skynet.wait(token_key)挂起。超时回调和正常响应都调用skynet.wakeup(token_key)唤醒。两者是竞速关系——先到者注入结果，迟到者检测到 token 已被清理则直接丢弃。
Q2: 为什么不用 skynet.call 自带的超时？
skynet.call的超时是销毁协程级别的，无法按单个 RPC 调用做精细化控制。比如一个协程里先后调了 A 和 B 两个服务，如果 A 超时协程就被杀了，B 永远不会执行。用自研 RpcManager 后，每个 RPC 独立超时，互不影响。
Q3: 零 fork 优化解决了什么问题？
初版每调用一次 RPC 就 fork 一个协程去等待结果。假设超时 5 秒，每秒 1000 QPS → 峰值 5000 个协程。skynet 的协程有独立 Lua stack，每个至少几 KB。改成 skynet.send + 定时器 + WaitResponse 后，整个调用只消耗调用方自己的一个协程，零额外创建。
Q4: snrpc 和 lua 协议有什么区别？
lua 协议是共享通道，所有消息走同一个 dispatch，收到后要解析 cmd 字符串。snrpc 是独立协议类型（id=13），skynet 内核直接按协议号路由，不需要字符串比对。另外语义上也更清晰——snrpc 就是 RPC 响应通道，和业务消息完全隔离。
Q5: 跨节点 RPC 的响应是怎么回来的？
分两步：目标节点的 relay_agent 执行完 handler 后，通过
cluster.send(源节点, "@relay_agent", "__rpc_ret__", caller_addr, token, packed)把结果发回源节点的 relay_agent。源节点 relay_agent 再通过skynet.send(caller_addr, "snrpc", token, packed)本地转发给调用方。调用方的 snrpc dispatch 直接路由到 RpcManager:Response，然后 skynet.wakeup 唤醒。
Q6: 熔断器为什么设计成三态？
二态（开/关）有 hysteresis 问题：恢复正常后如果立即全量放行，可能再次冲垮。三态（CLOSED/OPEN/HALF_OPEN）在检测到恢复后先放行一个探测请求，成功才完全关闭熔断器，让系统平滑恢复。
Q7: 协议头（header）里传了什么？
来源集群名（src_cluster）、命令名（cmd，即方法名）、请求 token、元数据（meta_data）、链路追踪上下文（trace_ctx）、回调参数（cb_args）。目前 header 通过 skynet 的 table 序列化透传，主要用于错误诊断——handler 出错时 relay_agent 从 header 里提取来源信息打日志。
六、完整接口清单（参考用）
消息通信:
skynet.call → 同步调用
skynet.send → 异步发送
skynet.ret → 返回值
cluster.proxy → 跨节点代理
cluster.send → 跨节点异步
cluster.call → 跨节点同步（降级）
cluster.register → 集群注册
协程控制:
skynet.wait → 协程挂起
skynet.wakeup → 协程唤醒
skynet.fork → 创建协程（仅降级）
skynet.self → 获取自身地址
定时器:
skynet.timeout → 基础定时器回调
timer.once → 一次性定时器
timer.every → 周期定时器
序列化:
skynet.pack → 多值序列化
skynet.unpack → 反序列化
skynet.packstring → body 序列化
协议 & 管理:
skynet.register_protocol → 自定义协议注册
skynet.dispatch → 协议处理器注册
skynet.uniqueservice → 全局唯一服务
skynet.localname → 本地服务查找
skynet.register → 服务注册
skynet.now → 当前时间
七、文件索引
| 文件 | 说明 |
|---|---|
lualib/rpc.lua | RPC 核心模块（入口、路由、Token 管理） |
service/relaysvr.lua | 集群路由中继服务 |
service/relay_agent.lua | 节点代理服务 |
lualib/rpc_example.lua | 参考实现（设计蓝图） |
lualib/timer.lua | 定时器模块（timer.once / timer.every） |
lualib/circuit_breaker.lua | 三态熔断器 |
lualib/error.lua | 结构化错误类型 |
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/weixin_40453090/article/details/168869537
