2

我有 SI 流，org.springframework.web.reactive.socket.WebSocketMessage它使用ByteBuf. 在某些时候，我的流程中发生了异常： 

org.springframework.messaging.MessageHandlingException：消息处理程序[_org.springframework.integration.errorLogger.handler]中发生错误；嵌套异常是 io.netty.util.IllegalReferenceCountException: refCnt: 0
    在 org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:184) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:175) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.integration.dispatcher.BroadcastingDispatcher.invokeHandler(BroadcastingDispatcher.java:224) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.integration.dispatcher.BroadcastingDispatcher.dispatch(BroadcastingDispatcher.java:180) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181) ~[spring-messaging-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160) ~[spring-messaging-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47) ~[spring-messaging-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108) ~[spring-messaging-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 org.springframework.integration.channel.MessagePublishingErrorHandler.handleError(MessagePublishingErrorHandler.java:93) ~[spring-integration-core-5.0.5.RELEASE.jar!/:5.0.5.RELEASE]
...
引起：io.netty.util.IllegalReferenceCountException：refCnt：0
    在 io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1417) ~[netty-buffer-4.1.24.Final.jar!/:4.1.24.Final]
    在 io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1356) ~[netty-buffer-4.1.24.Final.jar!/:4.1.24.Final]
    在 io.netty.buffer.AbstractByteBuf.getInt(AbstractByteBuf.java:417) ~[netty-buffer-4.1.24.Final.jar!/:4.1.24.Final]
    在 io.netty.buffer.ByteBufUtil.hashCode(ByteBufUtil.java:175) ~[netty-buffer-4.1.24.Final.jar!/:4.1.24.Final]
    在 io.netty.buffer.AbstractByteBuf.hashCode(AbstractByteBuf.java:1315) ~[netty-buffer-4.1.24.Final.jar!/:4.1.24.Final]
    在 org.springframework.core.io.buffer.NettyDataBuffer.hashCode(NettyDataBuffer.java:288) ~[spring-core-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 org.springframework.web.reactive.socket.WebSocketMessage.hashCode(WebSocketMessage.java:134) ~[spring-webflux-5.0.6.RELEASE.jar!/:5.0.6.RELEASE]
    在 java.lang.Object.toString(Object.java:236) ~[?:1.8.0_161]
    在 java.lang.String.valueOf(String.java:2994) ~[?:1.8.0_161]
    在 java.lang.StringBuilder.append(StringBuilder.java:131) ~[?:1.8.0_161]

之后，处理所有二进制 Web 套接字消息将失败，并出现以下异常：

2018-11-26T10:38:29,133 错误 --- [-server-epoll-7] osihLoggingHandler (:) org.springframework.messaging.MessageDeliveryException：无法将消息发送到通道“binaryWebSocketMessageChannel”；嵌套异常是 java.lang.IllegalStateException: The [binaryWebSocketMessageChannel] 没有订阅者接受消息，failedMessage=GenericMessage [payload=MyPayload(payload=org.springframework.web.reactive.socket.WebSocketMessage@38552d5, session=ReactorNettyWebSocketSession[ id=3e0be929, uri=http://localhost:8080/]), headers={id=b09a89ff-f7be-1b43-6f62-40e5c0b5695a, timestamp=1543225109132}]
    在 org.springframework.integration.support.utils.IntegrationUtils.wrapInDeliveryExceptionIfNecessary(IntegrationUtils.java:163)
    在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:475)
    在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:394)
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)
    在 org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    在 org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
    在 org.springframework.integration.router.AbstractMessageRouter.handleMessageInternal(AbstractMessageRouter.java:183)
    在 org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:158)
    在 org.springframework.integration.handler.AbstractMessageHandler.onNext(AbstractMessageHandler.java:205)
    在 org.springframework.integration.handler.AbstractMessageHandler.onNext(AbstractMessageHandler.java:55)
    在 org.springframework.integration.endpoint.ReactiveStreamsConsumer$1.hookOnNext(ReactiveStreamsConsumer.java:138)
    在 org.springframework.integration.endpoint.ReactiveStreamsConsumer$1.hookOnNext(ReactiveStreamsConsumer.java:127)
    在 reactor.core.publisher.BaseSubscriber.onNext（BaseSubscriber.java:158）
    在 reactor.core.publisher.FluxRetry$RetrySubscriber.onNext(FluxRetry.java:79)
...
原因：java.lang.IllegalStateException：[binaryWebSocketMessageChannel] 没有订阅者接受消息
    在 org.springframework.util.Assert.state(Assert.java:94)
    在 org.springframework.integration.channel.FluxMessageChannel.doSend(FluxMessageChannel.java:63)
    在 org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:445)
    ... 57 更多

谁能指出我尝试解决问题的任何方向？同样，在哪些情况下 SI EIP 组件（路由器、转换器、过滤器、服务激活器）会取消订阅频道？

作为参考，通道类型为org.springframework.integration.channel.FluxMessageChannel

编辑：

我的流程如下所示：

WebSocketMessage -> router: (BINARY)  -> binaryWebSocketMessageChannel -> ...
                            (!BINARY) -> nullChannel

（我知道过滤器在这里更合适，我计划稍后重构）

@ArtemBilan 带有示例的仓库在这里：https ://github.com/ioreskovic/Spring-Integration-flow-loses-subscriber

4

1 回答 1

2

关键是在那个 Spring Integration 版本Publisher中FluxMessageChannel被取消了。

我们从版本onErrorContinue()中的 Reactor开始使用。要解决您的问题，最好考虑将您的应用程序升级到最新的 Spring Boot 。3.25.12.1.1

作为我们的解决方法，您可以考虑在其中吞下一个异常BinaryWsmToBytesTransformer并且不要将其冒泡到FluxMessageChannel后面。

于 2018-11-30T19:35:35.670 回答