gRPC 服务端深度实现

本篇深入 gRPC 服务端的内部机制。我们不重复「如何写一个 Hello World gRPC 服务」，而是聚焦于生产级服务端必须掌握的内容：Server 选项的底层含义、四种服务方法的实现细节与陷阱、流式 RPC 的错误传播、拦截器链的执行顺序，以及优雅关停的正确姿势。

一、gRPC 服务端架构

gRPC 服务端的核心组件如下：

┌─────────────────────────────────────────────┐
│              grpc.Server                     │
│  ┌─────────────┐  ┌──────────────────────┐  │
│  │ Transport    │  │ Interceptor Chain    │  │
│  │ (HTTP/2)     │  │ (Unary / Stream)     │  │
│  └──────┬──────┘  └──────────┬───────────┘  │
│         │                     │              │
│  ┌──────▼─────────────────────▼───────────┐  │
│  │         Service Handler                │  │
│  │  (Unary / ServerStream / ClientStream  │  │
│  │   / BidiStream)                        │  │
│  └────────────────────────────────────────┘  │
└─────────────────────────────────────────────┘

Transport 层：基于 HTTP/2 的监听器，负责连接管理、帧解析、流复用。
Interceptor 层：在请求到达业务 handler 之前/之后执行横切逻辑。
Service Handler：业务代码实现，对应 proto 定义的 rpc 方法。

服务端的入口是 grpc.NewServer(opts...)，然后通过 RegisterXxxServer 注册服务，最后 server.Serve(lis) 开始接受连接。

二、Server 结构体配置

grpc.NewServer 接受可变数量的 ServerOption，这些选项决定了服务端的行为。下面逐一讲解生产环境最关键的配置项。

1. 消息大小限制

默认情况下 gRPC 单条消息上限是 4MB（math.MaxInt32 实际为 2GB，但默认 defaultServerMaxRecvMsgSize 为 4MB）。如果你的业务消息（如文件分块、批量列表）超过此限制，需要显式调整。

package main

import (
	"fmt"
	"google.golang.org/grpc"
)

func main() {
	// 消息大小相关选项
	server := grpc.NewServer(
		// 单条接收消息上限：16MB
		grpc.MaxRecvMsgSize(16*1024*1024),
		// 单条发送消息上限：16MB
		grpc.MaxSendMsgSize(16*1024*1024),
		// 单条 RPC 接收的并发流数量上限
		grpc.MaxConcurrentStreams(256),
	)

	fmt.Printf("server created: %T\n", server)
}

注意：调大消息上限会增加内存压力。如果单条消息确实很大，更优雅的方案是用流式 RPC 分块传输，而不是调大 MaxRecvMsgSize。

2. Keepalive 参数

Keepalive 机制用于检测连接是否存活，防止僵死连接占用资源。

package main

import (
	"fmt"
	"time"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)

func main() {
	kaPolicy := keepalive.ServerParameters{
		// 连续 30 秒没有活动（没有 ping 响应）则认为连接死掉
		Time: 30 * time.Second,
		// 发出 ping 后 5 秒没收到 ack 则断开
		Timeout: 5 * time.Second,
	}

	kaEnforcement := keepalive.EnforcementPolicy{
		// 客户端至少间隔 10 秒才能发 ping，更频繁则服务端拒绝
		MinTime: 10 * time.Second,
		// 客户端不允许在无活动流时 ping，否则服务端发 GOAWAY
		PermitWithoutStream: false,
	}

	server := grpc.NewServer(
		grpc.KeepaliveParams(kaPolicy),
		grpc.KeepaliveEnforcementPolicy(kaEnforcement),
	)

	fmt.Printf("server with keepalive: %T\n", server)
}

各参数含义：

Time：服务端发送 ping 的间隔（仅当连接上有流时才有意义）。
Timeout：ping 超时后断开连接。
MinTime：客户端 ping 的最小间隔，防止恶意 ping 洪泛。
PermitWithoutStream：是否允许客户端在没有活动流时 ping。

3. TLS 配置

生产环境必须启用 TLS。通过 grpc.Creds 传入传输凭证。

package main

