TCP Spring 集成 - 调度程序没有“响应”通道的订阅者答案

【问题标题】：TCP Spring Integration - Dispatcher has no subscribers for 'response' channelTCP Spring 集成 - 调度程序没有“响应”通道的订阅者
【发布时间】：2013-09-25 18:47:05
【问题描述】：

我正在使用 spring 集成通过提供一些消息并获得响应来对服务器进行 TCP 调用。我更喜欢使用通道适配器来发送和接收批量消息。我面临的问题是响应渠道。为响应频道获取“调度程序没有频道订阅者”。一切正常，除了响应没有在响应通道上传输。我可以看到服务器上的握手以及日志中的响应被放在响应和记录器通道上。但是在抛出异常之后。配置设置为：

<gateway id="clientPositionsGateway" service-interface="MyGatewayInterface">
        <method name="fetchClientPositions" request-channel="clientPositionsRequestChannel" />  
    </gateway>


    <channel id="clientPositionsRequestChannel" />

    <splitter input-channel="clientPositionsRequestChannel"
            output-channel="singleClientPositionsRequestChannel" />

    <channel id = "singleClientPositionsRequestChannel" />

    <transformer
        input-channel="singleClientPositionsRequestChannel"
        output-channel="dmQueryRequestChannel"
        ref="dmPosBaseQueryTransformer" />

    <channel id = "dmQueryRequestChannel">
        <!-- <dispatcher task-executor="executor"/> -->
    </channel>

    <ip:tcp-connection-factory id="csClient"
           type="client"
           host="somehost"
           port="12345"
           single-use="true"
           deserializer="connectionSerializeDeserialize"
            />

    <ip:tcp-outbound-channel-adapter id="dmServerOutboundAdapter"
            channel="dmQueryRequestChannel"
            connection-factory="csClient"
            order="2"
            />

    <ip:tcp-inbound-channel-adapter id="dmServerInboundAdapter"
            channel="dmQueryResponseChannel"
            connection-factory="csClient"
            error-channel="errorChannel"/>

<channel id="dmQueryResponseChannel"/>

【问题讨论】：

请显示 subscriber 到“dmQueryRequestChannel”的配置。试试这个<logging-channel-adapter channel="dmQueryRequestChannel">
17:23:30.347 [pool-1-thread-1] 调试 osititcTcpNetConnection - 收到消息 [Payload=CLIENT_ID|MANAGER|HOLD_TYPE AAAIH||CURR ][Headers={timestamp=1379678010347, id =c51a97ab-0fc6-4270-bdda-46a91e0d102e, ip_tcp_remotePort=22010, ip_address=hostaddress, ip_hostname=hostname, ip_connectionId=hostname:22010:93360e5b-82f8-4e06-906a-6a5989d30d583}4]77:17:30。 thread-1] DEBUG osichannel.ExecutorChannel - preSend on channel 'dmQueryResponseChannel'，消息：[Payload=CLIENT_ID|MANAGER|HOLD_TYPE AAAIH||CURR
][Headers={timestamp=1379678010347, id=c51a97ab-0fc6-4270-bdda-46a91e0d102e, ip_tcp_remotePort=22010, ip_address=hostaddress, ip_hostname=hostname, ip_connectionId=hostname:22010e:22010e: -4e06-906a-6a5989d0d583}] 17:23:30.347 [executor-1] 调试 osbfsDefaultListableBeanFactory - 返回单例 bean 'errorChannel' 的缓存实例
17:23:30.347 [executor-1] DEBUG osicPublishSubscribeChannel - preSend on channel 'errorChannel'，消息：[Payload=org.springframework.integration.MessageDispatchingException: Dispatcher 没有订阅者][Headers= {timestamp=1379678010347, id=5cdcae44-e874-4ad2-af47-4f9b91a8cdde}] 17:23:30.347 [executor-1] DEBUG osihandler.LoggingHandler - (inner bean)#2 收到消息：[Payload=org.springframework。 integration.MessageDispatchingException: Dispatcher 没有订阅者][Headers={timestamp=1379678010347, id=5cdcae44-e874-4ad2-af47-4f9b91a8cdde}]
你会回答我的问题吗？我了解什么 StackTrace。我的意思是可能没有“dmQueryResponseChannel”的订阅者或者它没有启动

