Spring Integration - 跨多个 TCP 出站网关的负载平衡答案

【问题标题】：Spring Integration - Load Balance across multiple TCP outbound gatewaysSpring Integration - 跨多个 TCP 出站网关的负载平衡
【发布时间】：2016-05-22 23:23:02
【问题描述】：

我有一个具有多个进程的（旧版）TCP 服务。每个进程在同一主机上运行，但在不同的端口上。该服务是单线程的，因此提高吞吐量的方法是在每个端口上循环每个请求。

我正在为这个遗留应用程序提供 AMQP 暴露。它非常简单 - 从 AMQP 队列中取出一个字符串，将其传递给应用程序，然后将响应字符串返回给 AMQP 回复队列。

这在单个端口上效果很好。但是，我想将请求分散到所有端口。

Spring Integration 似乎只提供AbstractClientConnectionFactory 实现，这些实现要么直接连接到单个主机/端口 (TcpNetClientConnectionFactory)，要么维护与单个主机/端口 (CachingClientConnectionFactory) 的连接池。在单个主机和多个端口之间没有任何池连接。

我尝试编写自己的AbstractClientConnectionFactory，维护AbstractClientConnectionFactory 对象池和它们之间的循环。但是，当目标服务消失或网络中断时，我遇到了几个与处理 TCP 连接有关的问题，而我无法解决。

这个问题也采用了方法：Spring Integration 4 - configuring a LoadBalancingStrategy in Java DSL，但解决方案是硬编码端点的数量。就我而言，端点的数量仅在运行时才知道，并且是用户可配置的设置。

所以，基本上我需要在运行时为每个端口动态创建一个TcpOutboundGateway，并以某种方式将其注册到我的IntegrationFlow 中。我尝试了以下方法：

@Bean
public IntegrationFlow xmlQueryWorkerIntegrationFlow() {
    SimpleMessageListenerContainer inboundQueue = getMessageListenerContainer();

    DirectChannel rabbitReplyChannel = MessageChannels.direct().get();

    IntegrationFlowBuilder builder = IntegrationFlows
            .from(Amqp.inboundGateway(inboundQueue)
                      .replyChannel(rabbitReplyChannel))    
            /* SOMEHOW DO THE ROUND ROBIN HERE */
            //I have tried:
            .channel(handlerChannel()) //doesnt work, the gateways dont get started and the message doesnt get sent to the gateway
            //and I have also tried:
            .handle(gateway1)
            .handle(gateway2) //doesnt work, it chains the handlers instead of round-robining between them
            //
            .transform(new ObjectToStringTransformer())
            .channel(rabbitReplyChannel); 

    return builder.get();

}

@Bean
//my attempt at dynamically adding handlers to the same channel and load balancing between them
public DirectChannel handlerChannel() {
    DirectChannel channel = MessageChannels.direct().loadBalancer(new RoundRobinLoadBalancingStrategy()).get();
    for (AbstractClientConnectionFactory factory : generateConnections()) {
        channel.subscribe(generateTcpOutboundGateway(factory));
    }
    return channel;
}

有谁知道我该如何解决这个问题？

【问题讨论】：

标签： spring spring-integration spring-java-config spring-dsl

【解决方案1】：

请参阅dynamic ftp sample - 本质上，每个出站网关都进入其自己的应用程序上下文，动态路由器路由到适当的通道（如有必要，将根据需要为其创建出站适配器）。

虽然示例使用 XML，但您可以使用 java 配置，甚至使用 Java DSL 来做同样的事情。

查看我对similar question for multiple IMAP mail adapters using Java configuration 的回复，然后查看follow-up question。

【讨论】：