import (
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func main() {
	// 从文件加载服务端证书和私钥
	// 实际使用时需要准备 server.crt 和 server.key 文件
	// creds, err := credentials.NewServerTLSFromFile("server.crt", "server.key")
	// if err != nil {
	//     log.Fatalf("failed to load TLS: %v", err)
	// }
	// server := grpc.NewServer(grpc.Creds(creds))

	// 这里用无凭证演示选项组装
	server := grpc.NewServer(
		grpc.Creds(credentials.NewTLS(nil)), // 传入真实 tls.Config
	)
	fmt.Printf("server with TLS: %T\n", server)
}

TLS 的详细配置（包括 mTLS 双向认证）将在第七篇「gRPC 安全与 TLS」中深入讲解。

4. 完整的生产级 Server 配置

package main

import (
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"time"
)

func newProductionServer() *grpc.Server {
	return grpc.NewServer(
		// 消息限制
		grpc.MaxRecvMsgSize(16*1024*1024),
		grpc.MaxSendMsgSize(16*1024*1024),
		grpc.MaxConcurrentStreams(256),

		// Keepalive
		grpc.KeepaliveParams(keepalive.ServerParameters{
			Time:    30 * time.Second,
			Timeout: 5 * time.Second,
		}),
		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
			MinTime:             10 * time.Second,
			PermitWithoutStream: false,
		}),

		// TLS（传入真实凭证）
		grpc.Creds(credentials.NewTLS(nil)),

		// 拦截器链（下一篇详解）
		grpc.ChainUnaryInterceptor(
			recoveryUnaryInterceptor,
			loggingUnaryInterceptor,
		),
		grpc.ChainStreamInterceptor(
			recoveryStreamInterceptor,
			loggingStreamInterceptor,
		),

		// 自定义 HeaderTableSize（HTTP/2 头部压缩表大小）
		grpc.HeaderTableSize(4096),
	)
}

// 以下是占位拦截器，实际实现见第四篇
func recoveryUnaryInterceptor(
	ctx interface{}, req interface{}, info *grpc.UnaryServerInfo,
	handler grpc.UnaryHandler,
) (interface{}, error) {
	return handler(ctx.(interface{}).(interface{}).(interface{}), req) // 简化占位
}

func loggingUnaryInterceptor(
	ctx interface{}, req interface{}, info *grpc.UnaryServerInfo,
	handler grpc.UnaryHandler,
) (interface{}, error) {
	return handler(ctx.(interface{}).(interface{}).(interface{}), req) // 简化占位
}

func recoveryStreamInterceptor(
	srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo,
	handler grpc.StreamHandler,
) error {
	return handler(srv, ss)
}

func loggingStreamInterceptor(
	srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo,
	handler grpc.StreamHandler,
) error {
	return handler(srv, ss)
}

func main() {
	server := newProductionServer()
	fmt.Printf("production server ready: %T\n", server)
}

三、注册服务

protoc 生成的代码中会有一个 RegisterXxxServer 函数，它内部调用 grpc.RegisterService 将服务描述符注册到 Server。

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
)

// 手写一个最小服务定义，模拟 protoc 生成代码的结构
// 这样无需 protoc 也能理解注册过程

// 1. 定义服务端接口
type GreeterServer interface {
	SayHello(ctx context.Context, req *HelloRequest) (*HelloReply, error)
}

// 2. 定义请求/响应结构
type HelloRequest struct {
	Name string
}
type HelloReply struct {
	Message string
}

// 3. 定义 handler 适配函数（将接口方法转为统一的 handler 签名）
func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context,
	dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor,
) (interface{}, error) {
	req := &HelloRequest{}
	if err := dec(req); err != nil {
		return nil, err
	}
	if interceptor == nil {
		return srv.(GreeterServer).SayHello(ctx, req)
	}
	info := &grpc.UnaryServerInfo{
		Server:     srv,
		FullMethod: "/greeter.Greeter/SayHello",
	}
	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
		return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))
	}
	return interceptor(ctx, req, info, handler)
}

// 4. 注册函数（模拟 protoc 生成的 RegisterGreeterServer）
func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
	s.RegisterService(&grpc.ServiceDesc{
		ServiceName: "greeter.Greeter",
		HandlerType: (*GreeterServer)(nil),
		Methods: []grpc.MethodDesc{
			{MethodName: "SayHello", Handler: _Greeter_SayHello_Handler},
		},
		Streams:  []grpc.StreamDesc{},
		Metadata: "greeter.proto",
	}, srv)
}

// 5. 实现服务
type myGreeter struct{}

func (g *myGreeter) SayHello(ctx context.Context, req *HelloRequest) (*HelloReply, error) {
	return &HelloReply{Message: "Hello " + req.Name}, nil
}

func main() {
	server := grpc.NewServer()
	RegisterGreeterServer(server, &myGreeter{})
	fmt.Println("GreeterService registered successfully")
	fmt.Printf("server: %T\n", server)
}

理解了这个手写过程，就能明白 protoc 生成的那一大段代码到底做了什么。

四、四种服务方法实现详解

1. Unary RPC：一元调用

最简单也最常用：客户端发一个请求，服务端回一个响应。

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// 模拟消息类型
type CalcRequest struct {
	Op    string
	A, B  float64
}
type CalcResponse struct {
	Result float64
}

// 服务端接口
type CalculatorServer interface {
	Calculate(ctx context.Context, req *CalcRequest) (*CalcResponse, error)
}

type calcServer struct{}

func (s *calcServer) Calculate(ctx context.Context, req *CalcRequest) (*CalcResponse, error) {
	// 检查 context 是否已取消
	select {
	case <-ctx.Done():
		return nil, status.FromContextError(ctx.Err()).Err()
	default:
	}

	var result float64
	switch req.Op {
	case "+":
		result = req.A + req.B
	case "-":
		result = req.A - req.B
	case "*":
		result = req.A * req.B
	case "/":
		if req.B == 0 {
			return nil, status.Error(codes.InvalidArgument, "division by zero")
		}
		result = req.A / req.B
	default:
		return nil, status.Errorf(codes.InvalidArgument, "unknown op: %s", req.Op)
	}
	return &CalcResponse{Result: result}, nil
}

