Python UDP双向通信场景下如何计算每秒数据包传输速率

你的思路存在两处核心错误，无法得到准确的数据包传输速率结果：

• 第一，__sizeof__() 方法返回的是Python对象在当前解释器进程中占用的内存字节数，包含Python对象的元数据开销，和你通过UDP网络实际传输的字节长度完全无关。UDP传输只能发送二进制字节序列，你需要先把要传输的Python对象序列化（比如用struct模块打包、pickle序列化、转成JSON字符串再编码为bytes等），序列化后的字节串的len()才是单个数据包的实际传输大小。
• 第二，通过总传输数据大小除以单个包大小来统计总包数的思路误差很大，实际场景中可能出现包大小不固定、UDP丢包、业务层分包逻辑偏差等情况，直接统计实际发送/接收的数据包个数，再结合耗时计算pps才是更准确的方案。

修正方案 #

计算pps的核心逻辑非常简单：总数据包个数 / 总传输耗时 = 每秒数据包数
以下是可直接运行的测试示例：

客户端代码示例 #

import socket
import time
import struct

# UDP套接字配置
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_addr = ('127.0.0.1', 12345)

# 要发送的测试数据，先序列化为字节（这里用struct打包为例，可根据需求替换为其他序列化方式）
test_data = (1,2,2,333333,33333,33333,3333,2222,2222,22222,2222,1)
# 按格式打包为bytes：I代表无符号整型，12个元素对应12个I
packed_data = struct.pack('12I', *test_data)
# 单个数据包实际传输大小
single_packet_size = len(packed_data)
print(f"单个数据包实际传输大小：{single_packet_size} 字节")

# 测试参数：发送10000个包测试速率
total_packets = 10000
start_time = time.time()

for i in range(total_packets):
    client_socket.sendto(packed_data, server_addr)
    # 等待接收服务端回包，不需要统计双向速率可以删掉这行
    recv_data, _ = client_socket.recvfrom(1024)

end_time = time.time()
time_cost = end_time - start_time
pps = total_packets / time_cost

print(f"总耗时：{time_cost:.2f} 秒，双向传输速率：{pps:.2f} pps")
client_socket.close()

服务端代码示例 #

import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(('127.0.0.1', 12345))
print("服务端启动，等待客户端连接...")

while True:
    data, client_addr = server_socket.recvfrom(1024)
    # 收到数据后直接回传
    server_socket.sendto(data, client_addr)

补充说明 #

• 如果只需要统计单向发送的pps，不需要等待回包，直接统计发送端的发包数和发送耗时即可
• 如果需要统计实际成功送达的pps，可以在服务端统计收到的总包数，再除以从收到第一个包到最后一个包的耗时
• 测试时如果发包速率过高可能出现UDP丢包，属于正常现象，UDP本身不保证可靠传输

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