《游戏服务端编程实践》5.4.1 消息转发核心代码示例

从客户端连接、网关解帧、路由查表到后端回包的完整链路出发,拆解固定帧头、不透明 Body、Session 读写超时与 BackendConn 串行化设计,并用 net.Pipe 写出可运行的端到端测试,覆盖未知路由、后端超时与断线丢弃等失败分支。

一、最小闭环

网关的职责不是理解业务,而是把一帧从客户端搬到正确的后端,再把结果搬回来。核心链路可以压成三步,每一步都要能单独失败、单独观测:

阶段输入网关动作失败时的回包
解帧长度前缀校验长度与版本,读固定头40001 协议错误
路由Route查本地路由表40401 未知路由
转发原始 Body调后端 RoundTrip50301 或 50401

二、帧头固定,Body 不透明

网关只读 12 字节固定头:版本、类型、路由、请求 ID,后面的 Body 原样透传。这样新增业务字段不需要改网关,网关也不会因为反序列化失败而误伤请求。长度前缀同时承担分帧和安全边界:超过 64 KiB 直接拒绝,避免对端用一个假长度申请巨量内存。

偏移字段长度说明
0Length4只算头加 Body,不含自身
4Version1不匹配直接拒绝
5Type1请求、响应、错误
6Route2网关路由键,不是业务 ID
8RequestID8客户端生成,全链路保持
16BodyN网关不解析

三、Session 与读写超时

Session 包住一条客户端连接,读由 ReadFrame 驱动,写统一走 WriteFrame。写超时必须设置:客户端半死不活时,Write 会一直阻塞,最终把网关的 goroutine 和内存拖住。读超时则由连接空闲策略控制,本章只保留写侧,避免示例膨胀。

四、Router 与 BackendConn

Forwarder 持有 map[uint16]Backend,路由键在配置下发时固定,不在请求路径上做字符串匹配。BackendConn 是一条到后端的长连接,内部用互斥锁把“写请求、读响应”做成原子操作。没有这把锁,两个并发请求的响应会按到达顺序交错,客户端拿到别人的包。

五、转发实现

// file: ch05/backend/forward.go
// tier: L1
// verify: go test ./ch05/backend/

package backend

import (
	"context"
	"encoding/binary"
	"errors"
	"io"
	"net"
	"sync"
	"time"
)

const (
	frameVersion = 1
	headerSize   = 12 // version(1) + type(1) + route(2) + requestID(8)
	maxFrameSize = 64 << 10

	FrameRequest  uint8 = 1
	FrameResponse uint8 = 2
	FrameError    uint8 = 3
)

var ErrFrameTooLarge = errors.New("backend: frame too large")

// Frame 是网关与后端之间的最小信封。Body 对网关不透明,网关只读固定头。
type Frame struct {
	Type      uint8
	Route     uint16
	RequestID uint64
	Body      []byte
}

func (f Frame) Encode() []byte {
	raw := make([]byte, 4+headerSize+len(f.Body))
	binary.BigEndian.PutUint32(raw[0:4], uint32(headerSize+len(f.Body)))
	raw[4], raw[5] = frameVersion, f.Type
	binary.BigEndian.PutUint16(raw[6:8], f.Route)
	binary.BigEndian.PutUint64(raw[8:16], f.RequestID)
	copy(raw[16:], f.Body)
	return raw
}

// ReadFrame 先读长度再读帧体;max 防止对端用超大长度拖垮内存。
func ReadFrame(r io.Reader, max uint32) (Frame, error) {
	var prefix [4]byte
	if _, err := io.ReadFull(r, prefix[:]); err != nil {
		return Frame{}, err
	}
	n := binary.BigEndian.Uint32(prefix[:])
	if max == 0 {
		max = maxFrameSize
	}
	if n < headerSize {
		return Frame{}, errors.New("backend: short frame")
	}
	if n > max {
		return Frame{}, ErrFrameTooLarge
	}
	raw := make([]byte, n)
	if _, err := io.ReadFull(r, raw); err != nil {
		return Frame{}, err
	}
	if raw[0] != frameVersion {
		return Frame{}, errors.New("backend: bad frame version")
	}
	return Frame{
		Type:      raw[1],
		Route:     binary.BigEndian.Uint16(raw[2:4]),
		RequestID: binary.BigEndian.Uint64(raw[4:12]),
		Body:      raw[12:],
	}, nil
}

type Session struct{ Conn net.Conn }

func (s *Session) ReadFrame() (Frame, error) { return ReadFrame(s.Conn, maxFrameSize) }

func (s *Session) WriteFrame(f Frame) error {
	_ = s.Conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
	defer s.Conn.SetWriteDeadline(time.Time{})
	_, err := s.Conn.Write(f.Encode())
	return err
}