// 手写 ServiceDesc 注册
func _Calc_Calculate_Handler(srv interface{}, ctx context.Context,
	dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor,
) (interface{}, error) {
	req := &CalcRequest{}
	if err := dec(req); err != nil {
		return nil, err
	}
	if interceptor == nil {
		return srv.(CalculatorServer).Calculate(ctx, req)
	}
	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/calc.Calculator/Calculate"}
	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
		return srv.(CalculatorServer).Calculate(ctx, req.(*CalcRequest))
	}
	return interceptor(ctx, req, info, handler)
}

func RegisterCalculatorServer(s *grpc.Server, srv CalculatorServer) {
	s.RegisterService(&grpc.ServiceDesc{
		ServiceName: "calc.Calculator",
		HandlerType: (*CalculatorServer)(nil),
		Methods:     []grpc.MethodDesc{{MethodName: "Calculate", Handler: _Calc_Calculate_Handler}},
		Streams:     []grpc.StreamDesc{},
		Metadata:    "calc.proto",
	}, srv)
}

func main() {
	lis, err := net.Listen("tcp", ":50052")
	if err != nil {
		log.Fatalf("listen failed: %v", err)
	}

	server := grpc.NewServer()
	RegisterCalculatorServer(server, &calcServer{})

	go func() {
		log.Printf("calculator server listening on %s", lis.Addr())
		if err := server.Serve(lis); err != nil {
			log.Printf("serve error: %v", err)
		}
	}()

	// 优雅退出
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	server.GracefulStop()
	log.Println("server stopped gracefully")
}

2. Server Streaming：服务端流

一次请求，多次响应。服务端通过 Stream.Send 逐条推送消息，结束时返回 nil；客户端通过 Recv 循环读取直到 io.EOF。

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"time"

	"google.golang.org/grpc"
)

type RangeRequest struct {
	From, To int32
}
type Number struct {
	Value int32
}

// 服务端流接口：第二个参数是 stream 对象
type NumberServer interface {
	Range(req *RangeRequest, stream grpc.ServerStreamingServer[Number]) error
}

// 为了让代码通用可编译，这里用 grpc.ServerStream 演示
type numberServer struct{}

func (s *numberServer) Range(req *RangeRequest, stream grpc.ServerStream) error {
	for i := req.From; i <= req.To; i++ {
		// 模拟逐条推送
		if err := stream.SendMsg(&Number{Value: i}); err != nil {
			return err
		}
		time.Sleep(100 * time.Millisecond) // 模拟延迟
	}
	return nil
}

// 注册
func _Number_Range_Handler(srv interface{}, stream grpc.ServerStream) error {
	req := &RangeRequest{}
	if err := stream.RecvMsg(req); err != nil {
		return err
	}
	return srv.(interface {
		Range(req *RangeRequest, stream grpc.ServerStream) error
	}).Range(req, stream)
}

func RegisterNumberServer(s *grpc.Server, srv interface {
	Range(req *RangeRequest, stream grpc.ServerStream) error
}) {
	s.RegisterService(&grpc.ServiceDesc{
		ServiceName: "num.NumberService",
		HandlerType: (*interface {
			Range(req *RangeRequest, stream grpc.ServerStream) error
		})(nil),
		Methods:  []grpc.MethodDesc{},
		Streams:  []grpc.StreamDesc{{StreamName: "Range", Handler: _Number_Range_Handler, ServerStreams: true}},
		Metadata: "num.proto",
	}, srv)
}

func main() {
	// 演示服务端流的核心逻辑（不真正监听，仅展示 Send/Recv 语义）
	// 真正运行时需要 grpc.Server.Serve

	// 模拟一个 Range 调用的内部循环
	ch := make(chan int32, 10)
	go func() {
		// 模拟服务端推送 1~5
		for i := int32(1); i <= 5; i++ {
			ch <- i
			time.Sleep(50 * time.Millisecond)
		}
		close(ch)
	}()

	// 模拟客户端接收
	for n := range ch {
		fmt.Printf("recv: %d\n", n)
	}
	fmt.Println("stream ended (EOF)")

	_ = context.Background
	_ = io.EOF
	_ = net.Listen
	_ = grpc.NewServer
	_ = log.Printf
	_ = numberServer{}
}

服务端流的要点：

Send 之间可以穿插任意业务逻辑（查数据库、调下游服务）。
如果客户端断开连接，Send 会返回错误，应立即 return。
流的结束由 return nil 触发，gRPC 框架会向客户端发送 io.EOF。

3. Client Streaming：客户端流

客户端多次发送，服务端一次性响应。适合批量上传、聚合统计。

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"os/signal"
	"syscall"

	"google.golang.org/grpc"
)

