深入解析 grpc-go v1.84:一个工业级 RPC 框架的 HTTP/2、流控与 goroutine 复用之术

// Server
type server struct { pb.UnimplementedGreeterServer }

func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
    return &pb.HelloReply{Message: "Hello " + in.Name}, nil
}

func main() {
    lis, _ := net.Listen("tcp", ":50051")
    s := grpc.NewServer()
    pb.RegisterGreeterServer(s, &server{})
    s.Serve(lis)
}

// Client
conn, _ := grpc.NewClient("localhost:50051", grpc.WithTransportCredentials(insecure.NewCredentials()))
defer conn.Close()
c := pb.NewGreeterClient(conn)
r, _ := c.SayHello(ctx, &pb.HelloRequest{Name: "world"})

┌────────────────────────────────────────────────────────────────────────┐
│  应用代码 (Service interface + generated stub)                          │
│                                                                          │
│  SayHello(ctx, *HelloRequest) (*HelloReply, error)                    │
└────────────────────┬─────────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────────┐
│  Interceptor 链 (Unary/Stream, Server/Client)                          │
│  logging / auth / retry / tracing / metrics / ...                       │
└────────────────────┬─────────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────────┐
│  Codec 抽象 + Service Description                                       │
│  encoding/proto / encoding/protobuf / encoding/gzip                     │
└────────────────────┬─────────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────────┐
│  Stream 抽象 (ClientStream / ServerStream / 泛型 4 种)                  │
│  SendMsg / RecvMsg / Header / Trailer / SetDeadline                     │
└────────────────────┬─────────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────────┐
│  internal/transport (HTTP/2 client + server)                            │
│  http2_client.go 1920 行 + http2_server.go 1513 行                       │
│  ── 流控、HPACK 头压缩、SETTINGS 帧、keepalive、graceful shutdown ──    │
└────────────────────┬─────────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────────┐
│  net.Conn (TCP / Unix / HTTP/2 h2c)                                     │
└────────────────────────────────────────────────────────────────────────┘

                       并列: 客户端还有 ──
┌────────────────────────────────────────────────────────────────────────┐
│  ClientConn → Resolver → Balancer → pickerWrapper → addrConn           │
│  (resolver/dns + balancer/roundrobin + subconn + ...)                  │
└────────────────────────────────────────────────────────────────────────┘

type Server struct {
    opts         serverOptions
    statsHandler stats.Handler

    mu  sync.Mutex
    lis map[net.Listener]bool
    conns map[string]map[transport.ServerTransport]bool  // listener addr → active transports
    serve bool
    drain bool
    cv    *sync.Cond                                       // signaled when conns close
    services map[string]*serviceInfo                       // service name → info
    events traceEventLog

    quit *grpcsync.Event
    done *grpcsync.Event

    serveWG    sync.WaitGroup  // counts active Serve goroutines
    handlersWG sync.WaitGroup  // counts active method handler goroutines

    channelz *channelz.Server

    serverWorkerChannel      chan func()
    serverWorkerChannelClose func()
}

func NewServer(opt ...ServerOption) *Server {
    opts := defaultServerOptions
    for _, o := range globalServerOptions { o.apply(&opts) }
    for _, o := range opt { o.apply(&opts) }
    s := &Server{
        lis:          make(map[net.Listener]bool),
        opts:         opts,
        statsHandler: istats.NewCombinedHandler(opts.statsHandlers...),
        conns:        make(map[string]map[transport.ServerTransport]bool),
        services:     make(map[string]*serviceInfo),
        quit:         grpcsync.NewEvent(),
        done:         grpcsync.NewEvent(),
        channelz:     channelz.RegisterServer(""),
    }
    chainUnaryServerInterceptors(s)   // ← 把 Unary Interceptor 串成链
    chainStreamServerInterceptors(s)  // ← Stream Interceptor 串成链
    s.cv = sync.NewCond(&s.mu)
    if EnableTracing {
        _, file, line, _ := runtime.Caller(1)
        s.events = newTraceEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
    }
    if s.opts.numServerWorkers > 0 {
        s.initServerWorkers()  // ← 启动 serverWorker 池
    }
    channelz.Info(logger, s.channelz, "Server created")
    return s
}

func (s *Server) Serve(lis net.Listener) error {
    s.mu.Lock()
    s.printf("serving")
    s.serve = true
    if s.lis == nil { /* Stop/GracefulStop 已调 */ return ErrServerStopped }

    s.serveWG.Add(1)
    defer func() {
        s.serveWG.Done()
        if s.quit.HasFired() { <-s.done.Done() }   // ← 等待所有 conn 退出
    }()

    ls := &listenSocket{ Listener: lis, channelz: channelz.RegisterSocket(...) }
    s.lis[ls] = true

    defer func() {
        s.mu.Lock()
        if s.lis != nil && s.lis[ls] { ls.Close(); delete(s.lis, ls) }
        s.mu.Unlock()
    }()

    s.mu.Unlock()

    var tempDelay time.Duration
    for {
        rawConn, err := lis.Accept()
        if err != nil {
            if ne, ok := err.(interface{ Temporary() bool }); ok && ne.Temporary() {
                if tempDelay == 0 { tempDelay = 5 * time.Millisecond } else { tempDelay *= 2 }
                tempDelay = min(tempDelay, 1*time.Second)
                timer := time.NewTimer(tempDelay)
                select {
                case <-timer.C:
                case <-s.quit.Done():
                    timer.Stop()
                    return nil   // ← Stop 触发,优雅退出
                }
                continue
            }
            // ...
            return err
        }
        tempDelay = 0
        s.serveWG.Add(1)
        go func() {
            s.handleRawConn(lis.Addr().String(), rawConn)
            s.serveWG.Done()
        }()
    }
}

