Spring集成超时客户端

Question

我对Spring集成的设想是：

1. 使用自定义协议（大小和内容）发送数据的十个生产者
2. 我必须解码这个自定义协议，然后处理结果。

所以我尝试了很多配置，目前最好的配置如下：

<bean id="serializer" class="com.MySerializerDeserializer" />
<task:executor id="myTaskExecutor" pool-size="5-300" queue-capacity="500000"/>
<int-ip:tcp-connection-factory id="serverTcpConFact"
    type="server"
    port="5566"
    using-nio="true"
    single-use="false"
    so-timeout="5000"
    task-executor="myTaskExecutor"
    deserializer="serializer" 
    serializer="serializer"/>

<int-ip:tcp-inbound-channel-adapter id="tcpInboundAdapter"
    channel="tcpInbound"
    connection-factory="serverTcpConFact" />

<int:channel id="tcpInbound" />

<int:service-activator input-channel="tcpInbound"
    ref="importService"
    method="handler" />

<bean id="importService" class="com.MyImportService" />

序列化类为：

public class MySerializerDeserializer implements Serializer< MyMessage >, Deserializer< MyMessage > {

    @Override
    public MyMessage deserialize(InputStream inputStream) throws IOException {
        DataInputStream dis = new DataInputStream(inputStream);
        int size = dis.readInt();
        byte[] b = new byte[size];
        dis.read(b);
        MyMessage s = new MyMessage ();
        String value = new String(b);
        s.setTest(value);
        return s;

我使用此代码来测试服务器：

class SimpleThread extends Thread {
    public void run() {
        try {
            String id = java.util.UUID.randomUUID().toString();
            try (Socket echoSocket = new Socket("localhost", 5566)) {
                DataOutputStream dos = new DataOutputStream(echoSocket.getOutputStream());
                for (int i = 0; i < 500; i++) {
                    String s = id + " " + i;
               dos.writeInt(s.length());
                    dos.write(s.getBytes());
                    dos.flush();
                    System.out.println(id + " - " + i + " - " + s.length());
                }
            }

当我用一个线程执行此操作时，如果我尝试执行多个线程，则效果很好，如：

  for (int i = 0; i < 5; i++) {
 new SimpleThread().start();
 }

spring集成服务器卡住了，我有以下警告：

WARN _[m [THREAD ID=myTaskExecutor-1] 2014-06-25 12:42:18 WARN  org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory:566 - Timing out TcpNioConnection 127.0.0.1:56273:5566:4e3caf61-1101-4881-a1cf-1c31610b33f3

而且它不工作，服务器无法接收消息。

我错在哪里？非常感谢。

编辑

我这样修改了线程轮询：

<bean id="myTaskExecutor" class="org.springframework.integration.util.CompositeExecutor">
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="io-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="8" />
            <property name="queueCapacity" value="50000" />
            <property name="rejectedExecutionHandler">
                <bean class="org.springframework.integration.util.CallerBlocksPolicy">
                    <constructor-arg value="10000" />
                </bean>
            </property>
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
            <property name="threadNamePrefix" value="assembler-" />
            <property name="corePoolSize" value="4" />
            <property name="maxPoolSize" value="10" />
            <property name="queueCapacity" value="50000" />
            <property name="rejectedExecutionHandler">
                <bean class="org.springframework.integration.util.CallerBlocksPolicy">
                    <constructor-arg value="10000" />
                </bean>
            </property>
        </bean>
    </constructor-arg>
</bean>

服务器更有响应性，但我仍然有问题

**编辑*****

导入服务是：

public class ImportService {

    public void handler(MyMessage inp) {
        System.out.println(Thread.currentThread().getName() + "******" + inp.getTest());

    }

Answer 1

@JR您可能想看看这个与您类似的示例。

https://github.com/rajeshgheware/spring-integration-samples/tree/master/spring-integration-samples-tcp-concurrent-server

我成功地为3900个并发客户端测试了这段代码。

Answer 2

我只是严格按照描述运行了您的测试（复制了您的代码），所有测试都按预期进行。500条消息传递到另一端。

我建议您打开跟踪级别日志记录，并可能向反序列化程序添加一些调试日志记录。