type UploadChunk struct {
	Sequence int32
	Data     []byte
}
type UploadResult struct {
	TotalChunks int32
	TotalBytes  int32
}

type FileServer interface {
	Upload(stream grpc.ClientStreamingServer[UploadChunk, UploadResult]) error
}

// 用 grpc.ServerStream 演示（兼容编译）
type fileServer struct{}

func (s *fileServer) Upload(stream grpc.ServerStream) error {
	var totalChunks, totalBytes int32
	for {
		chunk := &UploadChunk{}
		if err := stream.RecvMsg(chunk); err != nil {
			if err == io.EOF {
				// 客户端发送完毕，返回汇总结果
				return stream.SendMsg(&UploadResult{
					TotalChunks: totalChunks,
					TotalBytes:  totalBytes,
				})
			}
			return err
		}
		totalChunks++
		totalBytes += int32(len(chunk.Data))
		fmt.Printf("recv chunk #%d: %d bytes\n", chunk.Sequence, len(chunk.Data))
	}
}

func _File_Upload_Handler(srv interface{}, stream grpc.ServerStream) error {
	return srv.(interface {
		Upload(stream grpc.ServerStream) error
	}).Upload(stream)
}

func RegisterFileServer(s *grpc.Server, srv interface {
	Upload(stream grpc.ServerStream) error
}) {
	s.RegisterService(&grpc.ServiceDesc{
		ServiceName: "file.FileService",
		HandlerType: (*interface {
			Upload(stream grpc.ServerStream) error
		})(nil),
		Methods:  []grpc.MethodDesc{},
		Streams:  []grpc.StreamDesc{{StreamName: "Upload", Handler: _File_Upload_Handler, ClientStreams: true}},
		Metadata: "file.proto",
	}, srv)
}

func main() {
	// 演示客户端流的核心逻辑：服务端循环 Recv 直到 EOF
	ch := make(chan *UploadChunk, 3)
	go func() {
		chunks := []*UploadChunk{
			{Sequence: 1, Data: []byte("hello")},
			{Sequence: 2, Data: []byte(" world")},
			{Sequence: 3, Data: []byte("!")},
		}
		for _, c := range chunks {
			ch <- c
		}
		close(ch)
	}()

	var totalChunks, totalBytes int32
	for chunk := range ch {
		totalChunks++
		totalBytes += int32(len(chunk.Data))
	}
	fmt.Printf("upload complete: chunks=%d bytes=%d\n", totalChunks, totalBytes)

	_ = context.Background
	_ = fileServer{}
	_ = grpc.NewServer
	_ = net.Listen
	_ = log.Printf
	_ = os.Signal(nil)
	_ = syscall.SIGINT
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
}

客户端流的要点：

服务端在 RecvMsg 返回 io.EOF 时，用 SendMsg 发送最终响应。
客户端用 CloseAndRecv 发送 EOF 并等待响应。

4. Bidirectional Streaming：双向流

双方同时收发，最灵活也最复杂。服务端和客户端可以按任意顺序读写。

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"
)

type ChatMessage struct {
	From string
	Text string
}

// 双向流的典型模式：每收到一条消息就处理并回复
func handleChatStream(stream interface {
	RecvMsg(interface{}) error
	SendMsg(interface{}) error
}) error {
	for {
		msg := &ChatMessage{}
		if err := stream.RecvMsg(msg); err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}
		fmt.Printf("[server] recv from %s: %s\n", msg.From, msg.Text)

		// echo 回去
		reply := &ChatMessage{From: "server", Text: "echo: " + msg.Text}
		if err := stream.SendMsg(reply); err != nil {
			return err
		}
	}
}

// 模拟双向流的 mock 对象
type mockBidiStream struct {
	mu       sync.Mutex
	recvIdx  int
	recvBuf  []*ChatMessage
	sendBuf  []*ChatMessage
}

func (m *mockBidiStream) RecvMsg(v interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.recvIdx >= len(m.recvBuf) {
		return io.EOF
	}
	msg := m.recvBuf[m.recvIdx]
	m.recvIdx++
	*(v.(*ChatMessage)) = *msg
	return nil
}

func (m *mockBidiStream) SendMsg(v interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.sendBuf = append(m.sendBuf, v.(*ChatMessage))
	return nil
}

func main() {
	mock := &mockBidiStream{
		recvBuf: []*ChatMessage{
			{From: "alice", Text: "hello"},
			{From: "alice", Text: "how are you?"},
			{From: "alice", Text: "bye"},
		},
	}

	go func() {
		if err := handleChatStream(mock); err != nil {
			log.Printf("chat error: %v", err)
		}
	}()

	// 等待处理完成
	time.Sleep(100 * time.Millisecond)

	fmt.Println("\n--- server sent ---")
	for _, s := range mock.sendBuf {
		fmt.Printf("[sent] from %s: %s\n", s.From, s.Text)
	}

	_ = context.Background
}

双向流的进阶模式：

