优化 CZTop::Radio 的抗丢包能力之前，先要接受一个事实：Radio/Dish 套接字运行在不可靠的多播或 UDP 传输之上，协议层不会为每条消息返回 ACK，也不会自动重传。所谓抗丢包，本质上是把发送端、内核网络栈、接收端读取循环这三段链路中的无谓丢失尽量压到最低。
丢包通常集中在三个位置：发送进程向 Socket 队列写入时因缓冲满被丢弃、内核网络栈因发送速率超过网卡可承受上限而丢弃、接收端 Dish 队列已满但应用没有及时调用 receive 导致新包无处存放。理解这些环节后，调优才会有的放矢。
先定位丢包：Radio/Dish 的丢失发生在哪里
Radio/Dish 模式并不是一个可靠传输方案。它更像 UDP 多播的轻量封装，发送端把消息推给内核后就认为任务完成，接收端是否收到、顺序是否乱序，都不会反馈给发送方。因此，任何对 Radio 的调优都应该从丢包观测开始，而不是只调参数。
最有效的定位方式是在每条消息里写入递增序号。接收端如果发现序号跳跃，就知道中间丢了包；如果消息长时间不连续，还能判断丢包集中在哪个时间窗。下面是一个简单的丢包统计示例，它把序号放在二进制负载的最前面，并在每收到一千条时输出丢包率。
require "cztop"
dish = CZTop::Dish.new("udp://224.0.0.1:5555")
dish.join("telemetry")
expected = nil
received = 0
lost = 0
loop do
msg = dish.receive(timeout: 1000)
next unless msg
seq = msg.to_s.unpack1("Q>")
if expected and seq > expected
lost += seq - expected
end
expected = seq + 1
received += 1
if received % 1000 == 0
puts "received=#{received} lost=#{lost} loss_rate=#{lost.to_f / (received + lost)}"
end
end
这个统计本身不减少丢包，但它能让调优结果可度量。运行一段时间后，如果丢包率仍高于预期，就继续从发送速率、内核缓冲和接收循环三个方向收紧。
发送端调优：速率、缓冲与消息尺寸要一起改
发送端最容易出现的问题是瞬时速率过高。即使平均带宽不高，突发写入也可能瞬间填满内核队列，导致非阻塞发送失败或消息被静默丢弃。在多播场景里，ZMQ_RATE 用来限制发送端每秒可发送的数据量，单位是千比特。CZTop 中可以通过 options 代理写入，例如把速率限制在 80Mbps 左右，可以给局域网链路留出余量。
除了速率，ZMQ_SNDBUF 也值得调大。默认的内核发送缓冲区往往只有几十 KB，应用一旦出现微小抖动就会溢出。对于多播发送，建议把 sndbuf 设为 4MB 到 8MB 之间。注意这个值要写在 bind 或 connect 之前才更容易生效，不同操作系统对上限也有不同限制。
消息尺寸同样关键。多播报文超过 MTU 后被分片，任何一个分片丢失都会导致整个消息丢弃。因此，如果业务允许，把单条消息控制在 1400 字节以内是最稳妥的做法。下面的发送示例同时设置了速率、发送缓冲、恢复间隔和回环开关。
radio = CZTop::Radio.new("udp://224.0.0.1:5555")
radio.options.rate = 80_000
radio.options.sndbuf = 8 * 1024 * 1024
radio.options.recovery_ivl = 10
radio.options.mcast_loop = false
seq = 0
group = "telemetry"
loop do
payload = [seq, Time.now.to_f, 42.5].pack("Q> d d")
radio.send(payload, group: group)
seq += 1
sleep 0.005
end
这里 rate 的单位是 kbps，80_000 表示 80Mbps。recovery_ivl 控制多播恢复相关的定时器，虽然不能替代可靠传输，但设置一个合适的值有助于接收端在短暂丢包后恢复。mcast_loop 置为 false 可以避免发送主机自己收到回环报文，减少无意义的排队。
接收端调优：让 Dish 的读取循环尽量轻量
接收端丢包最常见的原因是应用读取速度跟不上到达速度。Dish 套接字在内核和设备层都有队列，如果应用没有及时调用 receive，这些队列会被新报文填满，之后到达的包只能丢弃。因此，抗丢包不能只靠发送端限速，接收端必须保持稳定的消费节奏。
第一步是把 ZMQ_RCVBUF 调大。对于多播接收，16MB 甚至更大的缓冲能吸收短时抖动。其次，把消息反序列化、落盘、写数据库等耗时操作移出读取线程。读取循环只做二进制解析和入队，其他处理交给独立工作线程，否则单个慢操作就会阻塞整个 Dish。
dish = CZTop::Dish.new("udp://224.0.0.1:5555")
dish.join("telemetry")
dish.options.rcvbuf = 16 * 1024 * 1024
loop do
msg = dish.receive(timeout: 500)
next unless msg
seq, timestamp, value = msg.to_s.unpack("Q> d d")
handle_message(seq, timestamp, value)
end
如果业务处理确实很重，可以使用内存队列充当缓冲，读取线程只把原始消息放入队列，多个工作线程并发消费。但要注意，内存队列也不能无限增长，否则只是把丢包点从网络栈移到了应用堆。更稳妥的做法是设置容量上限，超限时丢弃最旧消息并记录告警，这样至少能保护系统不会因内存耗尽崩溃。
用序号统计验证效果并持续监控
调整完参数后，需要回到丢包统计。每轮调优只改变一个变量，比如先固定接收端不变，逐步降低 rate，观察丢包率变化；再固定发送速率，调整 sndbuf 和 rcvbuf。如果一次改动多个参数，很难判断哪个因素真正起作用。
下面是一个推荐参数起点表，适合典型的千兆局域网多播环境。实际值仍要根据网卡、交换机、主机负载和消息频率调整。
| 参数 | 默认倾向 | 推荐起点 | 说明 |
|---|---|---|---|
| ZMQ_RATE | 偏高或不受限 | 80_000 kbps | 留出 20% 左右带宽余量 |
| ZMQ_SNDBUF | 几十 KB | 8MB | 发送端短时抖动缓冲 |
| ZMQ_RCVBUF | 几百 KB | 16MB | 接收端排队能力 |
| 单消息大小 | 可能超过 MTU | 小于 1400 字节 | 避免分片后整包丢弃 |
| 读取循环 | 同步阻塞串行处理 | 只收不发配工作线程 | 避免读循环被业务阻塞 |
最后，抗丢包是一个持续观察的过程。可以在统计程序里接入现有监控系统，按分钟输出收到数、丢失数和丢包率。只有把丢包数据纳入日常监控，才能在网络或主机负载变化时快速发现异常，继续调整 Radio 的速率和缓冲参数。
CZTop Radio抗丢包ZeroMQ套接字修改时间：2026-09-06 06:16:06