Python UDP Socket接收大数据包时recvfrom阻塞问题求助

问题根源 #

你的核心问题是对UDP协议特性理解有误：UDP是面向报文的无连接协议，客户端每个sendto()调用会生成一个独立的UDP数据报，服务器的recvfrom()每次只能接收一个完整的数据报，无法像TCP那样分块拆分读取。

当前代码的具体错误：

• 客户端用单个sendto()发送完整大数据，服务器却循环调用recvfrom()尝试分块读取，第一次recvfrom()已经拿到完整数据包，第二次调用会阻塞等待不存在的下一个数据包，导致程序卡住。
• 服务器循环中package_len -= receiverBufsize的逻辑错误，应该减去实际收到的字节数，而非固定减1024。

解决方案 #

方案1：处理不超过UDP最大长度的数据包（0-65507字节） #

UDP单个数据报最大有效载荷为65507字节（IP头+UDP头共占用48字节，65535-48=65507），若你的数据长度在这个范围内，直接修改服务器一次性接收完整数据报即可。

修正后的服务器代码

# -*- coding: utf-8 -*-
import json
import socket

ip_port = ('127.0.0.1', 8080)
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# SO_REUSEADDR只需在bind前设置一次
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(ip_port)
print('server listen......')

def recv_header(s_socket):
    try:
        msg, addr = s_socket.recvfrom(32)
        return json.loads(msg), addr
    except Exception as e:
        return str(e), None

while True:
    header_msg, client_addr = recv_header(s_socket=server)
    if client_addr is None:
        print(header_msg)
        continue
    data_body = b""
    package_len = int(header_msg['length'])
    print("package_len: {}".format(package_len))
    print("client_addr: {}".format(str(client_addr)))
    
    # 用足够大的缓冲区一次性接收完整UDP数据报
    body_part, addr = server.recvfrom(65535)
    # 验证数据来源是否为目标客户端，避免混杂其他请求
    if addr == client_addr:
        data_body = body_part
        # 可选：校验实际接收长度与声明长度是否一致
        if len(data_body) != package_len:
            print("警告：实际接收长度 {} 与声明长度 {} 不符".format(len(data_body), package_len))
    
    print("data_body: {}".format(data_body.decode('utf-8')))

修正后的客户端代码

# -*- coding: utf-8 -*-
import json
import socket
import random

ip_port = ('127.0.0.1', 8080)
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# UDP单个数据报最大有效载荷限制
MAX_UDP_PAYLOAD = 65507

def get_data(length):
    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789'
    return ''.join([random.choice(base_str) for _ in range(length)])

while 1:
    msg = raw_input('>>>：')
    if msg == 'q':
        break
    if not msg:
        continue
    try:
        data_len = int(msg)
        if data_len > MAX_UDP_PAYLOAD:
            print("错误：数据长度不能超过{}字节".format(MAX_UDP_PAYLOAD))
            continue
        data = get_data(data_len)
        header = json.dumps({"length": data_len})
        send_len = client.sendto(header.encode('utf-8'), ip_port)
        print("发送头部长度：{}".format(send_len))
        send_len = client.sendto(data.encode('utf-8'), ip_port)
        print("发送数据长度：{}".format(send_len))
    except ValueError:
        print("请输入有效数字")

方案2：支持超过65507字节的大文件传输 #

若必须发送超过UDP最大长度的数据，需要将数据拆分为多个UDP数据包，每个包携带序号标识，服务器按序号拼接重组数据。

客户端代码（分块发送）

# -*- coding: utf-8 -*-
import json
import socket
import random

ip_port = ('127.0.0.1', 8080)
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
client.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 每个UDP数据包的有效载荷大小
CHUNK_SIZE = 1024

def get_data(length):
    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789'
    return ''.join([random.choice(base_str) for _ in range(length)])

while 1:
    msg = raw_input('>>>：')
    if msg == 'q':
        break
    if not msg:
        continue
    try:
        data_len = int(msg)
        data = get_data(data_len).encode('utf-8')
        # 计算总块数
        total_chunks = (data_len + CHUNK_SIZE - 1) // CHUNK_SIZE
        # 发送头部：总长度、总块数
        header = json.dumps({"total_length": data_len, "total_chunks": total_chunks})
        client.sendto(header.encode('utf-8'), ip_port)
        print("发送头部：总长度{}，总块数{}".format(data_len, total_chunks))
        # 分块发送数据，每个块携带序号
        for idx in range(total_chunks):
            start = idx * CHUNK_SIZE
            end = start + CHUNK_SIZE
            chunk_data = data[start:end]
            # 发送块序号
            chunk_header = json.dumps({"chunk_idx": idx})
            client.sendto(chunk_header.encode('utf-8'), ip_port)
            send_len = client.sendto(chunk_data, ip_port)
            print("发送块{}，长度{}".format(idx, send_len))
    except ValueError:
        print("请输入有效数字")

服务器代码（按序拼接）

# -*- coding: utf-8 -*-
import json
import socket

ip_port = ('127.0.0.1', 8080)
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(ip_port)
print('server listen......')

CHUNK_SIZE = 1024

def recv_json(s_socket):
    try:
        msg, addr = s_socket.recvfrom(1024)
        return json.loads(msg), addr
    except Exception as e:
        return str(e), None

while True:
    # 接收总数据头部
    header_msg, client_addr = recv_json(server)
    if client_addr is None:
        print(header_msg)
        continue
    total_length = int(header_msg['total_length'])
    total_chunks = int(header_msg['total_chunks'])
    print("客户端{}：总长度{}，总块数{}".format(client_addr, total_length, total_chunks))
    
    # 存储已接收的块，按序号排序
    chunks = {}
    received_chunks = 0
    # 接收所有块
    while received_chunks < total_chunks:
        chunk_header, addr = recv_json(server)
        if addr != client_addr:
            continue
        chunk_idx = int(chunk_header['chunk_idx'])
        if chunk_idx in chunks:
            continue  # 跳过重复块
        # 接收对应块的数据
        chunk_data, addr = server.recvfrom(CHUNK_SIZE + 100)
        if addr == client_addr:
            chunks[chunk_idx] = chunk_data
            received_chunks += 1
            print("已接收块{}，累计{}块".format(chunk_idx, received_chunks))
    
    # 按序号拼接完整数据
    data_body = b''.join([chunks[idx] for idx in range(total_chunks)])
    # 校验总长度
    if len(data_body) != total_length:
        print("警告：实际接收总长度{}与声明长度{}不符".format(len(data_body), total_length))
    print("完整数据：{}".format(data_body.decode('utf-8')))

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