MQTT客户端因并发发布过多导致“Too many publishes in progress”错误

Question

在高并发场景下，MQTT客户端（如Eclipse Paho、MQTT.js等）频繁调用`publish()`方法而未等待QoS 1/2消息的`PUBACK`/`PUBREC`响应，会导致待确认的未完成发布请求（in-flight publishes）超出客户端内部队列上限（如Paho默认为10），触发“Too many publishes in progress”错误。该问题并非网络或Broker限制，而是客户端为防止内存溢出和状态混乱实施的主动流控机制。常见诱因包括：异步发布无节制、缺乏发布速率控制、未监听`onSuccess`/`onFailure`回调及时清理、或误将QoS 0当作“无状态”而忽略其底层仍占用发布槽位（部分客户端对QoS 0也计入in-flight计数）。若不加干预，将导致后续发布阻塞或静默失败，严重影响IoT设备批量上报、实时告警等关键链路的可靠性。

Answer 1

```html

一、现象层：错误表征与典型日志特征

“Too many publishes in progress” 是 Eclipse Paho（Java/Python）、MQTT.js 等主流客户端在 publish() 调用超限后抛出的明确异常（非网络超时或 Broker 拒绝）。该错误不伴随 CONNECTION_LOST 或 TIMEOUT，且复现稳定——每第 11 次（默认队列上限=10）无等待的 QoS 1 发布即触发。Paho Java 日志示例：org.eclipse.paho.client.mqttv3.MqttException: Too many publishes in progress。

二、机制层：in-flight 队列的本质与设计契约

• 定义：in-flight publish 指已发出但尚未收到对应 QoS 协议确认帧（QoS1→PUBACK；QoS2→PUBREC）的发布请求，客户端必须维持其状态以支持重传与去重。
• 内存模型：每个 in-flight 请求占用堆内存（含 payload 引用、packet ID、回调上下文），Paho 默认 maxInflight = 10，MQTT.js 默认 queueQoSZero = false 但 inflight 计数仍包含 QoS0（取决于 resend 实现）。
• 流控逻辑：客户端在 publish() 内部执行原子检查：if (inflightCount >= maxInflight) throw new MqttException(...)，属主动熔断，非被动阻塞。

三、诱因层：高频并发下的四大反模式

反模式	技术表现	隐蔽性
异步洪泛发布	循环中无 await / 无 callback 等待直接 publish(1000次)	高（日志无报错，仅部分消息丢失）
QoS 0 误判	认为 QoS0 “发完即弃”，未清理 in-flight slot（如 Paho Python 将 QoS0 计入计数）	极高（开发者普遍认知偏差）
回调泄漏	注册 onSuccess/onFailure 但未在回调内调用 client.clearInflight(packetId)（需手动触发）	中（仅调试时可见内存增长）
重连未重置	网络抖动后重建连接，但旧 in-flight 队列未清空，新连接立即满载	中（偶发，难复现）

四、诊断层：精准定位 in-flight 状态的工具链

推荐组合方案：

1. 客户端埋点：扩展 Paho 的 MqttCallbackExtended，在 deliveryComplete() 中打印 client.getPendingDeliveryTokens().length；
2. Broker 侧验证：启用 Mosquitto 的 log_type all，过滤 PUBACK 时间戳与客户端发送时间差；
3. 内存快照：JVM 应用使用 jmap -histo 查看 MqttToken 实例数是否线性增长。

五、解法层：从规避到治理的三级响应体系

graph LR
A[源头节流] -->|限速器| B(令牌桶：每秒≤8次 QoS1 publish)
B --> C[过程可控] -->|发布确认钩子| D[自动清理 in-flight]
D --> E[兜底防御] -->|连接重建钩子| F[强制清空 pending queue]

六、代码层：生产就绪的弹性发布封装示例（Paho Java）

public class ResilientMqttPublisher {
  private final MqttAsyncClient client;
  private final Semaphore inflightLimiter = new Semaphore(8); // 低于默认10，预留缓冲

  public void safePublish(String topic, byte[] payload, int qos) throws Exception {
    if (!inflightLimiter.tryAcquire(3, TimeUnit.SECONDS)) {
      throw new RuntimeException("Publish rejected: inflight saturated");
    }
    try {
      MqttMessage msg = new MqttMessage(payload);
      msg.setQos(qos);
      client.publish(topic, msg, null, new IMqttActionListener() {
        @Override
        public void onSuccess(IMqttToken token) {
          inflightLimiter.release(); // ✅ 关键：成功后释放许可
        }
        @Override
        public void onFailure(IMqttToken token, Throwable cause) {
          inflightLimiter.release(); // ✅ 失败同样释放，避免死锁
        }
      });
    } catch (Exception e) {
      inflightLimiter.release(); // ✅ 异常路径兜底
      throw e;
    }
  }
}

七、架构层：面向 IoT 规模化的发布中间件设计

当设备端需每秒上报 >50 条指标时，应脱离单客户端直连模式，引入：

• 边缘缓冲层：本地 SQLite + WAL 模式暂存未确认消息，按 Broker RTT 动态调整批量 size；
• 会话级流控：基于 MQTT 5.0 的 Server Keep Alive 与 Session Expiry Interval 协同实现服务端感知的客户端节流；
• QoS 分级策略：告警类消息强制 QoS1+重试，遥测类降级为 QoS0 并启用 cleanSession=false 保障会话连续性。

八、演进层：MQTT 5.0 对 in-flight 的根本性优化

MQTT 5.0 引入两项关键能力缓解该问题：

1. Shared Subscriptions：多个客户端订阅同一共享主题（$share/group/topic），将高并发发布压力分散至多个消费者实例；
2. Reason Code 扩展：Broker 可返回 QUOTA_EXCEEDED (147)，使客户端能区分是服务端限流还是客户端自身 in-flight 溢出，实现精准降级。

九、监控层：SLO 驱动的 in-flight 健康度指标

指标名称	采集方式	SLO 建议阈值
inflight_ratio	client.getPendingDeliveryTokens().length / maxInflight	< 0.7（持续5分钟）
puback_p99_latency_ms	Broker 日志解析 PUBACK 时间戳差	< 200ms（局域网）

十、认知层：超越“客户端 Bug”的系统性思维

该问题本质是 MQTT 协议在不可靠网络下对“状态一致性”的代价权衡：in-flight 队列是客户端为兑现 QoS 承诺所必须支付的内存与复杂度成本。解决方案不应止于调大 maxInflight，而需结合业务语义（如告警强实时 vs. 遥测最终一致）、网络拓扑（边缘-云分层）、协议演进（MQTT 5.0 Session Resumption）进行全栈协同设计。

```