3、粘包拆包现象复现

说实话，很多搞网络编程的朋友，第一次遇到粘包拆包时都是一脸懵。数据发出去好好的，怎么收回来就乱套了？

我当年第一次在生产环境遇到这个问题，是在一个实时交易系统里。客户端报了一笔订单，服务端解析出来的金额居然多了几位数。排查了半天，才发现是粘包搞的鬼。从那以后，我就养成了一个习惯——先复现，再分析，最后根治。

这一章，咱们就亲手把粘包和拆包这两个"妖怪"给揪出来。用 Python 写几行代码，再用 Wireshark 看看它们到底长什么样。

3.1 粘包场景复现

粘包，说白了就是多个数据包粘在一起被接收方一次性读走了。为什么会这样？因为 TCP 是流式协议，它不关心你发了几次 write，只关心字节流。

核心原因：Nagle 算法 + TCP 缓冲区合并

发送方连续发送小数据包时，Nagle 算法会合并它们，减少网络小包数量。接收方如果读得慢，缓冲区里就会积压多个包。

来看一段能稳定复现粘包的代码：

# server.py - 粘包复现服务端
import socket

def tcp_server():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('127.0.0.1', 8888))
    server_socket.listen(5)
    print("[服务端] 等待连接...")
    
    conn, addr = server_socket.accept()
    print(f"[服务端] 客户端已连接: {addr}")
    
    # 故意只读一次，模拟粘包
    data = conn.recv(1024)
    print(f"[服务端] 一次读取到的数据: {data}")
    print(f"[服务端] 数据长度: {len(data)} 字节")
    
    conn.close()
    server_socket.close()

if __name__ == '__main__':
    tcp_server()

# client.py - 粘包复现客户端
import socket
import time

def tcp_client():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(('127.0.0.1', 8888))
    
    # 连续发送3个小包
    messages = [b'Hello', b'World', b'TCP']
    for msg in messages:
        client_socket.send(msg)
        print(f"[客户端] 发送: {msg}")
        time.sleep(0.01)  # 极短间隔，触发Nagle合并
    
    time.sleep(1)
    client_socket.close()

if __name__ == '__main__':
    tcp_client()

运行结果会让你印象深刻：

[服务端] 一次读取到的数据: b'HelloWorldTCP'
[服务端] 数据长度: 13 字节

看到了吗？三个独立的消息粘成了一个整体。服务端根本分不清哪里是"Hello"的结尾，哪里是"World"的开头。

我的经验：粘包最容易出现在高频小数据包场景。比如心跳包、状态上报、日志采集。我曾经在一个物联网项目里，设备每100ms上报一次温度数据，结果服务端收到的全是粘在一起的字节流，解析出来的温度值完全乱套。

3.2 拆包场景复现

拆包正好相反——一个完整的数据包被拆成了多次读取。这通常发生在数据包较大，或者接收缓冲区设置得太小时。

# server_split.py - 拆包复现服务端
import socket

def tcp_server_split():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(('127.0.0.1', 8889))
    server_socket.listen(5)
    print("[服务端] 等待连接...")
    
    conn, addr = server_socket.accept()
    print(f"[服务端] 客户端已连接: {addr}")
    
    # 故意设置很小的缓冲区，模拟拆包
    buffer_size = 5
    while True:
        chunk = conn.recv(buffer_size)
        if not chunk:
            break
        print(f"[服务端] 读取到片段: {chunk}")
    
    conn.close()
    server_socket.close()

if __name__ == '__main__':
    tcp_server_split()

# client_split.py - 拆包复现客户端
import socket

def tcp_client_split():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(('127.0.0.1', 8889))
    
    # 发送一个较大的数据包
    large_msg = b'A' * 20
    client_socket.send(large_msg)
    print(f"[客户端] 发送: {large_msg} (长度: {len(large_msg)})")
    
    client_socket.close()

if __name__ == '__main__':
    tcp_client_split()

运行结果：

[服务端] 读取到片段: b'AAAAA'
[服务端] 读取到片段: b'AAAAA'
[服务端] 读取到片段: b'AAAAA'
[服务端] 读取到片段: b'AAAAA'

一个20字节的数据包，被拆成了4个5字节的片段。如果这是你的业务数据，解析逻辑就得处理这种"不完整"的情况。

注意：拆包不是bug，是TCP的正常行为。recv()返回的数据长度，永远小于等于你指定的缓冲区大小。千万别假设一次recv就能拿到完整消息。

3.3 用Wireshark观察现象

光看代码输出还不够，咱们得用Wireshark看看网络层面到底发生了什么。我个人习惯用Wireshark来验证自己的分析是否正确。

抓包步骤：

启动Wireshark，选择回环接口（Loopback: lo0 或 Npcap Loopback Adapter）
设置过滤条件：tcp.port == 8888
先启动服务端，再启动客户端
观察数据包列表

