java - Spring Integration TCP 连接在多线程负载/压力测试下被拒绝

Question

我有一个处理 TCP 消息的 Spring Integration 项目。

最简单的场景是PING消息(从source接收msg并回显)，SI项目中的流程如下:

1) 从源接收消息(通过 tcp-inbound-gateway)。源在每条消息后关闭套接字。

2) 转换器分析消息并设置(除其他外)带有回复 channel 名称的 header 值

3) 将 Header-Value-Router 应用于将其路由回源的消息。

XML 配置(简化版)如下:

<int-ip:tcp-connection-factory id="TCP_SRV"
                               type="server"
                               port="${router.port}"
                               using-nio="true"
                               single-use="true"
                               serializer="CustomSerializer"
                               deserializer="CustomSerializer"/>

<int-ip:tcp-inbound-gateway request-channel="rawInputFromSource"
                            reply-channel="outputBackToSource"
                            connection-factory="TCP_SRV"/>

<int:channel id="rawInputFromSource"/>

<int:transformer ref="inputFromSourceTransformer"
                 input-channel="rawInputFromSource"
                 output-channel="processedInputFromSource"/>

<int:channel id="processedInputFromSource"/>

<bean id="inputFromSourceTransformer" class="my.org.InputFromSourceTransformer"/>

<int:header-value-router input-channel="processedInputFromSource" 
                         header-name="RouteToChannel"/>

它在手动调用消息时从功能 pov 运行正常，但在压力测试下失败。一旦我增加超过 15 个线程(每个线程运行一个 for 循环发送 10 条消息)，我收到 java.net.ConnectException: Connection refused: connect 大约 20% 的尝试。

线程用来发送消息的代码片段:

byte[] sendAndReceive(byte[] data){
    byte[] result = new byte[data.length];
    try {
        Socket socket=new Socket("localhost", SI_PORT); // here is where the err occurs
        OutputStream output = socket.getOutputStream();
        InputStream  input = socket.getInputStream();
        output.write(data);
        input.read(result);
        socket.close();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    return result;
}

错误:

java.net.ConnectException: Connection refused: connect
at java.net.TwoStacksPlainSocketImpl.socketConnect(Native Method)
at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
at java.net.PlainSocketImpl.connect(PlainSocketImpl.java:172)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
at java.net.Socket.connect(Socket.java:589)
at java.net.Socket.connect(Socket.java:538)
at java.net.Socket.<init>(Socket.java:434)
at java.net.Socket.<init>(Socket.java:211)
at my.org.PerformanceTest.sendAndReceive(PerformanceTest.java:98)

要求说它必须通过 60 个线程。我能做些什么来解决这个问题？我尝试在工厂添加一个 task:executor id="threadPoolTask​​Executor"pool-size="5-10"queue-capacity="100"rejection-policy="CALLER_RUNS" 但这没有解决了问题。

非常感谢任何建议

Answer 1

增加服务器连接工厂的积压。

/**
 * The number of sockets in the connection backlog. Default 5;
 * increase if you expect high connection rates.
 * @param backlog The backlog to set.
 */
public void setBacklog(int backlog) {
    Assert.isTrue(backlog >= 0, "You cannot set backlog negative");
    this.backlog = backlog;
}

在具有 backlog 属性的 XML 配置中可用...

<xsd:attribute name="backlog" type="xsd:string">
    <xsd:annotation>
        <xsd:documentation>
            Specifies the connection backlog for server sockets. Does not
            apply to client factories.
        </xsd:documentation>
    </xsd:annotation>
</xsd:attribute>