问题内容
我们有一些 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 connectasync(@nullable stompheaders connectheaders, stompsessionhandler handler) { connectionhandlingstompsession session = createsession(connectheaders, handler); this.tcpclient.connectasync(session); return session.getsession();}
登录后复制
通过另一种变体重新连接的情况：
CompletableFuture connectAsync(TcpConnectionHandlerconnectionHandler, ReconnectStrategy reconnectStrategy);
登录后复制
以上就是使用 Spring Integration 适配器处理底层 MQTT 和 STOMP 连接的详细内容，更多请关注【创想鸟】其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至253000106@qq.com举报，一经查实，本站将立刻删除。
发布者：PHP中文网，转转请注明出处：https://www.chuangxiangniao.com/p/2622030.html
