3. 网络层与协议层：Socket通信、消息编解码、粘包拆包处理，以及如何与业务层解耦

好，咱们今天聊聊网络层和协议层。这一层说白了，就是游戏服务器跟客户端之间的“电话线”。你业务逻辑写得再漂亮，消息传不过去，或者传过去是乱码，那全白搭。

我个人习惯把网络层和协议层放在一起讲，因为它们俩是密不可分的。网络层负责“怎么传”，协议层负责“传什么”。咱们的目标很明确：让底层通信的细节，不要污染到上层的业务逻辑。

3.1 Socket通信：选对模式，事半功倍

Python里做Socket通信，最常用的就是 asyncio 了。它基于事件循环，非常适合处理大量长连接。

我见过不少新手，直接用多线程去处理每个连接。嗯，对于几百人的小游戏还行，一旦上了几千人，线程切换的开销就能把CPU吃满。用 asyncio 的话，一个线程就能搞定成千上万的连接，这才是游戏服务器的正确姿势。

核心思路： 每个客户端连接，对应一个 asyncio 的协程任务。这个任务只负责收发原始字节流，不关心字节流里是什么内容。

import asyncio

class NetworkServer:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self._connections = {}  # 连接ID -> (reader, writer)

    async def start(self):
        server = await asyncio.start_server(
            self._handle_connection, self.host, self.port
        )
        print(f'Server started on {self.host}:{self.port}')
        async with server:
            await server.serve_forever()

    async def _handle_connection(self, reader, writer):
        # 每个连接进来，分配一个唯一的ID
        conn_id = id(writer)
        self._connections[conn_id] = (reader, writer)
        try:
            # 这里只负责读原始数据，然后交给协议层去解析
            while True:
                data = await reader.read(4096)
                if not data:
                    break
                # 注意：这里只是收到了原始字节，还没处理粘包
                await self._on_raw_data(conn_id, data)
        except asyncio.CancelledError:
            pass
        finally:
            del self._connections[conn_id]
            writer.close()
            await writer.wait_closed()

    async def _on_raw_data(self, conn_id, data):
        # 这个方法会被协议层重写，或者通过回调注入
        # 网络层不关心data是什么，只负责传递
        pass

你看，网络层只做一件事：收字节，发字节。它不知道什么是“登录”，什么是“移动”。这就是解耦的第一步。

3.2 消息编解码：定义你的“语言”

客户端发过来的是一串字节，你得把它变成业务层能理解的对象。反过来，业务层要发消息，也得把对象变成字节。

我建议用 Protocol Buffers（protobuf）来做编解码。为什么？因为它跨语言、效率高、而且有严格的 schema 定义。你在项目中定义好 .proto 文件，前后端都按这个来，不容易出错。

一个小技巧： 消息ID（或者叫消息号）一定要用整数，不要用字符串。整数比较快，而且方便做路由。我曾经见过有人用字符串“CMD_LOGIN”做消息ID，结果每次都要做字符串比较，性能差了一截。

# 假设我们有一个 protobuf 定义好的消息
# message LoginRequest {
#   string account = 1;
#   string password = 2;
# }

class ProtocolCodec:
    def __init__(self):
        # 消息ID到消息类的映射
        self._msg_id_map = {
            1001: LoginRequest,  # 1001 代表登录请求
            1002: LoginResponse,
            # ...
        }

    def decode(self, data: bytes):
        """将字节解码为 (msg_id, message_object)"""
        # 前2个字节是消息ID，后面是protobuf数据
        msg_id = int.from_bytes(data[:2], 'big')
        msg_class = self._msg_id_map.get(msg_id)
        if not msg_class:
            raise ValueError(f'Unknown message id: {msg_id}')
        msg_obj = msg_class()
        msg_obj.ParseFromString(data[2:])
        return msg_id, msg_obj

    def encode(self, msg_id: int, message) -> bytes:
        """将消息对象编码为字节"""
        header = msg_id.to_bytes(2, 'big')
        body = message.SerializeToString()
        return header + body

这里有个细节：消息ID占2个字节，也就是最大支持65535种消息。对于绝大多数游戏来说，完全够用了。如果你觉得不够，可以扩展到4个字节，但一般没必要。

3.3 粘包拆包处理：TCP的“老毛病”