上面是「读一条写一条」的同步 echo 模式。实际聊天场景中，读写通常是异步并行的：

package main

import (
	"fmt"
	"io"
	"sync"
	"time"
)

type ChatMessage struct {
	From string
	Text string
}

type BidiStream interface {
	RecvMsg(interface{}) error
	SendMsg(interface{}) error
}

// 异步双向流处理：读写在不同 goroutine 中并行
func handleChatAsync(stream BidiStream) error {
	var wg sync.WaitGroup
	errCh := make(chan error, 2)

	// 读 goroutine
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			msg := &ChatMessage{}
			if err := stream.RecvMsg(msg); err != nil {
				if err == io.EOF {
					return
				}
				errCh <- err
				return
			}
			fmt.Printf("[recv] %s: %s\n", msg.From, msg.Text)
		}
	}()

	// 写 goroutine：定时广播系统消息
	wg.Add(1)
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(50 * time.Millisecond)
		defer ticker.Stop()
		count := 0
		for {
			select {
			case <-ticker.C:
				count++
				if count > 3 {
					return // 演示用，发 3 条后停止
				}
				msg := &ChatMessage{From: "system", Text: fmt.Sprintf("tick #%d", count)}
				if err := stream.SendMsg(msg); err != nil {
					errCh <- err
					return
				}
			}
		}
	}()

	wg.Wait()
	close(errCh)
	for err := range errCh {
		if err != nil {
			return err
		}
	}
	return nil
}

type mockStream struct {
	mu      sync.Mutex
	recvIdx int
	recvBuf []*ChatMessage
	sendBuf []*ChatMessage
}

func (m *mockStream) RecvMsg(v interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	time.Sleep(10 * time.Millisecond) // 模拟网络延迟
	if m.recvIdx >= len(m.recvBuf) {
		return io.EOF
	}
	*(v.(*ChatMessage)) = *m.recvBuf[m.recvIdx]
	m.recvIdx++
	return nil
}

func (m *mockStream) SendMsg(v interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.sendBuf = append(m.sendBuf, v.(*ChatMessage))
	return nil
}

func main() {
	stream := &mockStream{
		recvBuf: []*ChatMessage{
			{From: "alice", Text: "hi"},
			{From: "alice", Text: "anyone there?"},
		},
	}

	if err := handleChatAsync(stream); err != nil {
		fmt.Printf("error: %v\n", err)
	}

	fmt.Println("--- sent ---")
	for _, s := range stream.sendBuf {
		fmt.Printf("  %s: %s\n", s.From, s.Text)
	}
}

五、流式 RPC 的错误处理

流式 RPC 的错误处理与 Unary 有重要区别：

Unary：handler 返回 error，框架自动转为 gRPC status。
Server Streaming：在 Send 之后返回 error，客户端 Recv 会收到该 error 而非 EOF。
Client Streaming：在 Recv 过程中返回 error，客户端的 Send 或 CloseAndRecv 会收到 error。
Bidi Streaming：任一方在 Send/Recv 时返回 error 都会终止流。

package main

import (
	"context"
	"fmt"
	"io"
	"time"

	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type LogEntry struct {
	Level   string
	Message string
}

type mockLogStream struct {
	sendCount int
	maxSend   int
}

func (m *mockLogStream) SendMsg(v interface{}) error {
	m.sendCount++
	if m.sendCount > m.maxSend {
		// 模拟中途出错
		return status.Error(codes.ResourceExhausted, "log buffer full")
	}
	return nil
}

func (m *mockLogStream) RecvMsg(v interface{}) error {
	return io.EOF
}

// 服务端流中途出错：Send 返回错误
func streamLogs(stream interface {
	SendMsg(interface{}) error
}) error {
	logs := []*LogEntry{
		{Level: "INFO", Message: "starting"},
		{Level: "INFO", Message: "processing"},
		{Level: "ERROR", Message: "disk full"},
		{Level: "INFO", Message: "retrying"},
	}
	for _, log := range logs {
		if err := stream.SendMsg(log); err != nil {
			// 关键：返回 error，客户端 Recv 会收到此 error（而非 EOF）
			fmt.Printf("send failed at log %q: %v\n", log.Message, err)
			return err
		}
		fmt.Printf("sent: [%s] %s\n", log.Level, log.Message)
	}
	return nil
}

func main() {
	stream := &mockLogStream{maxSend: 2}
	err := streamLogs(stream)

	// 客户端侧错误处理
	if err != nil {
		st, ok := status.FromError(err)
		if ok {
			fmt.Printf("client received error: code=%v msg=%s\n", st.Code(), st.Message())
			switch st.Code() {
			case codes.ResourceExhausted:
				fmt.Println("-> 限流，应退避重试")
			case codes.Unavailable:
				fmt.Println("-> 服务不可用，应重试")
			default:
				fmt.Printf("-> 其他错误: %v\n", st.Code())
			}
		}
	}

	_ = context.Background
	_ = time.Second
}

核心原则：流式 RPC 中，一旦 handler 返回非 nil error，流就结束了。客户端通过 Recv 的返回值区分「正常结束」（io.EOF）和「异常结束」（其他 error）。

六、服务端拦截器

1. 一元拦截器

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc"
)