标签： java spring spring-integration

【解决方案1】：

正如 Artem 在他的评论中所说，“调度程序没有订阅者”在这里意味着没有配置端点来接收 dmQueryResponseChannel 上的响应，或者配置了该通道作为其输入通道的端点没有启动。

在任何情况下，即使您解决了这个问题，对请求/响应方案使用独立的适配器也是很棘手的，因为框架无法自动将响应与请求相关联。这就是出站网关的用途。您可以使用协作适配器，但您必须自己处理相关性。如果您使用请求/回复网关来启动流程，则必须使用一种技术，例如在 tcp-client-server-multiplex sample 中探讨的技术。这是因为使用独立适配器意味着您将丢失用于将响应返回到网关的 replyChannel 标头。

或者，您可以使用 void 返回网关来发送请求，并使用 <int:outbound-channel-adapter/> 以便框架将通过响应进行回调，您可以通过编程方式进行自己的关联。

【讨论】：

即使我使用 void 返回网关，消息也不会流向服务器。此外，使用出站网关会引发相同的异常。对如何将响应与请求通道关联起来感到很困惑！！！我没有使用协作适配器模型吗？
您需要显示这些组件的上游和下游的所有配置，例如我们需要查看您的回复通道连接到哪个端点。
使用该配置，您有 2 个订阅者到 dmQueryResponseChannel；变压器和入站网关。在变压器之后，您有一个简单的日志适配器。回复将是循环分发的消息，记录器永远不会到达网关。但是，正如我所说，此配置永远不会起作用，因为您无法将任意异步回复（例如由入站适配器接收）发送到网关回复通道 - 您必须恢复 replyChannel 标头。请使用出站网关重试并显示新配置并显示完整日志。

【解决方案2】：

如果您的 clientPositionsGateway 是从客户端线程调用的，则没有理由使用执行器通道。如果您进行百万循环 clientPositionsGateway 尝试使用未来网关：http://docs.spring.io/spring-integration/docs/2.2.5.RELEASE/reference/html/messaging-endpoints-chapter.html#async-gateway 并再次：没有执行程序通道。而且我看不出有理由在两个网关上使用reply-channel。还有一个：你在<tcp:outbound-gateway>之前有<splitter>，但是在<tcp:outbound-gateway>之后的<aggregator>在哪里？在您当前的情况下，您会收到来自 clientPositionsGateway 的回复，所有其他人都将被删除，因为 TemporaryReplyChannel 已经关闭。

【讨论】：

使用异步网关时，我收到异常调度程序在请求通道上没有订阅者，即使请求通道配置了网关。不使用聚合器的原因是，我得到了一组客户端，我将它们拆分并传递给转换器，而转换器又使用每个客户端来构建要发送到服务器的查询/消息。每条消息都有单独的回复。所以基本上我想要基于单个客户端上的查询的响应。
另外，没有从线程调用网关。我加载集合，拆分每个客户端，进行查询并传递给服务器。所以我希望每个查询都以并行方式运行。那肯定取决于线程数
需要您对上述情况和异常获取的意见
1.你真的需要同步请求/回复你的场景吗？ 2. 要并行运行，只需拆分您的集合并将结果放入执行程序通道即可。 3. 如果您需要所有项目的复杂结果，您需要在<tcp:outbound-gateway> 之后放置一个。 4. 要处理异常，您需要一个错误通道，例如在邮件标题中通过<header-enricher>为每个项目
是的，我确实需要并行处理，这就是为什么使用执行器通道。

【解决方案3】：

public interface ClientPositionsGateway {

    String fetchClientPositions(List<String> clientList); 

}

【讨论】：

感谢您的回答，您能解释一下为什么这可能有效吗？

【解决方案4】：

这是解决我问题的代码：

    @ContextConfiguration(locations={"/clientGIM2Position.xml"})
    @RunWith(SpringJUnit4ClassRunner.class)

    public class GetClientPositionsTest {

    @Autowired

    ClientPositionsGateway clientPositionsGateway;

    @Test

    public void testGetPositions() throws Exception {

    String positions = clientPositionsGateway.fetchClientPositions(clientList);

    System.out.println("returned !!!!" + positions);

           }
    }

