Parâmetros-chave para o handshake TCP de três vias

Preparação do Programa

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.exemplo</groupId>
    <artifactId>TesteAltaConcorrencia</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.112.Final</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>3.1.2</version>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-dependency-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>copy-dependencies</goal>
                        </goals>
                        <configuration>
                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

src/main/resources/logback.xml

<configuration debug="false">
  <appender name="SAIDA_CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
    <encoder>
      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
    </encoder>
  </appender>

  <root level="${logLevel:-info}">
    <appender-ref ref="SAIDA_CONSOLE" />
  </root>
</configuration>

Lado do Servidor

org.exemplo.servidor.ServidorPrincipal

package org.exemplo.servidor;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.AttributeKey;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.System.out;

public class ServidorPrincipal {

    private static final AtomicInteger conexoesAtivas = new AtomicInteger(0);
    private static final ExecutorService poolTrabalho = Executors.newFixedThreadPool(4);

    public static void main(String[] args) {
        final int porta = 7070;
        EventLoopGroup grupoChefe = new NioEventLoopGroup(1);
        EventLoopGroup grupoTrabalhador = new NioEventLoopGroup(4);

        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(grupoChefe, grupoTrabalhador)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childHandler(new InicializadorCanalFilho());

            ChannelFuture futuro = bootstrap.bind(porta).sync();
            out.println("Servidor iniciado e escutando na porta: " + porta);
            futuro.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            out.println("Servidor interrompido: " + e.getMessage());
            Thread.currentThread().interrupt();
        } finally {
            grupoChefe.shutdownGracefully();
            grupoTrabalhador.shutdownGracefully();
            poolTrabalho.shutdown();
        }
    }

    private static class InicializadorCanalFilho extends ChannelInitializer<Channel> {
        @Override
        protected void initChannel(Channel ch) {
            ch.pipeline().addLast(new ManipuladorEventosConexao());
        }
    }

    @ChannelHandler.Sharable
    private static class ManipuladorEventosConexao extends ChannelInboundHandlerAdapter {
        private final AttributeKey<Integer> idConexao = AttributeKey.valueOf("idConexao");

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            int id = conexoesAtivas.incrementAndGet();
            ctx.channel().attr(idConexao).set(id);
            out.println("Canal " + ctx.channel().remoteAddress() + " ativado. Conexão #" + id + " estabelecida.");
            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            Integer id = ctx.channel().attr(idConexao).get();
            out.println("Canal " + ctx.channel().remoteAddress() + " inativado. Conexão #" + id + " encerrada.");
            super.channelInactive(ctx);
        }
    }
}

Comando de Inicialização do Servidor

java -cp TesteAltaConcorrencia-1.0-SNAPSHOT.jar:./lib/* org.exemplo.servidor.ServidorPrincipal -Djava.net.preferIPv4Stack=true -Xmx2g -Xms2g

Lado do Cliante

org.exemplo.cliente.ClientePrincipal

package org.exemplo.cliente;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.System.out;

public class ClientePrincipal {

    public static void main(String[] args) throws InterruptedException {
        if (args.length < 1) {
            out.println("Uso: java ClientePrincipal <num_conexoes> [host] [porta]");
            return;
        }

        final int totalConexoes = Integer.parseInt(args[0]);
        final String host = args.length > 1 ? args[1] : "localhost";
        final int porta = args.length > 2 ? Integer.parseInt(args[2]) : 7070;

        AtomicInteger conexoesEstabelecidas = new AtomicInteger(0);
        CountDownLatch latchConexoes = new CountDownLatch(totalConexoes);
        AttributeKey<Integer> attrId = AttributeKey.valueOf("clientId");

        NioEventLoopGroup grupoEvento = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap()
                .group(grupoEvento)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) {
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                int id = conexoesEstabelecidas.incrementAndGet();
                                ctx.channel().attr(attrId).set(id);
                                out.println("Conexão #" + id + " com " + ctx.channel().remoteAddress() + " estabelecida.");
                                latchConexoes.countDown();
                                super.channelActive(ctx);
                            }

                            @Override
                            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                out.println("Conexão #" + ctx.channel().attr(attrId).get() + " encerrada.");
                                super.channelInactive(ctx);
                            }
                        });
                    }
                });

        out.println("Iniciando " + totalConexoes + " conexões com " + host + ":" + porta);
        long tempoInicio = System.currentTimeMillis();

        ChannelFuture[] futurosConexao = new ChannelFuture[totalConexoes];
        for (int i = 0; i < totalConexoes; i++) {
            futurosConexao[i] = bootstrap.connect(host, porta);
        }

        boolean todasConectadas = latchConexoes.await(30, TimeUnit.SECONDS);
        long tempoFim = System.currentTimeMillis();

        if (todasConectadas) {
            out.println("Todas as " + totalConexoes + " conexões estabelecidas em " + (tempoFim - tempoInicio) + "ms");
        } else {
            out.println("Apenas " + (totalConexoes - latchConexoes.getCount()) + " de " + totalConexoes + " conexões foram estabelecidas no tempo limite.");
        }

        // Manter conexões abertas por um momento para demonstração
        Thread.sleep(5000);

        for (ChannelFuture futuro : futurosConexao) {
            if (futuro.channel().isActive()) {
                futuro.channel().close();
            }
        }
        grupoEvento.shutdownGracefully().sync(5, TimeUnit.SECONDS);
    }
}

Parâmetros do Sistema Operacional

• fs.file-max: Define o número máximo de descritores de arquivo no sistema. Deve ser aumentado para suportar um grande número de conexões.
• net.core.somaxconn: Define o tamenho máximo da fila de pendência (backlog) do socket no kernel. Um valor baixo pode limitar a taxa de novas conexões em ambientes de alta concorrência.