func (s *Session) Close() error { return s.Conn.Close() }

// Backend 把“送一帧、取一帧”抽象出来,实现可以是长连接、RPC 或内存 hub。
type Backend interface {
	RoundTrip(ctx context.Context, req Frame) (Frame, error)
	Close() error
}

// BackendConn 用一条 TCP 长连接做一问一答。互斥锁串行化并发请求,
// 否则两个请求的响应会互相串包;代价是单连接吞吐有上限。
type BackendConn struct {
	mu     sync.Mutex
	conn   net.Conn
	broken bool
}

func NewBackendConn(conn net.Conn) *BackendConn { return &BackendConn{conn: conn} }

func (b *BackendConn) RoundTrip(ctx context.Context, req Frame) (Frame, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.broken {
		return Frame{}, errors.New("backend: connection discarded")
	}
	if err := ctx.Err(); err != nil {
		return Frame{}, err
	}
	if deadline, ok := ctx.Deadline(); ok {
		_ = b.conn.SetDeadline(deadline)
		defer b.conn.SetDeadline(time.Time{})
	}
	if _, err := b.conn.Write(req.Encode()); err != nil {
		b.broken = true
		_ = b.conn.Close()
		return Frame{}, err
	}
	resp, err := ReadFrame(b.conn, maxFrameSize)
	if err != nil {
		// 超时后对端可能仍会回包;不关闭连接会把迟到响应当成下一单。
		b.broken = true
		_ = b.conn.Close()
		if ne, ok := err.(net.Error); ok && ne.Timeout() {
			return Frame{}, context.DeadlineExceeded
		}
		return Frame{}, err
	}
	resp.RequestID, resp.Route, resp.Type = req.RequestID, req.Route, FrameResponse
	return resp, nil
}

func (b *BackendConn) Close() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.broken = true
	return b.conn.Close()
}

// Forwarder 按 Route 选后端,转发请求并回写响应。它不碰 Body。
type Forwarder struct {
	routes  map[uint16]Backend
	timeout time.Duration
}

func NewForwarder(timeout time.Duration) *Forwarder {
	return &Forwarder{routes: make(map[uint16]Backend), timeout: timeout}
}

func (f *Forwarder) Route(route uint16, backend Backend) { f.routes[route] = backend }

func (f *Forwarder) Serve(conn net.Conn) {
	sess := &Session{Conn: conn}
	defer sess.Close()
	for {
		req, err := sess.ReadFrame()
		if err != nil {
			return
		}
		f.handle(sess, req)
	}
}

func (f *Forwarder) handle(sess *Session, req Frame) {
	backend, ok := f.routes[req.Route]
	if !ok {
		_ = sess.WriteFrame(errorFrame(req, 40401, "unknown route"))
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), f.timeout)
	defer cancel()
	resp, err := backend.RoundTrip(ctx, req)
	if err != nil {
		code, msg := uint16(50301), "backend unavailable"
		if errors.Is(err, context.DeadlineExceeded) {
			code, msg = 50401, "backend timeout"
		}
		_ = sess.WriteFrame(errorFrame(req, code, msg))
		return
	}
	// 后端漏回请求 ID 时由网关补齐,客户端才能把响应配回请求。
	resp.RequestID, resp.Type = req.RequestID, FrameResponse
	_ = sess.WriteFrame(resp)
}

func errorFrame(req Frame, code uint16, msg string) Frame {
	body := make([]byte, 2+len(msg))
	binary.BigEndian.PutUint16(body[:2], code)
	copy(body[2:], msg)
	return Frame{Type: FrameError, Route: req.Route, RequestID: req.RequestID, Body: body}
}

六、端到端测试

测试用两段 net.Pipe 串起客户端、网关和后端:第一段模拟玩家连接,第二段模拟后端进程。请求 Body 里故意放入一个看似合法帧头的内容,后端原样回显,以此证明网关没有重新序列化业务字段。

// file: ch05/backend/forward_test.go
// tier: L1
// verify: go test ./ch05/backend/

package backend

import (
	"bytes"
	"encoding/binary"
	"net"
	"testing"
	"time"
)

// newGateway 用两段 net.Pipe 串起客户端、网关和后端进程。
func newGateway(t *testing.T, fn func(Frame) (Frame, error), timeout time.Duration) net.Conn {
	t.Helper()
	backendClient, backendServer := net.Pipe()
	go func() {
		defer backendServer.Close()
		for {
			req, err := ReadFrame(backendServer, 0)
			if err != nil {
				return
			}
			resp, err := fn(req)
			if err != nil {
				return
			}
			if _, err := backendServer.Write(resp.Encode()); err != nil {
				return
			}
		}
	}()
	backend := NewBackendConn(backendClient)
	t.Cleanup(func() { _ = backend.Close() })
	client, server := net.Pipe()
	fwd := NewForwarder(timeout)
	fwd.Route(7, backend)
	go fwd.Serve(server)
	t.Cleanup(func() { _ = client.Close() })
	return client
}

