如何更好地关联同一应用程序中的 Spring Integration TCP Inbound 和 Outbound Adapters？

Question

我目前有一个 Spring Integration 应用程序，它利用许多 TCP 入站和出站适配器组合来处理消息。所有这些适配器组合都使用相同的单个 MessageEndpoint 来处理请求，并使用相同的单个 MessagingGateway 来发送响应。

MessageEndpoint 的最终输出通道是DirectChannel，它也是 MessageGateway 的 DefaultRequestChannel。这个DirectChannel 使用默认的RoundRobinLoadBalancingStrategy，它正在循环搜索正确的出站适配器以发送给定的响应。当然，这种循环搜索并不总能在第一次搜索时找到合适的出站适配器，当它找不到时，它会相应地记录。这不仅会产生大量不需要的日志记录，而且还会引发一些性能问题，因为我预计在任何给定时间都会存在数百个入站/出站适配器组合。

我想知道是否有一种方法可以更紧密地关联入站和出站适配器，无需循环处理，并且每个响应都可以直接发送到相应的出站适配器？理想情况下，我希望以可以保持使用单个MessageEndpoint 和单个MessageGateway 的方式实现这一点。

注意：请将解决方案限制为使用入站/出站适配器组合的解决方案。我的实现无法使用 TcpInbound/TcpOutboundGateways，因为我需要向单个请求发送多个响应，据我所知，这只能通过使用入站/出站适配器来完成。

为了更清楚起见，下面是所描述的当前实现的精简版本。我试图清除任何不相关的代码只是为了让事情更容易阅读......

// Inbound/Outbound Adapter creation (part of a service that is used to dynamically create varying number of inbound/outbound adapter combinations)
public void configureAdapterCombination(int port) {

    TcpNioServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(port);
    // Connection Factory registered with Application Context bean factory (removed for readability)...

    TcpReceivingChannelAdapter inboundAdapter = new TcpReceivingChannelAdapter();
    inboundAdapter.setConnectionFactory(connectionFactory);
    inboundAdapter.setOutputChannel(context.getBean("sendFirstResponse", DirectChannel.class));
    // Inbound Adapter registered with Application Context bean factory (removed for readability)...

    TcpSendingMessageHandler outboundAdapter = new TcpSendingMessageHandler();
    outboundAdapter.setConnectionFactory(connectionFactory);
    // Outbound Adapter registered with Application Context bean factory (removed for readability)...

    context.getBean("outboundResponse", DirectChannel.class).subscribe(outboundAdapter);

}

// Message Endpoint for processing requests
@MessageEndpoint
public class RequestProcessor {

    @Autowired
    private OutboundResponseGateway outboundResponseGateway;

    // Direct Channel which is using Round Robin lookup
    @Bean
    public DirectChannel outboundResponse() {
        return new DirectChannel();
    }

    // Removed additional, unrelated, endpoints for readability...

    @ServiceActivator(inputChannel="sendFirstResponse", outputChannel="sendSecondResponse")
    public Message<String> sendFirstResponse(Message<String> message) {
        // Unrelated message processing/response generation excluded...
        outboundResponseGateway.sendOutboundResponse("First Response", message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class));
        return message;
    }

    // Service Activator that puts second response on the request channel of the Message Gateway
    @ServiceActivator(inputChannel = "sendSecondResponse", outputChannel="outboundResponse")
    public Message<String> processQuery(Message<String> message) {
        // Unrelated message processing/response generation excluded...
        return MessageBuilder.withPayload("Second Response").copyHeaders(message.getHeaders()).build();
    }

}

// Messaging Gateway for sending responses
@MessagingGateway(defaultRequestChannel="outboundResponse")
public interface OutboundResponseGateway {
    public void sendOutboundResponse(@Payload String payload, @Header(IpHeaders.CONNECTION_ID) String connectionId);
}

解决方案：

@Artem 在下面的 cmets/answers 中的建议似乎可以解决问题。只是想快速记录一下我是如何在创建时向每个出站适配器添加replyChannel 的。

我所做的是创建两个由应用程序维护的地图。每当创建新的入站/出站适配器组合时都会填充第一个映射，它是ConnectionFactory 名称到replyChannel 名称的映射。第二个映射是ConnectionId 到replyChannel 名称的映射，它通过EventListener 填充到任何新的TcpConnectionOpenEvent。

请注意，每个TcpConnectionOpenEvent 都将根据建立连接的位置/方式定义ConnectionFactoryName 和ConnectionId 属性。

从那里，每当收到新请求时，我都会使用这些映射和Message 上的“ip_connectionId”标头将replyChannel 标头添加到消息中。第一个响应是通过从应用程序的上下文中手动获取相应的replyChannel（基于replyChannel 标头的值）并在该通道上发送响应来发送的。第二个响应是通过 Spring Integration 使用消息上的 replyChannel 标头发送的，正如 Artem 在他的响应中所描述的那样。

此解决方案是作为概念的快速证明实施的，并且仅适用于我当前的实施。包括这一点，希望能够快速启动其他查看者自己的实现/解决方案。

Answer 1

好吧，我现在明白你关于round-robin 的观点了。您针对相同的通道创建了许多类似的 TCP 通道适配器。在这种情况下，确实很难区分一个流和另一个流，因为您对这些频道及其订阅者有一点控制。

解决方案之一是使用 Spring Integration Java DSL 及其动态流程：https://docs.spring.io/spring-integration/reference/html/dsl.html#java-dsl-runtime-flows

因此，您只需关注流程，无需担心运行时注册。但是由于您不在那里，并且您只处理简单的 Java 和注释配置，因此您实现目标要困难得多。不过还是……

您可能知道有类似replyChannel 的标头。当我们没有配置 outputChannel 时，它会被考虑在内。这样，您就可以为每个流创建一个独立的通道，并且所有流的配置都完全相同。

所以，

• 我会为每个configureAdapterCombination() 呼叫创建一个新频道。
• 将此方法传播到replyChannel.subscribe(outboundAdapter); 的那个方法中
• 在特定流的开头使用此通道将其填充到 replyChannel 标头中。

这样你的processQuery() service-activator 应该没有outputChannel。将从replyChannel 标头中选择它以实现正确的出站通道适配器相关性。

对于这种情况，您不需要@MessagingGateway，因为我们不再有固定的defaultRequestChannel。在sendFirstResponse() 服务方法中，您只需获取replyChannel 标头并手动发送新创建的消息。从技术上讲，它与您尝试对提到的@MessagingGateway 执行的操作完全相同。

对于 Java DSL 变体，我会在 PublishSubscribeChannel 上使用 filter 来丢弃那些不属于当前流的消息。无论如何，这是一个不同的故事。

当您配置特定的configureAdapterCombination() 时，尝试弄清楚如何为每个流创建一个回复通道。