Estoy usando Spring Integration para el servidor TCP que mantiene conexiones a unos pocos miles de clientes. Necesito el servidor para estrangular a los clientes en caso de una carga excesiva y no perder mensajes.

Mi configuración del servidor:

<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" />

Como el ejecutor de tareas predeterminado para la fábrica de conexiones no tiene límites, utilizo un ejecutor de tareas agrupado para evitar errores de falta de memoria.

Un cliente simple para pruebas de carga:

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();
    }
}

Cuando lo ejecuto, el servidor solo recibe unos 10-20 mensajes y luego el cliente recibe la excepción "Conexión rechazada: conectar". Después de eso, el servidor ya no puede aceptar conexiones nuevas, incluso después del tiempo de espera de conexión. Aumentar el tamaño del grupo solo ayuda a recibir un poco más de mensajes.

EDITAR

Estoy usando Spring Integration 3.0.2.RELEASE. Para la producción estoy usando 8-40 hilos, pero solo hace que esta prueba falle más tarde, después de varios cientos de conexiones.

MyService.test () no hace mucho ...

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

Aquí está el registro con el registro de nivel de rastreo.

Fuentes