Spring集成连接到服务器端口的多个客户端

Question

从我的应用程序中，我需要配置多个需要连接到单个服务器的客户端连接。为此，我使用Application Context BeanFactory创建了可变数量的bean，具体取决于我配置了多少客户端。这是2个客户端的代码：

//setup beans;
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
ctx.scan("pkg");
ConnectionFactory factory = new ConnectionFactory();
int clients = 2; //TODO read this value from file
ConfigurableListableBeanFactory beanFactory = ctx.getBeanFactory();
for (int count = 1; count <= clients; count++) {
    TcpNetClientConnectionFactory connectionFactory = factory.createClientConnectionFactory("127.0.0.1", 6680);

    //connection factory
    beanFactory.registerSingleton("connectionFactory_" + String.valueOf(count), connectionFactory);

    //inbound gateway
    MessageChannel input = new DirectChannel();
    MessageChannel output = new DirectChannel();
    TcpInboundGateway gateway = factory.createInboundGateway(connectionFactory, beanFactory, input, output, 10000, 20000);
    beanFactory.registerSingleton("gateway_" + String.valueOf(count), gateway);

    //message transformation and handling
    IntegrationFlow flow = factory.createFlow(input);
    beanFactory.registerSingleton("flow_" + String.valueOf(count), flow);
}
ctx.refresh();

//open connections
for(int count = 1; count <= clients; count++) {
    TcpInboundGateway gateway = ctx.getBean("gateway_" + count, TcpInboundGateway.class);
    //necessary for the client to connect
    gateway.retryConnection();
}

这是我的工厂方法：

@EnableIntegration
@IntegrationComponentScan
@Configuration
public class ConnectionFactory {      
    public TcpNetClientConnectionFactory createClientConnectionFactory(String ip, int port) {
        TcpNetClientConnectionFactory factory = new TcpNetClientConnectionFactory(ip, port);
        factory.setSingleUse(false);
        factory.setSoTimeout(10000);
        factory.setSerializer(new ByteArrayLfSerializer());
        factory.setDeserializer(new ByteArrayLfSerializer());

        return factory;
    }

    public TcpInboundGateway createInboundGateway(
        AbstractConnectionFactory factory,
        BeanFactory beanFactory,
        MessageChannel input,
        int replyTimeout,
        int retryInterval) {
        TcpInboundGateway gateway = new TcpInboundGateway();
        gateway.setRequestChannel(input);
        gateway.setConnectionFactory(factory);
        gateway.setClientMode(true);
        gateway.setReplyTimeout(replyTimeout);
        gateway.setRetryInterval(retryInterval);
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.initialize();
        gateway.setTaskScheduler(scheduler);
        gateway.setBeanFactory(beanFactory);

        return gateway;
    }

    public IntegrationFlow createFlow(MessageChannel input) {
        IntegrationFlowBuilder builder = IntegrationFlows.from(input);
        builder.transform(Transformers.objectToString()).handle(System.out::println);

        return builder.get();
    }
}

当我运行程序时，两个客户端都连接到我的服务器。然而，一旦服务器将其第一个有效负载发送到每个客户端，我就会得到以下异常（每个客户端一个）：

Exception sending message: GenericMessage [payload=byte[5], headers={ip_tcp_remotePort=6680, ip_connectionId=localhost:6680:33372:e26b9973-a32e-4c28-b808-1f2556576d01, ip_localInetAddress=/127.0.0.1, ip_address=127.0.0.1, id=4443ca34-fb53-a753-7603-53f6d7d82e11, ip_hostname=localhost, timestamp=1464098102462}]
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'unknown.channel.name'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:81) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:442) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:150) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.GenericMessagingTemplate.doSendAndReceive(GenericMessagingTemplate.java:45) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.messaging.core.AbstractMessagingTemplate.sendAndReceive(AbstractMessagingTemplate.java:42) ~[spring-messaging-4.2.4.RELEASE.jar:4.2.4.RELEASE]
    at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:97) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:422) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceiveMessage(MessagingGatewaySupport.java:390) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.ip.tcp.TcpInboundGateway.doOnMessage(TcpInboundGateway.java:119) ~[spring-integration-ip-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.ip.tcp.TcpInboundGateway.onMessage(TcpInboundGateway.java:97) ~[spring-integration-ip-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.ip.tcp.connection.TcpNetConnection.run(TcpNetConnection.java:182) ~[spring-integration-ip-4.2.5.RELEASE.jar:na]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_31]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_31]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_31]
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:153) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:120) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77) ~[spring-integration-core-4.2.5.RELEASE.jar:na]
    ... 14 common frames omitted