为什么会粘包？因为TCP是流式协议。你发两次 send，底层可能合并成一次发送。或者你发一次，底层可能拆成多次接收。

嗯，这里要注意：绝对不能假设一次 recv 就收到一个完整的消息。我刚开始做游戏时，就犯过这个错。客户端发了一个登录包，我这边 recv 正好收到了完整的数据，一切正常。结果上线后，玩家一多，各种解析错误就冒出来了。排查了半天，才发现是粘包导致的。

避坑指南： 我曾经在一个项目里，看到有人用换行符 \n 来分割消息。这在文本协议里还行，但在二进制协议里，万一消息体里恰好有一个 0x0A（换行符的ASCII码），那就全乱了。千万别这么干。

正确的做法是：长度前缀法。在消息头里，用固定长度的字段表示消息体的长度。

class PacketAssembler:
    """负责从字节流中组装出完整的消息包"""
    def __init__(self):
        self._buffer = bytearray()
        # 包头长度：2字节消息ID + 2字节消息体长度
        self.HEADER_LEN = 4

    def feed(self, data: bytes):
        """喂入新收到的字节"""
        self._buffer.extend(data)

    def get_packets(self):
        """尝试从缓冲区中取出完整的消息包"""
        packets = []
        while len(self._buffer) >= self.HEADER_LEN:
            # 先读消息体长度（第2-3字节）
            body_len = int.from_bytes(self._buffer[2:4], 'big')
            total_len = self.HEADER_LEN + body_len

            if len(self._buffer) < total_len:
                # 数据还不够，等下一次
                break

            # 取出一个完整的包
            packet = bytes(self._buffer[:total_len])
            packets.append(packet)
            # 移除已处理的数据
            del self._buffer[:total_len]

        return packets

你看，这个 PacketAssembler 类，就是专门处理粘包拆包的。它维护一个内部缓冲区，每次有新数据进来，就尝试从中提取完整的消息包。提取不出来的，就留在缓冲区里，等下一次数据。

3.4 与业务层解耦：让网络层“闭嘴”

好了，现在网络层能收发了，协议层能编解码了，粘包拆包也处理了。接下来最关键的一步：怎么把这些东西跟业务层解耦？

我的做法是：定义一个消息分发器。网络层收到完整消息后，交给协议层解码，然后协议层把解码后的消息对象，通过分发器交给业务层。业务层只关心消息对象，完全不关心它是从哪来的、怎么编解码的。

class MessageDispatcher:
    def __init__(self):
        self._handlers = {}  # msg_id -> handler_function

    def register(self, msg_id: int, handler):
        """注册消息处理器"""
        self._handlers[msg_id] = handler

    async def dispatch(self, conn_id: int, msg_id: int, msg_obj):
        """分发消息到对应的处理器"""
        handler = self._handlers.get(msg_id)
        if not handler:
            print(f'No handler for msg_id: {msg_id}')
            return
        # 这里可以加一些前置处理，比如日志、鉴权等
        await handler(conn_id, msg_obj)

# 业务层示例
class LoginHandler:
    def __init__(self, dispatcher: MessageDispatcher):
        dispatcher.register(1001, self.handle_login)

    async def handle_login(self, conn_id: int, request: LoginRequest):
        # 这里只处理登录逻辑，不关心网络细节
        print(f'User {request.account} is logging in...')
        # ... 验证账号密码 ...
        response = LoginResponse()
        response.success = True
        # 发送响应，通过回调或者事件的方式
        await self._send_response(conn_id, 1002, response)

你看，LoginHandler 里完全没有 socket、recv、send 这些字眼。它只跟消息对象打交道。这就是解耦。

总结一下分层的好处：

  • 网络层：只负责Socket读写，不关心数据内容。
  • 协议层：只负责编解码和粘包拆包，不关心业务逻辑。
  • 业务层：只关心消息对象，不关心底层通信细节。

哪一层出了问题，就改哪一层，互不影响。这才是工程化的做法。

最后说一句，解耦不是一蹴而就的。你可以在初期先写一个简单的版本，把网络和业务混在一起。但随着项目变大，一定要及时重构。我见过太多项目，因为前期图省事，后期改得痛不欲生。嗯，希望你能少走这些弯路。

← 上一章 📖 返回目录 下一章 →