我的基于 spring 集成的 TCP 服务器运行良好。

如果服务需要很长时间，它还会处理服务器的回复超时。当服务花费的时间比设置的回复超时时间长时，它会将消息发送到错误 channel ，错误 channel 又将错误消息发送回客户端。

这是我的代码:

<int:transformer id="errorHandler"
    input-channel="errorChannel"
    ref="myTransformer" method="transform" />
<bean id="myTransformer" class="com.sample.MyTransformer" />

public class MyTransformer {
    private static Logger logger = Logger.getLogger(MyTransformer.class);

    public String transform(org.springframework.integration.handler.ReplyRequiredException e) {
        logger.error("timeout exception is thrown");
        return "Error in processing request.";
    }
}

上面的代码有效，客户端得到“处理请求时出错”，服务器日志中有条目“超时异常被抛出”。但是我在日志中也看到了以下异常:

2016-06-30 16:25:27,827 ERROR [org.springframework.integration.handler.LoggingHandler] org.springframework.integration.handler.ReplyRequiredException: No reply produced by handler 'org.springframework.integration.config.ServiceActivatorFactoryBean#0', and its 'requiresReply' property is set to true.
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:147)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:120)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:442)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:150)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessagingTemplate.sendAndReceive(AbstractMessagingTemplate.java:42)
    at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:97)
    at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:422)
    at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceiveMessage(MessagingGatewaySupport.java:390)
    at org.springframework.integration.ip.tcp.TcpInboundGateway.doOnMessage(TcpInboundGateway.java:119)
    at org.springframework.integration.ip.tcp.TcpInboundGateway.onMessage(TcpInboundGateway.java:97)
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.run(TcpNetConnection.java:182)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
    at java.lang.Thread.run(Thread.java:662)

MyTransformer 的实现似乎不正确。

请问如何定制变压器？ 如何在转换方法中获取有效负载，以便我可以回复客户端“处理请求时出错。有效载荷 = ' + 有效载荷？

谢谢

更新:为了避免记录 ReplyRequiredException，我更改了 tcp-inbound-gateway 上的错误 channel ，如下所示:

<int-ip:tcp-inbound-gateway id="gatewayCrLf"
    connection-factory="crLfServer"
    request-channel="requestChannel"
    error-channel="tcpErrorChannel"
    reply-timeout="10000"
    />
<int:channel id="tcpErrorChannel" /> 

<int:service-activator input-channel="requestChannel" ref="gateway" requires-reply="true"/>
<int:gateway id="gateway" default-request-channel="timeoutChannel" default-reply-timeout="10000" />
<int:object-to-string-transformer id="serverBytes2String"
    input-channel="timeoutChannel"
    output-channel="serviceChannel"/>

<int:channel id="timeoutChannel">
    <int:dispatcher task-executor="timeoutExecutor"/>
</int:channel>

<bean id="timeoutExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="5" />
    <property name="maxPoolSize" value="100" />
    <property name="queueCapacity" value="50" />
</bean>

但我收到 MessageDeliveryException。 tcpErrorChannel 设置似乎有误。你能建议一下吗？这是堆栈跟踪:

2016-07-05 12:17:34,266 ERROR [org.springframework.integration.ip.tcp.connection.TcpNetConnection] Exception sending message: GenericMessage [payload=byte[67], headers={timestamp=1467735444239, id=30eb099e-955d-1bd3-1789-49aa9fc84b6f, ip_tcp_remotePort=64055, ip_address=127.0.0.1, ip_localInetAddress=/127.0.0.1, ip_hostname=127.0.0.1, ip_connectionId=127.0.0.1:64055:5678:908d39a1-d027-4753-b144-59b9c0390fd7}]
org.springframework.messaging.MessagingException: failure occurred in error-handling flow; nested exception is org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'org.springframework.context.support.FileSystemXmlApplicationContext@4876db09.tcpErrorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:452)
    at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceiveMessage(MessagingGatewaySupport.java:390)
    at org.springframework.integration.ip.tcp.TcpInboundGateway.doOnMessage(TcpInboundGateway.java:119)
    at org.springframework.integration.ip.tcp.TcpInboundGateway.onMessage(TcpInboundGateway.java:97)
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.run(TcpNetConnection.java:182)
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
    at java.lang.Thread.run(Thread.java:662)
Caused by: org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'org.springframework.context.support.FileSystemXmlApplicationContext@4876db09.tcpErrorChannel'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:81)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:442)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:150)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:45)
    at org.springframework.messaging.core.AbstractMessagingTemplate.sendAndReceive(AbstractMessagingTemplate.java:42)
    at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:97)
    at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:449)
    ... 7 more
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:153)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:120)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    ... 14 more

上述异常将通过将 errorChennel 更改为 tcpErrorChannel 而消失:

非常感谢 Artem 的完美解决方案。

最佳答案

看起来是你的另一个问题的继续reply-timeout meaning in tcp-inbound-gateway in spring integration .

嗯，ReplyRequiredException是 <int:service-activator input-channel="requestChannel" ref="gateway" requires-reply="true"/> 的结果.

requires-reply="true" - 请阅读它的描述。你可以玩这个，把它做成false (默认一个)，但是你的“超时”逻辑将不起作用。所以，你别无选择，除非继续忍受:

ERROR [org.springframework.integration.handler.LoggingHandler] org.springframework.integration.handler.ReplyRequiredException

不过，这只是对默认 errorChannel 的订阅者之一的影响: http://docs.spring.io/spring-integration/reference/html/configuration.html#namespace-errorhandler .

现在关于你的最后一个问题。所有异常都包装到 ErrorMessage 中并发送到那个 errorChannel .通常异常被包装到 MessagingException 中，例如在我们的例子中，它是这样的:

else if (this.requiresReply && !isAsync()) {
    throw new ReplyRequiredException(message, "No reply produced by handler '" +
            getComponentName() + "', and its 'requiresReply' property is set to true.");
}

注意那个message ctor 参数它恰好是一个 requestMessage在我们的场景中没有回复。您可以提取所需的 payload从中。

所以，现在你可以得到 failedMessage来自 ReplyRequiredException在你的transform()并提取其 payload为了你的目标。

更新

Since this ReplyRequiredException is a known exception, there is no need to see it as Error.

好吧，即使它是一个已知的异常，但它在其他组件和场景中也是一个重要的异常，当它可能不合适时不会收到回复。 ReplyRequiredException在 Framework 组件中非常常见，因此将它移到不同的类别对于您的应用程序来说可能是一个很大的架构错误。

为了避免这些日志，你不应该使用默认的 errorChannel ，但来自您的 TCP 入站网关的其他一些。类似于 error-channel="tcpErrorChannel"就这样。默认 errorChannel仍然会有LoggingHandler作为订阅者，但您不会向该 channel 发送错误消息。

关于java - 在 Spring Integration 中定制变压器，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38133012/