// 一元拦截器签名
// func(ctx, req, info, handler) (resp, err)
func loggingUnaryInterceptor(ctx context.Context, req interface{},
	info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (interface{}, error) {
	start := time.Now()
	// 调用真正的 handler
	resp, err := handler(ctx, req)
	log.Printf("[unary] %s cost=%v err=%v", info.FullMethod, time.Since(start), err)
	return resp, err
}

func metricsUnaryInterceptor(ctx context.Context, req interface{},
	info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
) (interface{}, error) {
	start := time.Now()
	resp, err := handler(ctx, req)
	duration := time.Since(start).Seconds()
	// 实际项目中这里上报到 Prometheus
	fmt.Printf("[metrics] %s duration=%.3fs success=%v\n", info.FullMethod, duration, err == nil)
	return resp, err
}

func main() {
	// 单个拦截器
	_ = grpc.NewServer(grpc.UnaryInterceptor(loggingUnaryInterceptor))

	// 多个拦截器链
	server := grpc.NewServer(grpc.ChainUnaryInterceptor(
		loggingUnaryInterceptor,
		metricsUnaryInterceptor,
	))
	fmt.Printf("server with interceptor chain: %T\n", server)
}

2. 流拦截器

流拦截器需要包装 grpc.ServerStream 来拦截每次 SendMsg/RecvMsg。

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc"
)

// 包装 ServerStream 以拦截每次消息收发
type wrappedStream struct {
	grpc.ServerStream
	method  string
	recvCnt int
	sendCnt int
}

func (w *wrappedStream) RecvMsg(m interface{}) error {
	err := w.ServerStream.RecvMsg(m)
	if err == nil {
		w.recvCnt++
		log.Printf("[stream %s] recv #%d", w.method, w.recvCnt)
	}
	return err
}

func (w *wrappedStream) SendMsg(m interface{}) error {
	err := w.ServerStream.SendMsg(m)
	if err == nil {
		w.sendCnt++
		log.Printf("[stream %s] send #%d", w.method, w.sendCnt)
	}
	return err
}

func loggingStreamInterceptor(srv interface{}, ss grpc.ServerStream,
	info *grpc.StreamServerInfo, handler grpc.StreamHandler,
) error {
	start := time.Now()
	wrapped := &wrappedStream{ServerStream: ss, method: info.FullMethod}
	err := handler(srv, wrapped)
	log.Printf("[stream] %s cost=%v recv=%d send=%v err=%v",
		info.FullMethod, time.Since(start), wrapped.recvCnt, wrapped.sendCnt, err)
	return err
}

func main() {
	server := grpc.NewServer(grpc.StreamInterceptor(loggingStreamInterceptor))
	fmt.Printf("server with stream interceptor: %T\n", server)
	_ = context.Background
}

3. 拦截器链的执行顺序

grpc.ChainUnaryInterceptor 的执行顺序是洋葱模型：第一个注册的拦截器在最外层，最后注册的最靠近 handler。

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
)

func makeInterceptor(name string) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req interface{},
		info *grpc.UnaryServerInfo, handler grpc.UnaryHandler,
	) (interface{}, error) {
		fmt.Printf("  [%s] before handler\n", name)
		resp, err := handler(ctx, req)
		fmt.Printf("  [%s] after handler\n", name)
		return resp, err
	}
}

func main() {
	// 模拟拦截器链的执行顺序
	// 注册顺序: A, B, C
	// 执行顺序: A before -> B before -> C before -> handler -> C after -> B after -> A after
	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
		fmt.Println("  [handler] executing")
		return "result", nil
	}

	// 手动模拟链式调用
	chain := []grpc.UnaryServerInterceptor{
		makeInterceptor("A-recovery"),
		makeInterceptor("B-logging"),
		makeInterceptor("C-auth"),
	}

	// 构建洋葱
	var finalHandler grpc.UnaryHandler = handler
	for i := len(chain) - 1; i >= 0; i-- {
		interceptor := chain[i]
		next := finalHandler
		finalHandler = func(ctx context.Context, req interface{}) (interface{}, error) {
			return interceptor(ctx, req, &grpc.UnaryServerInfo{FullMethod: "/test"}, next)
		}
	}

	fmt.Println("Calling with interceptor chain A -> B -> C:")
	finalHandler(context.Background(), "request")
}

输出为：

Calling with interceptor chain A -> B -> C:
  [A-recovery] before handler
  [B-logging] before handler
  [C-auth] before handler
  [handler] executing
  [C-auth] after handler
  [B-logging] after handler
  [A-recovery] after handler

最佳实践：recovery 放最外层（兜底），auth 放最内层（最靠近业务），logging/tracing 在中间。

七、优雅关停：GracefulStop

