如何通过广播传输大Pickle数据？UDP受限求替代协议

以下几种协议/工具可以满足你通过广播发送完整大Pickle数据的需求，无需让接收端处理字节片段：

1. ZeroMQ (ZMQ) 发布/订阅模式 #

ZeroMQ的PUB-SUB模型原生支持多播/广播，并且会自动处理消息的分片、重组与可靠传输（可通过配置确保消息完整性）。你无需手动拆分数据，直接发送序列化后的大Pickle字节流即可，接收端会拿到完整的消息。

发送端代码 #

import zmq
import pickle

context = zmq.Context()
socket = context.socket(zmq.PUB)
# 绑定到广播可用的TCP端口
socket.bind("tcp://*:37020")

# 序列化大型数据
data = {"large_dataset": [i for i in range(100000)]}
serial_data = pickle.dumps(data)
# 直接发送完整数据
socket.send(serial_data)

接收端代码 #

import zmq
import pickle

context = zmq.Context()
socket = context.socket(zmq.SUB)
# 订阅所有消息（空字节串表示接收该端口下的所有广播内容）
socket.setsockopt(zmq.SUBSCRIBE, b"")
# 连接到发送端的广播地址
socket.connect("tcp://<你的广播IP>:37020")

# 接收完整消息并反序列化
serial_data = socket.recv()
data = pickle.loads(serial_data)
print(len(data["large_dataset"]))

2. MQTT 发布/订阅协议 #

MQTT是轻量级消息协议，基于发布/订阅模型实现“广播”效果——所有订阅对应主题的客户端都会收到完整消息。主流MQTT Broker（如Mosquitto、EMQX）支持配置大消息上限，且通过QoS（服务质量）机制确保消息完整送达，接收端无需处理分片。

发送端代码（使用paho-mqtt库） #

import paho.mqtt.client as mqtt
import pickle

# 连接本地MQTT Broker（可替换为部署在局域网的Broker）
client = mqtt.Client()
client.connect("localhost", 1883, 60)

data = {"large_dataset": [i for i in range(100000)]}
serial_data = pickle.dumps(data)
# 发布到指定主题，QoS=1确保消息至少送达一次
client.publish("large_data_topic", serial_data, qos=1)

client.disconnect()

接收端代码 #

import paho.mqtt.client as mqtt
import pickle

def on_message(client, userdata, msg):
    # 直接接收完整Pickle数据并反序列化
    data = pickle.loads(msg.payload)
    print(len(data["large_dataset"]))

client = mqtt.Client()
client.on_message = on_message
client.connect("localhost", 1883, 60)
# 订阅目标主题
client.subscribe("large_data_topic", qos=1)

client.loop_forever()

3. WebSocket 广播（基于SocketIO） #

如果客户端支持WebSocket，可通过SocketIO实现广播。它基于TCP自动处理可靠传输，发送端直接发送大Pickle数据，所有连接的客户端都会收到完整消息，无需手动处理分片。

服务端（发送端）代码（使用flask-socketio） #

from flask import Flask
from flask_socketio import SocketIO, emit
import pickle

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on('connect')
def handle_connect():
    print("客户端已连接")

# 触发广播
def broadcast_large_data():
    data = {"large_dataset": [i for i in range(100000)]}
    serial_data = pickle.dumps(data)
    socketio.emit('large_data', serial_data, broadcast=True)

if __name__ == '__main__':
    broadcast_large_data()
    socketio.run(app, host='0.0.0.0', port=5000)

客户端代码 #

import socketio
import pickle

sio = socketio.Client()

@sio.on('large_data')
def handle_large_data(serial_data):
    data = pickle.loads(serial_data)
    print(len(data["large_dataset"]))

sio.connect('http://<你的广播IP>:5000')
sio.wait()

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