使用 Spring Integration 适配器处理底层 MQTT 和 STOMP 连接

问题内容

我们有一些 spring integration 流程来处理通过 mqtt 或 stomp 到达的消息。为此，我们使用适配器 mqttpahomessagedrivenchanneladapter 和 stompinboundchanneladapter。 在 mqtt 的情况下，我们观察到，如果流中的任何端点抛出异常，适配器将关闭连接并且不再接收消息。同样，如果我们重新启动代理，则不会再次建立与其的连接。

为了处理异常问题，我们将错误通道名称设置为spring默认处理的值“errorchannel”的适配器。我们的目的是只记录异常，而不关闭底层连接。这是在整个流程中处理异常的正确方法吗？

关于重新连接问题，我们对每种传输协议都有不同的方法。

• 对于 mqtt，我们将 connectionoptions 的 automaticreconnect 设置为 true：

var clientfactory = new defaultmqttpahoclientfactory();
clientfactory.getconnectionoptions().setautomaticreconnect(true);

var adapter = new mqttpahomessagedrivenchanneladapter("tcp://localhost:1883", mqttasyncclient.generateclientid(), clientfactory, "/topic/mytopic");
adapter.seterrorchannelname("errorchannel");

• 对于 stomp，我们将上下文中的 taskscheduler 设置为 reactornettytcpstompclient：

var stompClient = new ReactorNettyTcpStompClient(host, port);
stompClient.setTaskScheduler(taskScheduler);

var stompSessionManager = new ReactorNettyTcpStompSessionManager(stompClient);

var adapter = new StompInboundChannelAdapter(stompSessionManager, "/queue/myQueue");
adapter.setErrorChannelName("errorChannel");

这是处理这个问题的最佳方法吗？

正确答案

是的，errorchannel 选项是抑制向 mqtt 客户端抛出异常的好方法。不必是全局 errorchannel ，它可能在许多不同的地方使用。 setautomaticreconnect(true) 确实推荐用于入站通道适配器。

reactornettytcpstompclient 的 taskscheduler 不适用于重新连接。请参阅其 javadocs。我认为重新连接逻辑在 reactornettytcpstompclient 中没有使用：

public completablefuture<stompsession> connectasync(@nullable stompheaders connectheaders, stompsessionhandler handler) {
    connectionhandlingstompsession session = createsession(connectheaders, handler);
    this.tcpclient.connectasync(session);
    return session.getsession();
}

通过另一种变体重新连接的情况：

CompletableFuture<Void> connectAsync(TcpConnectionHandler<P> connectionHandler, ReconnectStrategy reconnectStrategy);

以上是使用 Spring Integration 适配器处理底层 MQTT 和 STOMP 连接的详细内容。更多信息请关注PHP中文网其他相关文章！