GracefulStop 会停止接受新连接和新请求，但会等待所有正在处理的请求完成。与之相对，Stop 会立即断开所有连接。

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"google.golang.org/grpc"
)

type EchoServer interface {
	Echo(ctx context.Context, msg string) (string, error)
}
type echoServer struct {
	mu      sync.Mutex
	active  int
	stopped bool
}

func (s *echoServer) Echo(ctx context.Context, msg string) (string, error) {
	s.mu.Lock()
	if s.stopped {
		s.mu.Unlock()
		return "", fmt.Errorf("server is stopping")
	}
	s.active++
	s.mu.Unlock()

	defer func() {
		s.mu.Lock()
		s.active--
		s.mu.Unlock()
	}()

	// 模拟慢请求
	time.Sleep(500 * time.Millisecond)
	return msg, nil
}

func main() {
	lis, err := net.Listen("tcp", ":50053")
	if err != nil {
		log.Fatal(err)
	}

	server := grpc.NewServer()
	log.Printf("server listening on %s", lis.Addr())

	go func() {
		if err := server.Serve(lis); err != nil {
			log.Printf("serve: %v", err)
		}
	}()

	// 监听信号
	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	sig := <-quit
	log.Printf("received signal: %v, shutting down gracefully...", sig)

	// 设置关停超时
	done := make(chan struct{})
	go func() {
		// GracefulStop 会等待所有正在处理的 RPC 完成
		server.GracefulStop()
		close(done)
	}()

	select {
	case <-done:
		log.Println("graceful shutdown complete")
	case <-time.After(10 * time.Second):
		log.Println("graceful shutdown timeout, forcing stop")
		server.Stop() // 强制断开
	}

	_ = context.Background
	_ = echoServer{}
}

GracefulStop vs Stop：

方法	行为
`GracefulStop`	停止接受新连接/请求，等待在途请求完成
`Stop`	立即关闭所有连接，在途请求被中断

生产环境应始终用 GracefulStop + 超时兜底，防止个别慢请求导致进程无法退出。

八、完整示例：聊天服务（双向流）

下面是一个完整可运行的聊天服务，综合运用了双向流、拦截器、优雅关停。

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// === 消息定义 ===
type ChatMessage struct {
	From string
	Text string
	Time int64
}

// === 服务接口 ===
type ChatServiceServer interface {
	JoinChat(stream grpc.ServerStream) error
}

// === 聊天室实现 ===
type chatRoom struct {
	mu      sync.RWMutex
	members map[string]chan *ChatMessage
}

func newChatRoom() *chatRoom {
	return &chatRoom{members: make(map[string]chan *ChatMessage)}
}

func (r *chatRoom) join(name string) (<-chan *ChatMessage, func()) {
	r.mu.Lock()
	defer r.mu.Unlock()
	ch := make(chan *ChatMessage, 16)
	r.members[name] = ch

	// 返回退订函数
	leave := func() {
		r.mu.Lock()
		defer r.mu.Unlock()
		delete(r.members, name)
		close(ch)
	}
	return ch, leave
}

func (r *chatRoom) broadcast(sender string, msg *ChatMessage) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	for name, ch := range r.members {
		if name == sender {
			continue // 不回发给自己
		}
		select {
		case ch <- msg:
		default:
			log.Printf("warn: %s's channel is full, dropping message", name)
		}
	}
}

// === 服务端 handler ===
type chatServer struct {
	room *chatRoom
}

func (s *chatServer) JoinChat(stream grpc.ServerStream) error {
	// 第一条消息必须是 join 请求，携带用户名
	first := &ChatMessage{}
	if err := stream.RecvMsg(first); err != nil {
		return status.Error(codes.InvalidArgument, "failed to receive join message")
	}

	userName := first.From
	log.Printf("[join] %s entered the room", userName)

	// 注册到聊天室
	msgCh, leave := s.room.join(userName)
	defer leave()

	// 广播上线通知
	s.room.broadcast(userName, &ChatMessage{
		From: "system", Text: userName + " joined", Time: time.Now().Unix(),
	})

	// 启动写 goroutine：把广播消息发给当前客户端
	errCh := make(chan error, 1)
	go func() {
		for msg := range msgCh {
			if err := stream.SendMsg(msg); err != nil {
				errCh <- err
				return
			}
		}
		errCh <- nil
	}()

	// 读循环：读取客户端消息并广播
	for {
		msg := &ChatMessage{}
		if err := stream.RecvMsg(msg); err != nil {
			if err == io.EOF {
				break
			}
			log.Printf("[chat] %s recv error: %v", userName, err)
			break
		}
		msg.Time = time.Now().Unix()
		log.Printf("[chat] %s: %s", msg.From, msg.Text)
		s.room.broadcast(userName, msg)
	}

	// 广播下线通知
	s.room.broadcast(userName, &ChatMessage{
		From: "system", Text: userName + " left", Time: time.Now().Unix(),
	})

	select {
	case err := <-errCh:
		return err
	default:
		return nil
	}
}

