主题
Server端是如何实现高并发的？
一句话概括
miniRPC Server 通过三级 goroutine 并发模型实现高并发：连接级并发（每个连接一个 goroutine）、请求级并发（每个请求一个 goroutine）、响应串行化（互斥锁保证写入安全），同时用 sync.WaitGroup 实现优雅关闭。
三级并发模型全景
┌─────────────────────────────────────────────────┐
│ Server.Accept │
│ (主 goroutine) │
│ for { lis.Accept() } │
└───────┬──────────────┬──────────────┬───────────┘
│ │ │
go ServeConn go ServeConn go ServeConn ← 第 1 级：连接级并发
│ │ │
┌───────▼───┐ ┌──────▼────┐ ┌─────▼─────┐
│ serveCodec │ │ serveCodec │ │ serveCodec │
│ (连接 1) │ │ (连接 2) │ │ (连接 3) │
└───┬───────┘ └───────────┘ └───────────┘
│
┌──────────────┼──────────────┐
│ │ │
go handleReq go handleReq go handleReq ← 第 2 级：请求级并发
│ │ │
▼ ▼ ▼
svc.call() svc.call() svc.call()
│ │ │
└──────────────┼──────────────┘
│
sending.Lock() ← 第 3 级：响应串行化
cc.Write(h, body)
sending.Unlock()第 1 级：连接级并发 —— 每个连接独占一个 goroutine
go
func (server *Server) Accept(lis net.Listener) {
for {
conn, err := lis.Accept()
if err != nil {
log.Println("rpc server: accept error:", err)
return
}
go server.ServeConn(conn) // 每个新连接启动一个独立 goroutine
}
}为什么这样设计
lis.Accept()是阻塞调用，如果在主 goroutine 中同步处理连接，那在处理一个客户端时其他客户端就得排队等待。- 用
go server.ServeConn(conn)把每个连接的处理交给独立 goroutine，主 goroutine 立即回到Accept等待下一个连接。 - 这意味着 N 个客户端同时连接，就有 N 个 goroutine 并行处理，互不干扰。
为什么用 goroutine 而不是线程池
Go 的 goroutine 是用户态轻量级线程，初始栈只有几 KB（而 OS 线程通常 1~8 MB），创建和切换的开销极小。即使同时有上万个连接，上万个 goroutine 也不会成为问题。这是 Go 语言天然适合高并发网络服务的根本原因。
第 2 级：请求级并发 —— 同一连接上的请求并行处理
单个连接上，serveCodec 在一个循环中串行读取请求，但并行处理请求：
go
func (server *Server) serveCodec(cc codec.Codec, opt *codec.Option) {
sending := new(sync.Mutex) // 发送锁
wg := new(sync.WaitGroup) // 等待所有请求完成
for {
req, err := server.readRequest(cc) // 串行读取
if err != nil {
if req == nil {
break
}
req.h.Error = err.Error()
server.sendResponse(cc, req.h, invalidRequest, sending)
continue
}
wg.Add(1)
go server.handleRequest(cc, req, sending, wg, opt.HandleTimeout) // 并行处理
}
wg.Wait() // 等所有 handleRequest 完成
_ = cc.Close()
}关键问题：为什么读取必须串行，处理可以并行？
读取必须串行的原因：
TCP 是字节流协议，数据在连接上是严格有序的。一条消息由 Header 和 Body 组成，如果两个 goroutine 同时从连接读数据，可能出现 goroutine A 读了 Header，goroutine B 读了本该属于 A 的 Body，整个协议就错乱了。所以 readRequest 必须在单个 goroutine 中串行执行。
TCP 连接上的字节流（严格有序）：
[Header1][Body1][Header2][Body2][Header3][Body3]...
↑
单个 goroutine 按顺序一个一个读处理可以并行的原因：
readRequest 完成后，请求的所有数据（Header、argv、replyv）已经完整地读到了内存中的 request 结构体里。此时这个请求就是一个自包含的独立单元，它的处理（调用服务方法、计算结果）跟其他请求没有数据依赖，所以完全可以放到独立 goroutine 中并行处理。
时间线：
主 goroutine： 读req1 → 读req2 → 读req3 → ... → 读完(EOF) → wg.Wait()
↓ ↓ ↓
goroutine 1： 处理req1 ────────────────────── 发响应1
goroutine 2： 处理req2 ──── 发响应2 ← 先完成的先发
goroutine 3： 处理req3 ──────────────── 发响应3并行处理的收益
假设客户端在一个连接上发了 5 个请求，每个请求处理耗时 100ms：
- 串行处理：总耗时 = 5 × 100ms = 500ms
- 并行处理：总耗时 ≈ 100ms（5 个请求同时执行）
这就是 example/main.go 中 5 个 Greeter.SayHello 调用能"几乎同时"返回的原因。
第 3 级：响应串行化 —— 互斥锁保证写入完整性
处理是并行的，但写响应必须串行，否则两个 goroutine 同时往连接写数据会导致 Header 和 Body 交错错乱：
go
func (server *Server) sendResponse(cc codec.Codec, h *codec.Header, body interface{}, sending *sync.Mutex) {
sending.Lock() // 同一时刻只能有一个 goroutine 写响应
defer sending.Unlock()
if err := cc.Write(h, body); err != nil {
log.Println("rpc server: write response error:", err)
}
}如果不加锁会怎样
无锁的灾难场景：
goroutine 1 正在写：[Header1] [Body1]
goroutine 2 同时写： [Header2] [Body2]
客户端收到的字节流：[Header1][Header2][Body1][Body2]
↑
这不是合法消息！客户端无法正确解析。加锁后的正确行为
有锁的正确行为：
goroutine 1 获得锁：[Header1][Body1]
goroutine 2 获得锁：[Header2][Body2]
客户端收到的字节流：[Header1][Body1][Header2][Body2]
↑ ↑
完整的消息 1 完整的消息 2 ✓ 可正确解析为什么用 sync.Mutex 而不是 channel
这里的需求是"互斥访问共享资源（TCP 连接）"，Mutex 语义上最直接。但用 channel 同样可以实现，核心思路是：不让多个 goroutine 直接写连接，而是把"要写的东西"发到一个 channel，由唯一一个消费者 goroutine 负责写入。
channel 实现互斥写入的原理
go
type response struct {
h *codec.Header
body interface{}
}
// channel 方案的 serveCodec
func (server *Server) serveCodec(cc codec.Codec, opt *codec.Option) {
respCh := make(chan *response, 16) // 带缓冲的 channel
wg := new(sync.WaitGroup)
// 唯一的写入者 goroutine —— 只有它能碰 cc.Write
go func() {
for resp := range respCh {
cc.Write(resp.h, resp.body) // 单 goroutine 顺序写，天然不会交叉
}
}()
for {
req, err := server.readRequest(cc)
if err != nil { break }
wg.Add(1)
go func(req *request) {
defer wg.Done()
req.svc.call(req.mtype, req.argv, req.replyv)
// 不直接写连接，而是把响应"投递"到 channel
respCh <- &response{h: req.h, body: req.replyv.Interface()}
}(req)
}
wg.Wait()
close(respCh) // 所有请求处理完后关闭 channel，写入者 goroutine 退出
cc.Close()
}为什么这样就互斥了？关键在于：
Mutex 方案：多个 goroutine 竞争同一把锁，谁抢到谁写
goroutine 1 ──Lock()──Write()──Unlock()──────────────
goroutine 2 ──────────Lock()（阻塞等待）──Write()──Unlock()──
goroutine 3 ──────────────────Lock()（阻塞等待）──Write()──Unlock()──
channel 方案：多个 goroutine 只负责"投递"，唯一的消费者负责写
goroutine 1 ──respCh <- resp1──(完事)
goroutine 2 ──respCh <- resp2──(完事)
goroutine 3 ──respCh <- resp3──(完事)
│
▼
写入者 goroutine：for resp := range respCh {
cc.Write(resp.h, resp.body) ← 永远只有这一个 goroutine 在写
}channel 方案之所以互斥，不是因为 channel 本身有锁的语义，而是因为它把"写连接"这个操作收敛到了唯一一个 goroutine 中。只有一个 goroutine 在执行 cc.Write，自然就不可能有两个 goroutine 同时写，互斥就天然成立了。这就是 Go 的经典哲学：
Don't communicate by sharing memory; share memory by communicating.
不要通过共享内存来通信（Mutex 方案），而是通过通信来共享内存（channel 方案）。
为什么 miniRPC 选了 Mutex
| 维度 | Mutex 方案 | channel 方案 |
|---|---|---|
| 额外 goroutine | 无 | 需要一个专门的写入者 goroutine |
| 内存开销 | 一个 Mutex（8 字节） | channel 结构 + 缓冲区 + goroutine 栈 |
| 代码复杂度 | 3 行（Lock/Write/Unlock） | 需要定义 response 结构体、启动/关闭 goroutine |
| 背压控制 | 无（写完就释放锁） | 有（channel 满时 goroutine 自动阻塞） |
| 适用场景 | 简单互斥 | 需要解耦生产/消费、需要排队、需要背压 |
对于 miniRPC 这种"只需要互斥写入"的简单场景，Mutex 方案代码更少、心智负担更小。channel 方案的优势在更复杂的场景中才能体现，比如需要对响应排队、限流、或者做批量合并写入时。
优雅关闭：WaitGroup 的作用
go
wg := new(sync.WaitGroup)
for {
req, err := server.readRequest(cc)
// ...
wg.Add(1) // 读到一个请求，计数 +1
go server.handleRequest(cc, req, sending, wg, opt.HandleTimeout)
}
wg.Wait() // 所有 handleRequest 完成后才继续
_ = cc.Close() // 关闭连接为什么需要 WaitGroup
当 readRequest 返回 EOF（客户端关闭连接或所有请求已发送完毕），主循环退出。但此时可能还有一些 handleRequest goroutine 正在执行中：
时间线：
主 goroutine： 读req1 → 读req2 → 读req3 → EOF退出循环
↓ ↓ ↓
goroutine 1： 处理req1 ─── 完成 ✓
goroutine 2： 处理req2 ───────── 还在处理中...
goroutine 3： 处理req3 ──────────── 还在处理中...
← 如果这里直接 cc.Close()，
goroutine 2 和 3 的响应无法发出！wg.Wait() 确保主 goroutine 阻塞等待所有 handleRequest 执行完毕（每个 handleRequest 在最后调用 defer wg.Done()），然后才关闭连接。这样不会丢失任何正在处理中的请求。
超时控制与并发的关系
handleRequest 中的超时控制也是并发设计的一部分：
go
func (server *Server) handleRequest(cc codec.Codec, req *request, sending *sync.Mutex, wg *sync.WaitGroup, timeout time.Duration) {
defer wg.Done()
if timeout == 0 {
// 不设超时，直接同步调用
err := req.svc.call(req.mtype, req.argv, req.replyv)
// ...
return
}
// 设了超时：用子 goroutine 执行，主 goroutine 监控超时
called := make(chan struct{})
sent := make(chan struct{})
go func() {
err := req.svc.call(req.mtype, req.argv, req.replyv)
close(called)
// 发送响应...
close(sent)
}()
select {
case <-time.After(timeout):
// 超时了，立即返回错误响应
req.h.Error = fmt.Sprintf("rpc server: request handle timeout: expect within %s", timeout)
server.sendResponse(cc, req.h, invalidRequest, sending)
case <-called:
<-sent // 等待响应发送完成
}
}超时控制的并发模型：
handleRequest goroutine： 子 goroutine：
│
select { │ svc.call()（可能很慢）
case <-time.After(timeout): │
// 超时，立即返回错误 │
case <-called: │ close(called) ← 调用完成
<-sent │ sendResponse
} │ close(sent) ← 发送完成- 如果
svc.call()在超时前完成 →calledchannel 关闭，走case <-called分支，正常返回结果 - 如果超时先到 → 走
case <-time.After(timeout)分支，立即给客户端返回超时错误，不再等待慢请求
并发安全分析总结
| 操作 | 并发方式 | 保护机制 | 为什么 |
|---|---|---|---|
| 接受连接 | 主 goroutine 循环 Accept | 无需保护 | Accept 本身是串行的 |
| 处理连接 | 每连接一个 goroutine | 连接间完全隔离 | 各连接有独立的 Codec、锁、WaitGroup |
| 读取请求 | 串行（单 goroutine） | 无需保护 | TCP 字节流必须按序读取 |
| 处理请求 | 并行（每请求一个 goroutine） | 请求数据自包含 | 读完后各请求互不依赖 |
| 发送响应 | sync.Mutex 互斥 | sending 锁 | 防止响应数据交叉写入 |
| 等待完成 | sync.WaitGroup | wg.Wait() | 确保关闭前所有请求处理完毕 |
| 服务注册表 | sync.Map | 无锁并发安全 | 支持并发读写服务映射 |
| 方法调用计数 | atomic.AddUint64 | 原子操作 | 无锁计数，性能最优 |
与 net/rpc 标准库的并发模型对比
miniRPC 的并发模型与 Go 标准库 net/rpc 几乎一致，都是：
Accept 循环 → 每连接 go ServeConn → 串行读 + 并行处理 + 互斥写两者的核心差异不在并发模型上，而在编解码协商和超时控制上（net/rpc 无协商机制且不支持超时）。
这个模型简洁而高效，是 Go 网络编程的经典范式——用 goroutine 替代线程池，用 channel/Mutex 做同步，用 WaitGroup 做优雅关闭。
服务端是如何实现超时控制的？
超时控制的全局位置
miniRPC 中有两处超时，都由客户端在 Option 中指定：
go
type Option struct {
MagicNumber int
CodecType Type
ConnectTimeout time.Duration // 客户端连接超时（客户端自己控制）
HandleTimeout time.Duration // 服务端处理超时（传给服务端，由服务端控制）
}ConnectTimeout：控制的是客户端建立连接 + 握手的超时，完全在客户端dialTimeout中实现，与服务端无关。HandleTimeout：控制的是服务端处理单个请求的超时，由客户端在握手时通过Option告知服务端，服务端在handleRequest中执行。
本题聚焦 HandleTimeout 在服务端的实现。
超时值的传递链路
客户端 服务端
Option{HandleTimeout: 3s}
│
│ json.Encode(Option)
└─────────────────────────────► json.Decode(&opt)
│
opt.HandleTimeout = 3s
│
serveCodec(cc, &opt)
│
go handleRequest(..., opt.HandleTimeout)
│
select + time.After(3s) 实现超时serveCodec 中的传递：
go
func (server *Server) serveCodec(cc codec.Codec, opt *codec.Option) {
// ...
for {
req, err := server.readRequest(cc)
// ...
wg.Add(1)
go server.handleRequest(cc, req, sending, wg, opt.HandleTimeout) // ← 传入超时值
}
// ...
}注意：HandleTimeout 对该连接上的每个请求都生效，是连接级别的配置，而非单个请求可以独立设置。
handleRequest 的超时实现
handleRequest 根据 timeout 是否为 0 分为两条路径：
路径 1：timeout == 0（无超时）
go
if timeout == 0 {
err := req.svc.call(req.mtype, req.argv, req.replyv) // 同步调用，阻塞到完成
if err != nil {
req.h.Error = err.Error()
server.sendResponse(cc, req.h, invalidRequest, sending)
return
}
server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
return
}直接在当前 goroutine 中同步调用服务方法。无论方法执行多久都等着，不做任何超时干预。
路径 2：timeout > 0（有超时控制）
go
called := make(chan struct{})
sent := make(chan struct{})
go func() {
err := req.svc.call(req.mtype, req.argv, req.replyv)
close(called) // ① 标记"方法执行完毕"
if err != nil {
req.h.Error = err.Error()
server.sendResponse(cc, req.h, invalidRequest, sending)
close(sent) // ② 标记"响应发送完毕"
return
}
server.sendResponse(cc, req.h, req.replyv.Interface(), sending)
close(sent) // ② 标记"响应发送完毕"
}()
select {
case <-time.After(timeout): // 超时先到
req.h.Error = fmt.Sprintf("rpc server: request handle timeout: expect within %s", timeout)
server.sendResponse(cc, req.h, invalidRequest, sending)
case <-called: // 方法执行完毕先到
<-sent // 等响应也发完
}这是整个超时控制的核心，用到了 两个 channel + select 的经典模式。
逐步拆解：两个 channel 各自的职责
为什么需要两个 channel？
一个请求的处理分为两个阶段：
阶段 1：svc.call() —— 执行服务方法（可能很耗时）
阶段 2：sendResponse() —— 写响应到连接（需要抢锁，可能等待）called 和 sent 分别标记这两个阶段的完成：
子 goroutine 的执行流程：
svc.call() ← 可能耗时很长
close(called) ← "方法调用完了"
sendResponse() ← 可能因为 sending 锁而等待
close(sent) ← "响应也发完了"select 的两个分支
select {
case <-time.After(timeout): // 分支 A：超时先触发
case <-called: // 分支 B：方法先执行完
<-sent
}分支 A — 超时先到：
时间线（timeout = 3s）：
handleRequest goroutine： 子 goroutine：
│ │
│ select { │ svc.call() ← 在慢慢执行...
│ case <-time.After(3s): │
│ // 3秒到了！方法还没完成 │
│ sendResponse(超时错误) │
│ ... │ ...（仍在执行，但已经无人关心结果）
│ } │
▼ 函数返回 ▼超时后，handleRequest 立即给客户端返回一个超时错误响应，不再等待方法执行完成。
分支 B — 方法先执行完：
时间线（timeout = 3s，方法 1s 就执行完）：
handleRequest goroutine： 子 goroutine：
│ │
│ select { │ svc.call() ← 1s 完成
│ ... │ close(called) ← 通知主 goroutine
│ case <-called: │
│ // 好的，方法执行完了 │ sendResponse(正常结果)
│ <-sent │ close(sent) ← 响应也发完了
│ // 响应也发完了，完美！ │
│ } │
▼ 函数返回 ▼收到 called 信号后，还要 <-sent 等响应发完。因为 sendResponse 需要抢 sending 锁，如果此时有其他响应正在写，需要排队。必须等写完才能安全返回（否则 wg.Done() 之后 WaitGroup 可能提前归零导致连接关闭）。
为什么不用 context.WithTimeout？
Go 中做超时控制最常见的方式是 context.WithTimeout，但 miniRPC 服务端没有用它，而是用了 select + time.After。原因是：
| 维度 | context 方案 | select + time.After 方案 |
|---|---|---|
| 需要方法签名支持 | 是，服务方法必须接收 ctx context.Context 参数 | 否，服务方法签名不变 |
| 取消正在执行的方法 | 可以（方法内部检查 ctx.Done()） | 不能（方法仍在后台跑完） |
| 实现复杂度 | 需要改方法签名 + 方法内部配合 | 无侵入，纯框架层实现 |
miniRPC 的服务方法签名是 func (t *T) Method(args Args, reply *Reply) error，没有 context 参数。框架无法把取消信号传递到方法内部，所以只能在外部用 select 做"超时后不等结果直接返回错误"。
这意味着超时后方法本身并不会被中断——子 goroutine 中的 svc.call() 仍然会跑完，只是它的结果不会再被使用（handleRequest 已经返回了超时响应）。
超时场景的完整时间线：
0s 3s(timeout) 10s(方法终于完成)
handleRequest： |---select---|---返回超时响应---▼(函数结束)
等待中... ↑ time.After触发
子 goroutine： |---svc.call()-----------------------------▼(执行完毕)
close(called)
sendResponse ← 这次写入仍会执行
但客户端已经收到了超时错误
close(sent)这是一个已知的局限：超时只是让客户端更快拿到错误响应，不能真正取消正在执行的服务方法。要实现真正的取消，需要：
- 修改方法签名，加入
context.Context参数 - 方法内部配合检查
ctx.Done()
与客户端超时的对比
miniRPC 中有三层超时，各自控制不同的环节：
客户端 服务端
│ │
│ ① ConnectTimeout │
│ net.DialTimeout(network, addr, t) │
│ ← 控制 TCP 握手 + Option 发送 │
│ │
│ ② context.Context 超时 │
│ client.Call(ctx, method, args, &r) │
│ ← 控制整个调用的端到端超时 │
│ │
│ │ ③ HandleTimeout
│ │ select + time.After(t)
│ │ ← 控制服务端单个请求处理的超时
│ │| 超时类型 | 控制方 | 作用范围 | 实现方式 |
|---|---|---|---|
| ConnectTimeout | 客户端 | TCP 连接建立 + 握手 | net.DialTimeout + select |
| context 超时 | 客户端 | 整个调用（从发送到收到响应） | select { case <-ctx.Done() } |
| HandleTimeout | 服务端 | 服务方法执行 + 响应发送 | select { case <-time.After(t) } |
三层超时各司其职，客户端的 context 超时是最外层的兜底——即使服务端没有设置 HandleTimeout，客户端也可以通过 context.WithTimeout 控制自己愿意等多久：
go
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err := client.Call(ctx, "Greeter.SayHello", args, &reply)
// 5 秒内没收到响应 → err = "rpc client: call failed: context deadline exceeded"总结
服务端超时控制的核心设计：
- 超时值由客户端指定：通过
Option.HandleTimeout在握手时传递给服务端 - select + time.After 模式：不侵入服务方法签名，在框架层实现超时
- 两个 channel 分工：
called标记方法执行完毕，sent标记响应发送完毕，确保资源安全释放 - 局限性：只能做到"超时后不等结果"，不能真正取消正在执行的方法