这个想法是数据将被读取，通过我为In边界网关配置的通道发送到变压器，然后变压器将数据转换为字符串，然后将其打印出来。

为什么框架不知道把数据放在哪个通道？据我所知，我确实在入站网关工厂方法中为每个客户端创建了一个唯一的通道。有人能看看我的配置，让我知道我错过了什么，因为我绝对被这个难倒了。

Answer 1

以下是简化的工作解决方案：

豆子。JAVA

package beanconfig;

import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;

@Configuration
@EnableIntegration
public class Beans {
    //Beans can be configured here
}

集成测试。JAVA

import org.junit.Test;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer;
import org.springframework.integration.transformer.ObjectToStringTransformer;
import org.springframework.messaging.MessageChannel;

public class IntegrationTest {
    private String generateComponentName(String baseName, int instanceCount) {
        return baseName + "_" + instanceCount;
    }

    @Test
    public void integrationTest1() throws Exception {
        try(AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext()) {
            ctx.scan("beanconfig");
            ctx.refresh();

            ConfigurableListableBeanFactory beanFactory = ctx.getBeanFactory();

            int numberOfClients = 2; //TODO configure from file

            for (int count = 0; count < numberOfClients; count++) {
                //connection factory
                TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory("127.0.0.1", 6680);
                connectionFactory.setSingleUse(false);
                connectionFactory.setSoTimeout(10000);
                connectionFactory.setSerializer(new ByteArrayLfSerializer());
                connectionFactory.setDeserializer(new ByteArrayLfSerializer());

                //inbound gateway
                TcpInboundGateway inboundGateway = new TcpInboundGateway();
                inboundGateway.setRequestChannel(new DirectChannel());
                inboundGateway.setConnectionFactory(connectionFactory);
                inboundGateway.setClientMode(true);
                inboundGateway.setReplyTimeout(10000);
                inboundGateway.setRetryInterval(20000);

                //message transformation and flow
                String flowName = generateComponentName("flow", count);
                IntegrationFlow flow = IntegrationFlows.from(inboundGateway)
                    .transform(new ObjectToStringTransformer())
                    .handle(h -> System.out.println("Message received: " + h.getPayload()))
                    .get();
                beanFactory.registerSingleton(flowName, flow);
                beanFactory.initializeBean(flow, flowName);
            }

            ctx.start();

            //TODO do proper validation here
            Thread.sleep(10000);
        }
    }
}

基本上，我最初的尝试有几个地方出了问题。以下是我为使其工作而做的更改：

1） 创建AnnotationConfigApplicationContext时，必须将配置类作为参数创建，该参数标记为@EnableIntegration注释。如果没有，则必须通过包含此注释的上下文扫描组件。我在第一次尝试时确实这样做了，但调用refresh太晚了，应该直接在ctx之后调用它。扫描因为我的ctx。refresh（）是在我的beanfactory注册之后创建集成bean时，实际上没有设置@EnableIntegration。移动ctx。刷新（）位于ctx正下方。scan（）解决了这个问题。

2）注册到上下文中的每个bean也必须由beanFactory初始化。这是为了确保BeanPostProcers运行（这不是通过ynsterSingleton自动完成的）。

3）然后需要调用ctx.start（）来启用在ctx.refresh（）之后创建的bean。

Answer 2

没有人会使用来自网关的消息。setReplyChannel（输出） 。

至少我们没有看到这样的事情：

之后将打印出来。

在大多数情况下，如果您的某些SubscribableChannel没有任何订阅者，我们有Dispatcher没有订阅者：未配置或停止。

编辑

忘记我之前的表情。它适用于出站情况。

您的TcpInboundGateway很好。尽管您不需要设置ReplyChannel（），因为您始终可以依赖默认内置的临时ReplyChannel来等待来自下游流的一些结果。

你的IntegrationFlow看起来也不错。这是正确的<代码>。transform（）不会向任何其他频道发送任何内容。它只依赖于标头中的临时ReplyChannel。

我认为您的问题是您没有为您的任何@Configuration类指定@EnableIntgraiton：http://docs.spring.io/spring-integration/reference/html/overview.html#_configuration

编辑2

参见GH问题。

因此，除了代码之外，您还需要：

>

* <p>The given instance is supposed to be fully initialized; the registry
* will not perform any initialization callbacks (in particular, it won't
* call InitializingBean's {@code afterPropertiesSet} method).

在ctx之后就可以这样做了。refresh（）允许注册所有必要的BeanPostProcessor，包括一个用于Spring Integration Java DSL解析的BeanPostProcessor。

调用ctx。start（）启动所有生命周期。因为常规ctx无法看到这些手动添加的新内容。refresh（）处理。