// === 注册 ===
func _Chat_JoinChat_Handler(srv interface{}, stream grpc.ServerStream) error {
	return srv.(*chatServer).JoinChat(stream)
}

func RegisterChatServiceServer(s *grpc.Server, srv *chatServer) {
	s.RegisterService(&grpc.ServiceDesc{
		ServiceName: "chat.ChatService",
		HandlerType: (*chatServer)(nil),
		Methods:     []grpc.MethodDesc{},
		Streams: []grpc.StreamDesc{
			{StreamName: "JoinChat", Handler: _Chat_JoinChat_Handler, ServerStreams: true, ClientStreams: true},
		},
		Metadata: "chat.proto",
	}, srv)
}

// === 拦截器 ===
func recoveryStreamInterceptor(srv interface{}, ss grpc.ServerStream,
	info *grpc.StreamServerInfo, handler grpc.StreamHandler,
) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = status.Errorf(codes.Internal, "panic: %v", r)
			log.Printf("[recovery] panic in %s: %v", info.FullMethod, r)
		}
	}()
	return handler(srv, ss)
}

// === main ===
func main() {
	lis, err := net.Listen("tcp", ":50054")
	if err != nil {
		log.Fatalf("listen: %v", err)
	}

	room := newChatRoom()
	server := grpc.NewServer(
		grpc.ChainStreamInterceptor(recoveryStreamInterceptor),
		grpc.MaxConcurrentStreams(128),
	)
	RegisterChatServiceServer(server, &chatServer{room: room})

	go func() {
		log.Printf("chat server listening on %s", lis.Addr())
		if err := server.Serve(lis); err != nil {
			log.Printf("serve: %v", err)
		}
	}()

	quit := make(chan os.Signal, 1)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	log.Println("shutting down chat server...")
	server.GracefulStop()
	log.Println("chat server stopped")

	_ = context.Background
}

这个聊天服务演示了双向流最经典的模式：读写分离在两个 goroutine 中并行，通过 channel 串联广播逻辑。它是理解 gRPC 流式通信的最佳实践模板。

九、小结

本篇深入 gRPC 服务端的实现细节：

Server 配置：消息大小限制、Keepalive 参数、TLS 凭证、并发流数，每一项都有生产环境的具体含义。
服务注册：理解了手写 ServiceDesc 的过程，就能看懂 protoc 生成的注册代码。
四种方法：Unary、Server Stream、Client Stream、Bidi Stream，每种都有独特的 handler 签名和收发模式。
错误处理：流式 RPC 中 io.EOF 表示正常结束，其他 error 表示异常终止。
拦截器链：洋葱模型执行顺序，recovery 最外、auth 最内。
优雅关停：GracefulStop + 超时兜底是生产标配。
完整示例：聊天服务展示了双向流读写分离并行的经典模式。

下一篇将转向客户端，深入连接管理、重试、负载均衡和连接状态监控。

入门篇

进阶篇

资深篇

gRPC 服务端深度实现

一、gRPC 服务端架构

二、Server 结构体配置

1. 消息大小限制

2. Keepalive 参数

3. TLS 配置

4. 完整的生产级 Server 配置

三、注册服务

四、四种服务方法实现详解

1. Unary RPC：一元调用

2. Server Streaming：服务端流

3. Client Streaming：客户端流

4. Bidirectional Streaming：双向流

五、流式 RPC 的错误处理

六、服务端拦截器

1. 一元拦截器

2. 流拦截器

3. 拦截器链的执行顺序

七、优雅关停：GracefulStop

八、完整示例：聊天服务（双向流）

九、小结

gRPC 服务端深度实现 ​

一、gRPC 服务端架构 ​

二、Server 结构体配置 ​

1. 消息大小限制 ​

2. Keepalive 参数 ​

3. TLS 配置 ​

4. 完整的生产级 Server 配置 ​

三、注册服务 ​

四、四种服务方法实现详解 ​

1. Unary RPC：一元调用 ​

2. Server Streaming：服务端流 ​

3. Client Streaming：客户端流 ​

4. Bidirectional Streaming：双向流 ​

五、流式 RPC 的错误处理 ​

六、服务端拦截器 ​

1. 一元拦截器 ​

2. 流拦截器 ​

3. 拦截器链的执行顺序 ​

七、优雅关停：GracefulStop ​

八、完整示例：聊天服务（双向流） ​

九、小结 ​

gRPC 服务端深度实现

一、gRPC 服务端架构

二、Server 结构体配置

1. 消息大小限制

2. Keepalive 参数

3. TLS 配置

4. 完整的生产级 Server 配置

三、注册服务

四、四种服务方法实现详解

1. Unary RPC：一元调用

2. Server Streaming：服务端流

3. Client Streaming：客户端流

4. Bidirectional Streaming：双向流

五、流式 RPC 的错误处理

六、服务端拦截器

1. 一元拦截器

2. 流拦截器

3. 拦截器链的执行顺序

七、优雅关停：GracefulStop

八、完整示例：聊天服务（双向流）

九、小结