func call(t *testing.T, conn net.Conn, req Frame) Frame {
	t.Helper()
	if _, err := conn.Write(req.Encode()); err != nil {
		t.Fatalf("write: %v", err)
	}
	resp, err := ReadFrame(conn, 0)
	if err != nil {
		t.Fatalf("read: %v", err)
	}
	return resp
}

func errorCode(f Frame) uint16 {
	if f.Type != FrameError || len(f.Body) < 2 {
		return 0
	}
	return binary.BigEndian.Uint16(f.Body[:2])
}

func TestForwardEndToEnd(t *testing.T) {
	// 后端原样回显 body,并故意回错的 RequestID,验证网关按请求改写。
	conn := newGateway(t, func(req Frame) (Frame, error) {
		return Frame{Type: FrameResponse, Route: req.Route, RequestID: 999, Body: req.Body}, nil
	}, time.Second)
	bad := call(t, conn, Frame{Type: FrameRequest, Route: 8, RequestID: 6, Body: []byte("x")})
	if bad.Type != FrameError || bad.RequestID != 6 || errorCode(bad) != 40401 {
		t.Fatalf("unknown route: %+v code=%d", bad, errorCode(bad))
	}
	// body 里塞入一个看似合法帧头的内容,证明网关不做业务反序列化。
	body := []byte{0, 0, 0, 12, 0xff, 0x00, 0x7f, 0x80, 0x01}
	resp := call(t, conn, Frame{Type: FrameRequest, Route: 7, RequestID: 42, Body: body})
	if resp.Type != FrameResponse || resp.RequestID != 42 || resp.Route != 7 || !bytes.Equal(resp.Body, body) {
		t.Fatalf("resp = %+v", resp)
	}
}

func TestForwardBackendTimeout(t *testing.T) {
	conn := newGateway(t, func(req Frame) (Frame, error) {
		time.Sleep(300 * time.Millisecond)
		return Frame{Type: FrameResponse, Body: req.Body}, nil
	}, 40*time.Millisecond)
	start := time.Now()
	resp := call(t, conn, Frame{Type: FrameRequest, Route: 7, RequestID: 11, Body: []byte("slow")})
	if elapsed := time.Since(start); resp.Type != FrameError || errorCode(resp) != 50401 || elapsed > 250*time.Millisecond {
		t.Fatalf("resp=%+v code=%d elapsed=%v", resp, errorCode(resp), elapsed)
	}
	// 超时后连接被丢弃,下一次请求快速失败而不是读到迟到的响应。
	second := call(t, conn, Frame{Type: FrameRequest, Route: 7, RequestID: 12, Body: []byte("again")})
	if second.Type != FrameError || errorCode(second) != 50301 {
		t.Fatalf("second = %+v code=%d", second, errorCode(second))
	}
}

七、失败场景:后端超时与断线

后端超时后,BackendConn 会把连接标记为损坏并关闭,而不是留着等下一条请求。原因很实际:超时的请求可能稍后仍然回包,如果连接继续复用,这个迟到的响应就会被当成下一单的结果;测试里的第二次请求因此快速得到 50301。未知路由在网关本地直接拒绝,不占用后端连接;后端不可达统一映射为 50301,超时映射为 50401。错误包同样携带原请求的 RequestID,客户端才能把失败配回具体的业务调用。

八、运行

cd .local/checks/gs-practice-lab
gofmt -l ch05/backend
go vet ./ch05/backend/
go test ./ch05/backend/ -count=1

本机输出 ok gspractice/ch05/backend 0.463s;gofmt -l 与 go vet 均无输出。

小结

转发核心只有三件事:固定帧头让网关可解析,Body 不透明让网关可演进,RequestID 贯穿让客户端可关联。Session 负责连接生命周期,Forwarder 负责路由,BackendConn 负责一问一答的串行化;超时后丢弃连接是防止串包的关键动作。

常见错误与边界

  • 转发时把 Body 反序列化再序列化,未知字段被丢掉,版本升级后老网关静默改坏数据。
  • 后端超时只回错误、不关闭连接,迟到的响应被下一单读到,出现“张冠李戴”的回包。
  • 多个 goroutine 共用一个后端连接却不加锁,响应按到达顺序串包。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「game」更多文章

  1. 《游戏服务端编程实践》附录 D:参考资料与社区索引
  2. 《游戏服务端编程实践》附录 C:示例项目目录结构
  3. 《游戏服务端编程实践》附录 B:性能调优 Checklist