func (s *Server) processUnaryRPC(ctx context.Context, stream *transport.ServerStream, info *serviceInfo, md *MethodDesc, trInfo *traceInfo) (err error) {
    // 1. trace event: HeaderReceived
    if trInfo != nil { trInfo.tr.LazyLog(&trInfo.firstLine, true) }

    // 2. 选择 codec(根据 content-subtype)
    codec := s.getCodec(stream.Method())

    // 3. 解析 compression flag
    var comp encoding.Compressor
    if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
        comp = s.dc.Get(rc)
    }
    if comp == nil { return status.Errorf(codes.Unimplemented, ...) }

    // 4. 反序列化 request
    p := newCodecReader(stream, codec, comp)
    var req any
    if err := p.ReadHeader(&req); err != nil { ... }
    if err := p.Unmarshal(req); err != nil { ... }

    // 5. 应用拦截器链
    reply, appErr := md.Handler(s.serviceImpl, ctx, p.Decode, s.opts.unaryInt)

    // 6. 序列化 + 发送 response
    ...
}

type ClientConn struct {
    target          string
    resolvedViaResolver bool
    authority       string

    dopts           dialOptions
    // ...

    mu              sync.Mutex
    resolverWrapper *ccResolverWrapper
    balancerWrapper *ccBalancerWrapper
    pickerWrapper   *pickerWrapper
    conns           map[*addrConn]struct{}
    // ...

    sc              *ServiceConfig
    // ...
}

type ServerTransport interface {
    HandleStreams(func(*ServerStream))
    WriteHeader(s *ServerStream, md []byte, appData []byte, opts *WriteOptions) error
    Write(s *ServerStream, hdr []byte, data mem.BufferSlice, opts *WriteOptions) error
    WriteStatus(s *ServerStream, st *status.Status) error
    Close()
    RemoteAddr() net.Addr
    Drain(...)  // graceful shutdown
}

type ClientTransport interface {
    NewStream(ctx context.Context, addr string, method string, opts ...CallOption) (*ClientStream, error)
    Close(...)
    CloseStream(...)
    SetConnectivityState(...)
    WaitForStateChange(...)
}

type controlbuf struct {
    cb        []http2.Frame         // 待发送的控制帧队列
    sendQuota uint32                // 当前可发送的字节数
    // ...
    sendQuotaCh chan struct{}       // 收到 WINDOW_UPDATE 时唤醒
    // ...
}

func (c *controlbuf) wait() {
    if c.sendQuota > 0 { return }
    <-c.sendQuotaCh
}

func (c *controlbuf) try(f http2.Frame) bool {
    // 检查流控配额
    // 配额够 → 直接放入队列
    // 配额不够 → 返回 false(等 quota)
}

type http2Server struct {
    ctx                context.Context
    conn               net.Conn
    // ...
    tconn              *http2.Transport
    // ...
    streams            map[uint32]*serverStream  // stream ID → stream
    sendQuotaPool      map[uint32]int            // 每个 stream 的发送配额
    // ...
}

func (t *http2Server) handleHeaders(f *http2.MetaHeadersFrame) {
    // 1. 解析 :method=POST, :path=/helloworld.Greeter/SayHello
    // 2. 检查是否有 service / method
    // 3. 查 services map,找到 handler
    // 4. 创建 serverStream,加入 t.streams
    // 5. handleStream(stream)
}

type Code uint32

const (
    OK Code = 0
    Canceled Code = 1
    Unknown Code = 2
    InvalidArgument Code = 3
    DeadlineExceeded Code = 4
    NotFound Code = 5
    AlreadyExists Code = 6
    PermissionDenied Code = 7
    ResourceExhausted Code = 8
    FailedPrecondition Code = 9
    Aborted Code = 10
    OutOfRange Code = 11
    Unimplemented Code = 12
    Internal Code = 13
    Unavailable Code = 14
    DataLoss Code = 15
    Unauthenticated Code = 16
)

// internal/grpcsync/event.go
type Event struct {
    fired atomic.Bool
    c     chan struct{}
}

func (e *Event) Fire() bool {
    if e.fired.CompareAndSwap(false, true) {
        close(e.c)
        return true
    }
    return false
}

func (e *Event) Done() <-chan struct{} { return e.c }
func (e *Event) HasFired() bool        { return e.fired.Load() }

$ find . -name "*.go" -not -path "./.git*" -not -path "./testdata/*" | xargs wc -l | tail -1
  334,947 total

核心 5 文件 = 8230 行:
server.go                2,261 行
clientconn.go            1,975 行
stream.go                1,975 行
rpc_util.go              1,207 行
dialoptions.go             812 行

internal/transport/    14,612 行(http2 4207 + controlbuf + flowcontrol + ...)

模块规模(top 12):
internal/transport/      14,612 行
server.go                 2,261 行
clientconn.go             1,975 行
stream.go                 1,975 行
internal/transport/networktype/    ~ 1,800 行
internal/transport/transport_test  4,240 行(测试)
rpc_util.go               1,207 行
internal/transport/http2_client.go  1,920 行
internal/transport/http2_server.go  1,513 行
xds/                      ~ 3,500 行
credentials/              ~ 2,800 行
balancer/                 ~ 2,200 行

子包数量:
42 个 internal 子包
12 个内置 balancer
9 种 credentials
3 种 encoding(proto / protobuf / gzip)
2 种 status(公开 + internal)