Socket大消息分块传输异常：接收端无法获取块数

问题背景 #

我编写了一套基于Socket的客户端与服务器数据传输代码，可传输字符串及自定义类对象。由于部分消息大小超过1024字节，采用分块传输策略：先计算所需的1024字节块数并发送，让接收端据此接收对应块数的数据，但接收端无法正确获取块数。

发送端原代码 #

for message in messages:
    client,msg=message
    print(msg)
    data = pickle.dumps(msg)
    print(data)
    if client in wlist:
        # Calculate the number of chunks needed
        num_chunks = (len(data) + 1023) // 1024
        # Send the number of chunks first
        pickle_num=pickle.dumps(num_chunks)
        print(pickle_num)
        client.send(pickle_num)
        # Send data in chunks of 1024 bytes
        for i in range(num_chunks):
            start_index = i * 1024
            end_index = min((i + 1) * 1024, len(data))
            chunk = data[start_index:end_index]
            client.send(chunk)
        print(num_chunks)
        messages.remove(message)

接收端原代码 #

def client_socket(client,msg_send,locker,msg_recieve):
    def receive_data(client, message):
        # Receive the number of chunks
        num_chunks = pickle.loads(message)
        # Receive data chunks
        received_data = ""
        for _ in range(num_chunks):
            chunk = client.recv(1024)
            received_data += chunk
        print(received_data)
        data = pickle.loads(received_data)
        print(data)
        # Unpickle the received data
        return data
    while True:
        locker.acquire()
        for i in msg_send:
            msg=pickle.dumps(i)
            client.send(msg)
            msg_send.remove(i)
        locker.release()
        rlist, wlist, xlist = select.select([client] ,[client], [], 0.5)
        for client in rlist:
            try:
                message=client.recv(1024)
            except:
                #תקלת תקשורת
                messagebox.showerror('error', ' communication failed')
            else:
                if message != "":
                    msg = receive_data(client, message)
                    locker.acquire()
                    msg_recieve.append(msg)
                    locker.release()

问题根源分析 #

• TCP流式协议的粘包/拆包问题：client.recv(1024)无法保证一次性接收完整的块数pickle数据，可能只收到部分字节，或者后续数据块与块数数据粘在一起，导致pickle.loads解析失败。
• 字节与字符串类型不匹配：接收端received_data初始化为字符串""，但recv返回bytes类型，直接拼接会引发类型错误。
• 未确保数据完整发送：client.send()的返回值是实际发送的字节数，若网络缓冲区已满，无法一次性发送全部数据，导致接收端数据不完整。
• 遍历列表时删除元素的逻辑漏洞：接收端遍历msg_send并删除元素，会导致部分消息被漏发（遍历过程中列表长度变化，跳过后续元素）。

修复方案 #

1. 封装可靠的发送/接收函数，解决粘包问题 #

TCP是流式协议，需要明确的边界分割数据。通过先发送固定长度的数据长度（4字节大端整数），再发送实际数据，确保接收端能完整获取数据段。

import struct

def send_all(sock, data):
    # 先发送数据长度（4字节大端格式）
    length = struct.pack('!I', len(data))
    sent = 0
    while sent < len(length):
        sent += sock.send(length[sent:])
    # 循环发送实际数据，直到全部发送完成
    sent = 0
    while sent < len(data):
        sent += sock.send(data[sent:])

def recv_exact(sock, length):
    # 接收指定长度的字节数据，确保完整接收
    received = b""
    while len(received) < length:
        chunk = sock.recv(length - len(received))
        if not chunk:
            raise ConnectionError("Connection closed")
        received += chunk
    return received

2. 修改发送端代码 #

使用封装的send_all函数发送块数和数据块，同时修复遍历删除列表元素的问题：

for message in messages[:]:  # 复制列表，避免遍历过程中删除元素导致漏处理
    client, msg = message
    print(msg)
    data = pickle.dumps(msg)
    print(data)
    if client in wlist:
        num_chunks = (len(data) + 1023) // 1024
        # 发送块数的pickle数据
        pickle_num = pickle.dumps(num_chunks)
        send_all(client, pickle_num)
        # 分块发送实际数据
        for i in range(num_chunks):
            start_index = i * 1024
            end_index = min((i + 1) * 1024, len(data))
            chunk = data[start_index:end_index]
            send_all(client, chunk)
        print(num_chunks)
        messages.remove(message)

3. 修改接收端代码 #

修复类型错误，使用recv_exact确保完整接收数据，同时优化消息发送逻辑：

import struct
import pickle
import select
import tkinter.messagebox as messagebox

# 复用send_all和recv_exact函数
def send_all(sock, data):
    length = struct.pack('!I', len(data))
    sent = 0
    while sent < len(length):
        sent += sock.send(length[sent:])
    sent = 0
    while sent < len(data):
        sent += sock.send(data[sent:])

def recv_exact(sock, length):
    received = b""
    while len(received) < length:
        chunk = sock.recv(length - len(received))
        if not chunk:
            raise ConnectionError("Connection closed")
        received += chunk
    return received

def client_socket(client, msg_send, locker, msg_recieve):
    def receive_data(client):
        # 先接收块数数据的长度
        num_len = struct.unpack('!I', recv_exact(client, 4))[0]
        # 接收完整的块数pickle数据
        pickle_num = recv_exact(client, num_len)
        num_chunks = pickle.loads(pickle_num)
        # 接收所有数据块
        received_data = b""
        for _ in range(num_chunks):
            # 先接收当前块的长度
            chunk_len = struct.unpack('!I', recv_exact(client, 4))[0]
            chunk = recv_exact(client, chunk_len)
            received_data += chunk
        print(received_data)
        data = pickle.loads(received_data)
        print(data)
        return data
    
    while True:
        locker.acquire()
        # 复制列表遍历，避免删除元素导致漏发
        for msg in list(msg_send):
            data = pickle.dumps(msg)
            send_all(client, data)
            msg_send.remove(msg)
        locker.release()
        
        rlist, wlist, xlist = select.select([client], [client], [], 0.5)
        for sock in rlist:
            try:
                msg = receive_data(sock)
                locker.acquire()
                msg_recieve.append(msg)
                locker.release()
            except ConnectionError:
                messagebox.showerror('error', 'Communication failed')
            except Exception as e:
                messagebox.showerror('error', f'Error: {str(e)}')

关键修复点说明 #

• 粘包问题解决：通过"长度+数据"的传输格式，让接收端明确每个数据段的边界，确保完整接收。
• 类型错误修复：将received_data初始化为b""，统一使用bytes类型处理二进制数据。
• 可靠发送：send_all循环发送直到所有字节都被发送，避免因网络缓冲区导致的数据丢失。
• 遍历删除逻辑优化：使用list(msg_send)复制列表后再遍历，删除原列表元素不会影响遍历过程，确保所有消息都被发送。

内容的提问来源于stack exchange，提问作者some one