你会看到这样的现象：

序号	源端口	目标端口	长度	Info
1	12345	8888	13	PSH, ACK - 合并后的数据
2	8888	12345	0	ACK

注意看：客户端明明调用了3次send()，但Wireshark里只看到1个数据包，长度正好是13字节（Hello + World + TCP）。这就是Nagle算法合并的结果。

关键观察点：

粘包时：多个send()对应1个TCP segment
拆包时：1个send()对应多个TCP segments
正常时：1个send()对应1个TCP segment（但业务层仍需处理边界）

3.4 不同网络延迟下的表现差异

你想想看，粘包拆包的表现和网络延迟关系很大。我在不同环境下测试过，差异非常明显：

网络环境	RTT	粘包概率	拆包概率	典型表现
本地回环	< 0.1ms	极高	低	小包几乎100%粘包
局域网	0.5-2ms	高	中	频繁粘包，偶发拆包
广域网	20-100ms	中	高	大包易拆包，小包仍可能粘
跨洲链路	> 200ms	低	极高	几乎每次大包都被拆

为什么会这样？说白了就是时间窗口的问题。本地回环延迟极低，Nagle算法很容易在发送方把多个小包合并。而跨洲链路延迟高，TCP窗口调整频繁，数据包在传输过程中更容易被拆分。

我曾经踩过的坑：在开发环境（本地回环）测试一切正常，一上生产环境（跨机房）就出问题。后来才发现，开发环境因为延迟低，粘包现象不明显；生产环境延迟高，拆包频繁，导致解析逻辑崩溃。所以，一定要在不同网络环境下测试你的粘包拆包处理逻辑。

3.5 问题复现的代码模板

为了方便你快速复现和测试，我整理了一个通用模板。以后遇到类似问题，直接套用就行。

# tcp_sticky_split_template.py
import socket
import threading
import time

class TCPStickySplitTest:
    """粘包拆包测试模板"""
    
    def __init__(self, host='127.0.0.1', port=9999):
        self.host = host
        self.port = port
        self.server_socket = None
        self.client_socket = None
    
    def start_server(self, buffer_size=1024, delay_recv=False):
        """启动服务端"""
        def server_thread():
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(1)
            
            conn, addr = self.server_socket.accept()
            print(f"[服务端] 连接来自: {addr}")
            
            while True:
                data = conn.recv(buffer_size)
                if not data:
                    break
                print(f"[服务端] 收到 {len(data)} 字节: {data}")
                if delay_recv:
                    time.sleep(0.5)  # 模拟慢读取，加剧粘包
            
            conn.close()
        
        threading.Thread(target=server_thread, daemon=True).start()
        time.sleep(0.1)
    
    def send_messages(self, messages, interval=0.01):
        """发送多条消息"""
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client_socket.connect((self.host, self.port))
        
        for msg in messages:
            self.client_socket.send(msg)
            print(f"[客户端] 发送 {len(msg)} 字节: {msg}")
            time.sleep(interval)
        
        time.sleep(1)
        self.client_socket.close()
    
    def cleanup(self):
        """清理资源"""
        if self.server_socket:
            self.server_socket.close()

# 使用示例
if __name__ == '__main__':
    test = TCPStickySplitTest()
    
    # 测试粘包：小包+短间隔
    print("=== 测试粘包场景 ===")
    test.start_server(buffer_size=1024)
    test.send_messages([b'Hello', b'World', b'TCP'], interval=0.01)
    test.cleanup()
    time.sleep(1)
    
    # 测试拆包：大包+小缓冲区
    print("\n=== 测试拆包场景 ===")
    test.start_server(buffer_size=5)
    test.send_messages([b'A' * 20], interval=0)
    test.cleanup()

这个模板有几个好处：

可配置缓冲区大小：调小buffer_size模拟拆包，调大模拟粘包
可控制发送间隔：间隔越短，粘包越严重
支持延迟接收：服务端读得慢，粘包概率更高
线程安全：服务端和客户端在不同线程运行，方便调试

重要提醒：这个模板只是用来复现现象的，不是生产代码。生产环境绝对不能依赖"一次recv就能拿到完整消息"这种假设。下一章我们会讲如何正确解决这个问题。

3.6 本章知识体系

下面这张图，帮你把本章的核心逻辑串起来：

嗯，这一章的内容就到这儿。你可以在自己的机器上跑一遍代码，亲眼看看粘包拆包到底是怎么回事。只有亲手复现过，后面学解决方案时才会有更深的体会。

📚 课程目录

3、粘包拆包现象复现

3.1 粘包场景复现

3.2 拆包场景复现

3.3 用Wireshark观察现象

3.4 不同网络延迟下的表现差异

3.5 问题复现的代码模板

3.6 本章知识体系