【讨论】：

【解决方案5】：

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans" 
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns="http://www.springframework.org/schema/integration" 
    xmlns:ip="http://www.springframework.org/schema/integration/ip"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:task="http://www.springframework.org/schema/task"
    xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
            http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
            http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
            http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">

    <!-- intercept and log every message -->

    <logging-channel-adapter id="logger" level="DEBUG" />
    <wire-tap channel = "logger" />

    <gateway id="clientPositionsGateway"
         service-interface="com.example.ClientPositionsGateway">
        <method name="fetchClientPositions" request-channel="clientPositionsRequestChannel" reply-channel="dmQueryResponseChannel"/>
    </gateway>

    <channel id="clientPositionsRequestChannel" />

    <splitter input-channel="clientPositionsRequestChannel"
            output-channel="singleClientPositionsRequestChannel" />

    <channel id = "singleClientPositionsRequestChannel" />

    <transformer
        input-channel="singleClientPositionsRequestChannel"
        output-channel="dmQueryRequestChannel"
        ref="dmPosBaseTransQueryTransformer" />

    <channel id = "dmQueryRequestChannel">
        <dispatcher task-executor="executor"/>
    </channel>

    <ip:tcp-connection-factory id="csClient" 
           type="client" 
           host="hostserver"
           port="22010"
           single-use="true"
           deserializer="connectionSerializeDeserialize"
           />

    <ip:tcp-outbound-gateway id="dmServerGateway" 
           request-channel="dmQueryRequestChannel" 
           reply-channel="dmQueryResponseChannel"
           connection-factory="csClient"  />

    <channel id="dmQueryResponseChannel">
        <dispatcher task-executor="executor"/>
    </channel>

    <channel id="serverBytes2StringChannel" />

    <bean id="connectionSerializeDeserialize" class="com.example.DMQueryResponseSerializer"/>
    <bean id="dmPosBaseTransQueryTransformer" class="com.example.DMPOSBaseTransQueryTransformer"/> 
    <task:executor id="executor" pool-size="5"/>

</beans:beans>

【讨论】：

这已经成功运行了 :) 非常感谢你们 Gary 和 Artem :)
只是想知道这个配置可以实现百万条消息的并行处理？
很高兴您了解网关概念 ;-)！关于并行性：您真的确定您的环境可能支持百万线程吗？ ;-) 或者换句话说：你的机器有百万个 CPU 内核吗？
对不起，我的意思是数百万个线程。只是想问一下这种设计是否适合在百万请求消息的情况下使用，每个请求都获取大量数据。

【解决方案6】：

配置设置：

<gateway id="clientPositionsGateway" service-interface="com.example.ClientPositionsGateway">
    <method name="fetchClientPositions" request-channel="clientPositionsRequestChannel"  reply-channel="dmQueryResponseChannel"/>  
</gateway>

<channel id="clientPositionsRequestChannel" />

<splitter input-channel="clientPositionsRequestChannel"
        output-channel="singleClientPositionsRequestChannel" />

<channel id = "singleClientPositionsRequestChannel" />

<transformer
    input-channel="singleClientPositionsRequestChannel"
    output-channel="dmQueryRequestChannel"
    ref="dmPosBaseQueryTransformer" />

<logging-channel-adapter channel="clientBytes2StringChannel"/>  

<channel id = "dmQueryRequestChannel">
     <dispatcher task-executor="executor"/> 
</channel>

<ip:tcp-connection-factory id="csClient" 
       type="client" 
       host="serverHost"
       port="22010"
       single-use="true"
       deserializer="connectionSerializeDeserialize"
       />

<ip:tcp-outbound-channel-adapter id="dmServerOutboundAdapter"
        channel="dmQueryRequestChannel" 
        connection-factory="csClient"
        /> 

<ip:tcp-inbound-channel-adapter id="dmServerInboundAdapter"
        channel="dmQueryResponseChannel" 
        connection-factory="csClient" 
        error-channel="errorChannel"/>

<transformer input-channel="dmQueryResponseChannel" output-channel="clientBytes2StringChannel" ref="dmPOSBaseQueryResponseTransformer"
        />

<channel id="dmQueryResponseChannel"/>

<channel id="clientBytes2StringChannel"/>

【讨论】：