Spring集成-用于大容量应用的可靠TCP

Question

我使用服务器，它保持与几千个客户端的连接。我需要服务器来节流客户端，以防负载过大，而且不会丢失消息。

我的服务器配置：

<task:executor id="myTaskExecutor"
    pool-size="4-8"
    queue-capacity="0"
    rejection-policy="CALLER_RUNS" />

<int-ip:tcp-connection-factory id="serverTcpConFact"
    type="server"
    port="60000"
    using-nio="true"
    single-use="false"
    so-timeout="300000"
    task-executor="myTaskExecutor" />

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

<channel id="tcpInbound" />

<service-activator input-channel="tcpInbound"
    ref="myService"
    method="test" />

<beans:bean id="myService" class="org.test.tcpserver.MyService" />

由于连接工厂的默认任务执行器是无限制的，所以我使用池任务执行器来防止内存不足。

用于负载测试的简单客户端：

public class TCPClientTest {
    static Socket socket;
    static List<Socket> sl = new ArrayList<>();
    static DataOutputStream out;

    public static void main(String[] args) throws Exception {
        for (int i = 0; i < 10000; i++) {
            socket = new Socket("localhost", 60000);
            sl.add(socket);
            out = new DataOutputStream(socket.getOutputStream());
            out.writeBytes("connection " + i + "\r\n");
            System.out.println("Using connection #" + i);
        }
        System.in.read();
    }
}

当我运行它时，服务器只接收大约10-20条消息，然后客户端得到“ConnectionResive: connect”异常。之后，即使在连接超时之后，服务器也不能接受任何新的连接。增加池的大小只会帮助获得更多的消息。

编辑

我使用SpringIntegration3.0.2.RELEASE，在生产中我使用了8-40个线程，但是它只会使这个测试在几百个连接之后失败。

MyService.test()做的不多..。

public class MyService {
    public void test(byte[] input) {
        System.out.println("Received: " + new String(input));
    }
}

下面是带有跟踪级别日志的日志。

资料来源

Answer 1

我知道是什么问题了，请打开一个JIRA问题。

问题是在执行器中有一个CALLER_RUNS长度队列的0拒绝策略。

有一个线程处理所有IO事件(通常是myTaskExecutor-1)；当读取事件触发时，他会排队执行以读取数据；读取器线程将队列执行以组装数据(在您的情况下，由CRLF终止)将阻塞完整消息)。

在这种情况下，当没有可用的线程时，CALLER_RUNS策略意味着IO选择器线程执行读取，并成为汇编程序线程，该线程阻止等待由于他被阻塞而无法到达的数据，并在调度了不同的线程之后读取数据。因为他被阻止了，所以他无法处理新的接受事件。

这是我测试的日志显示了问题..。

TRACE: [May-18 10:43:38,923][myTaskExecutor-1] tcp.connection.TcpNioServerConnectionFactory - Port 60000 SelectionCount: 2
DEBUG: [May-18 10:43:38,923][myTaskExecutor-1] tcp.connection.TcpNioConnection - localhost:58509:60000:bdc36c59-c31b-470e-96c3-6270e7c46a2f Reading...
DEBUG: [May-18 10:43:38,924][myTaskExecutor-1] tcp.connection.TcpNioConnection - localhost:58509:60000:bdc36c59-c31b-470e-96c3-6270e7c46a2f Running an assembler
TRACE: [May-18 10:43:38,924][myTaskExecutor-1] tcp.connection.TcpNioConnection - localhost:58509:60000:bdc36c59-c31b-470e-96c3-6270e7c46a2f Nio message assembler running...
DEBUG: [May-18 10:43:38,926][myTaskExecutor-1] tcp.serializer.ByteArrayCrLfSerializer - Available to read:0

第二行显示用于读取的选择器线程；他检测到此套接字需要汇编程序，并成为汇编程序，阻塞，等待数据。

你真的相信使用无限制的任务执行器会出现问题吗？这些事件通常都很短，所以线程会很快被回收。

将执行器的队列容量增加到0以上也会有帮助，但这并不能完全保证问题不会发生(尽管不太可能出现较大的队列大小)。

除了为IO选择器和读取器线程使用专用的任务执行器之外，我还不知道如何解决这个问题，这样它们就不会被用作汇编程序。

Answer 2

昨天，我编写了一个示例来创建使用spring集成的tcp高性能服务器代码。我使用JMeter TCP采样器成功地对1000个并发客户端请求进行了测试。

下面是代码- https://github.com/rajeshgheware/spring-integration-samples，包括JMeter测试配置文件。

我在64位笔记本电脑上成功地测试了1000个并发的tcp客户端请求，其中包含英特尔核心i5 M520 2.4GHz (服务器代码和运行在这台机器上的JMeter测试)。

我还尝试了1500个并发客户端请求，但注意到服务器不能满足许多请求。我将继续努力增强这段代码，以服务10000个并发客户端请求(我知道我可能需要从亚马逊获得一个很好的EC2机器，用于这个测试:)