java - MqttPahoMessageHandler 和重连率管理

Question

我们注意到一些将 Spring Integration 与 MQTT 结合使用的应用程序在很短的时间间隔内出现了很多重新连接尝试，这导致我们正在使用的代理出现问题。我认为部分问题是 MqttPahoMessageHandler 似乎没有使用任何连接退避算法。对于入站连接，它看起来像 MqttPahoMessageDrivenChannelAdapter 使用常量退避(例如 recoveryInterval)，我认为 Paho 的较新版本 (>= 1.1) 有支持用于指数退避。使用 MqttPahoMessageHandler，似乎每条消息发布都会导致新的连接尝试(如果尚未连接的话)。

在我们的环境中，我们看到了以下场景:

1. 我们的代理在短时间内收到大量连接，导致 MQTT 将消息排队。
2. 遗嘱队列最终达到峰值容量。
3. MQTT 连接不会成功:
   1. TCP 连接已建立
   2. 发送带有遗嘱的 MQTT CONNECT 消息
   3. 经纪人尝试将遗嘱消息排队，但失败了
   4. 代理终止 TCP 连接

发生这种情况时，如果我们尝试发送大量出站消息，我们会看到来自 MqttPahoMessageHandler + Paho 的大量连接尝试。

出站连接 (MqttPahoMessageHandler) 不使用退避算法进行重新连接是否有原因，类似于用于入站连接 (MqttPahoMessageDrivenChannelAdapter) 或与Paho？

我想修改此行为，但首先想了解这是否是有意设计的决定？

Answer 1

所有 Spring Integration 出站端点都将尝试按需连接。

您可以添加 Circuit Breaker Advice以防止这种情况。当检测到可配置数量的故障时，断路器打开，随后的尝试将很快失败。一段时间后，断路器半开，允许进行尝试；如果失败，它会再次打开。