A Terceira Pedra — Mil Conexões ao Mesmo Tempo: Virtual Threads e Keep-Alive

Nesta aula o Farol passa a atender cada conexão numa thread virtual própria, mantém conexões abertas entre requisições, distingue conexões ociosas de requisições lentas, limita as conexões simultâneas e desliga com ordem, e um cliente de carga escrito à mão mede o que tudo isso vale.

"A Primeira Pedra" terminou com um número incômodo: um curl esperou 9,4 segundos por uma resposta de milissegundos, porque um cliente calado ocupava o único atendente do Farol. E "A Segunda Pedra" mostrou, em cada resposta, o cabeçalho Connection: close: o Farol fecha a conexão depois de cada requisição, e um navegador que precisa de uma página, uma folha de estilo e dez imagens abre doze conexões. Esta aula resolve os dois problemas, aplicando ao servidor o que a Fase 7 inteira ensinou. As threads virtuais de "Um Milhão de Threads" permitem o estilo mais simples de servidor, uma thread por conexão, sem limite prático de conexões. O Semaphore de "Ferramentas de Precisão" limita o que é escasso. E o desligamento ordenado de "Não Crie Threads, Entregue Tarefas" garante que nenhuma requisição em andamento seja cortada ao meio.

Uma thread virtual por conexão

A mudança central é pequena: em vez de atender a conexão na mesma thread que a aceitou, o servidor a entrega a uma thread virtual nova e volta imediatamente ao accept(). A nova versão do Servidor:

package br.com.exemplo.farol.nucleo;

import br.com.exemplo.farol.http.*;
import br.com.exemplo.farol.protocolo.EscritorDeResposta;

import java.io.*;
import java.net.*;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.*;

/**
 * O núcleo do Farol: aceita conexões e entrega cada uma a uma thread virtual própria.
 */
public final class Servidor implements AutoCloseable {
    private final ServerSocket soqueteDoServidor;
    private final Configuracao configuracao;
    private final Manipulador manipulador;
    private final EscritorDeResposta escritor = new EscritorDeResposta();

    private final ExecutorService conexoes =
            Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("conexao-", 1).factory());
    private final Semaphore vagas;
    private final Set<Conexao> abertas = ConcurrentHashMap.newKeySet();
    private final CountDownLatch desligado = new CountDownLatch(1);
    private volatile Thread aceitador;

    public Servidor(int porta, Configuracao configuracao, Manipulador manipulador) throws IOException {
        this.configuracao = Objects.requireNonNull(configuracao);
        this.manipulador = Objects.requireNonNull(manipulador);
        this.vagas = new Semaphore(configuracao.maximoDeConexoes());
        this.soqueteDoServidor = new ServerSocket(porta, 1024);   // fila de espera do accept: 1024 conexões
    }

    public int porta() {
        return soqueteDoServidor.getLocalPort();
    }

    /** Número de conexões abertas neste momento. */
    public int conexoesAbertas() {
        return abertas.size();
    }

    /** Começa a aceitar conexões numa thread própria e retorna imediatamente. */
    public Servidor iniciar() {
        aceitador = Thread.ofPlatform().name("farol-aceitador").start(this::aceitarEnquantoAberto);
        Registro.info("Farol ouvindo em http://localhost:" + porta());
        return this;
    }

    /** Bloqueia até o servidor ser desligado. */
    public void aguardar() throws InterruptedException {
        desligado.await();
    }

    private void aceitarEnquantoAberto() {
        while (!soqueteDoServidor.isClosed()) {
            Socket soquete;
            try {
                soquete = soqueteDoServidor.accept();
            } catch (IOException e) {
                if (!soqueteDoServidor.isClosed()) Registro.erro("falha ao aceitar conexão", e);
                continue;
            }
            if (!vagas.tryAcquire()) {
                conexoes.submit(() -> recusarPorLotacao(soquete));
                continue;
            }
            Conexao conexao = new Conexao(soquete, configuracao, manipulador, escritor, vagas::release);
            abertas.add(conexao);
            try {
                conexoes.submit(() -> {
                    try {
                        conexao.run();
                    } finally {
                        abertas.remove(conexao);
                    }
                });
            } catch (RejectedExecutionException e) {   // chegou durante o desligamento
                abertas.remove(conexao);
                vagas.release();
                conexao.fecharAgora();
            }
        }
    }

    private void recusarPorLotacao(Socket soquete) {
        try (soquete) {
            soquete.setSoTimeout(1000);
            var resposta = Resposta.erro(Status.INDISPONIVEL, "servidor lotado, tente de novo").comCabecalho("Retry-After", "1");
            escritor.escrever(resposta, new BufferedOutputStream(soquete.getOutputStream()), false, true);
        } catch (IOException e) {
            // o cliente já foi embora
        }
    }

    /**
     * Desligamento ordenado: para de aceitar, fecha as conexões ociosas, deixa as ocupadas
     * terminarem a requisição atual e, depois do prazo, fecha o que sobrar.
     */
    @Override
    public void close() throws IOException {
        if (soqueteDoServidor.isClosed()) return;
        Registro.info("desligando: " + abertas.size() + " conexões abertas");
        soqueteDoServidor.close();
        abertas.forEach(Conexao::encerrar);
        conexoes.shutdown();
        try {
            if (!conexoes.awaitTermination(configuracao.prazoDeDesligamento().toMillis(), TimeUnit.MILLISECONDS)) {
                Registro.info("prazo esgotado: fechando " + abertas.size() + " conexões à força");
                abertas.forEach(Conexao::fecharAgora);
                conexoes.shutdownNow();
            }
            if (aceitador != null) aceitador.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        Registro.info("Farol desligado");
        desligado.countDown();
    }
}

O executor é criado com Executors.newThreadPerTaskExecutor e uma fábrica de threads virtuais com nome: Thread.ofVirtual().name("conexao-", 1) numera as threads a partir de 1. O newVirtualThreadPerTaskExecutor de "Um Milhão de Threads" faria o mesmo trabalho, mas com threads sem nome, e o registro do servidor ficaria cheio de números sem sentido. Nome de thread é uma ferramenta de diagnóstico barata, como "Não Crie Threads, Entregue Tarefas" já tinha mostrado com a ThreadFactory.

O accept() roda numa única thread de plataforma, a farol-aceitador, que não faz nada além de aceitar e distribuir. O construtor do ServerSocket recebe agora um segundo número, o tamanho da fila de espera do sistema operacional: conexões que chegam enquanto o aceitador está ocupado esperam ali, e o padrão do Java, 50, é pequeno para picos de milhares de conexões.

Três estruturas de "Ferramentas de Precisão" cuidam da contabilidade. O Semaphore vagas tem uma licença por conexão permitida, e é o limite do recurso que "Um Milhão de Threads" recomendou: as threads virtuais são baratas, mas cada conexão ocupa um descritor de arquivo do sistema operacional, memória para os buffers e, no mundo real, talvez uma conexão com o banco de dados. Quando não há vaga, a conexão é recusada com 503 Service Unavailable e o cabeçalho Retry-After, que sugere ao cliente quando tentar de novo. A recusa acontece numa thread virtual própria, para que um cliente lento em ler o 503 não atrase o aceitador. O conjunto abertas, criado com ConcurrentHashMap.newKeySet(), registra as conexões vivas, para o desligamento. E o CountDownLatch desligado permite que o main espere o servidor terminar.

A API também mudou: iniciar() começa a aceitar conexões numa thread própria e retorna, e aguardar() bloqueia até o desligamento. Nos testes, isso permite subir um servidor e usá-lo na mesma thread; no Principal, permite instalar o tratamento de desligamento antes de esperar.

Conexões persistentes

A outra metade da aula está na nova classe Conexao, que atende várias requisições pela mesma conexão, uma depois da outra, enquanto ela for persistente:

package br.com.exemplo.farol.nucleo;

import br.com.exemplo.farol.http.*;
import br.com.exemplo.farol.protocolo.*;

import java.io.*;
import java.net.*;
import java.util.Locale;
import java.util.Optional;

/**
 * Uma conexão com um cliente, atendida do começo ao fim por uma única thread virtual.
 * Atende várias requisições em sequência enquanto a conexão for persistente (keep-alive).
 */
final class Conexao implements Runnable {
    private final Socket soquete;
    private final Configuracao configuracao;
    private final Manipulador manipulador;
    private final EscritorDeResposta escritor;
    private final Runnable aoTerminar;

    private volatile boolean ocupada = false;      // atendendo uma requisição agora?
    private volatile boolean encerrando = false;   // o servidor pediu para fechar depois da atual?

    Conexao(Socket soquete, Configuracao configuracao, Manipulador manipulador,
            EscritorDeResposta escritor, Runnable aoTerminar) {
        this.soquete = soquete;
        this.configuracao = configuracao;
        this.manipulador = manipulador;
        this.escritor = escritor;
        this.aoTerminar = aoTerminar;
    }

    @Override
    public void run() {
        try (soquete) {
            atenderEnquantoViva();
        } catch (SocketException e) {
            // conexão fechada pelo cliente, ou por nós durante o desligamento: nada a registrar
        } catch (IOException e) {
            Registro.info("conexão perdida: " + e.getMessage());
        } catch (RuntimeException e) {
            Registro.erro("falha inesperada numa conexão", e);
        } finally {
            aoTerminar.run();
        }
    }

    private void atenderEnquantoViva() throws IOException {
        Limites limites = configuracao.limites();
        var leitor = new LeitorDeRequisicao(soquete.getInputStream(), limites);
        var saida = new BufferedOutputStream(soquete.getOutputStream());

        for (int atendidas = 0; ; atendidas++) {
            // 1. Esperar a próxima requisição, com o prazo de conexão ociosa
            soquete.setSoTimeout((int) configuracao.prazoOcioso().toMillis());
            try {
                if (encerrando || !leitor.aguardarProxima()) return;
            } catch (SocketTimeoutException e) {
                if (atendidas == 0)   // conectou e nunca pediu nada: avisa antes de fechar
                    escritor.escrever(Resposta.erro(Status.TEMPO_ESGOTADO, "nenhuma requisição recebida"), saida, false, true);
                return;               // conexão ociosa entre requisições: fecha em silêncio
            }

            // 2. Ler a requisição inteira, com o prazo de leitura
            ocupada = true;
            soquete.setSoTimeout((int) limites.prazoDeLeitura().toMillis());
            Requisicao requisicao;
            try {
                Optional<Requisicao> lida = leitor.ler();
                if (lida.isEmpty()) return;
                requisicao = lida.get();
            } catch (ErroDeProtocolo e) {
                // Depois de um erro de protocolo não dá para saber onde começa a próxima requisição: fecha
                Registro.info("recusada: " + e.status() + ": " + e.getMessage());
                escritor.escrever(Resposta.erro(e.status(), e.getMessage()), saida, false, true);
                return;
            } catch (SocketTimeoutException e) {
                escritor.escrever(Resposta.erro(Status.TEMPO_ESGOTADO, "requisição incompleta"), saida, false, true);
                return;
            }

            // 3. Atender e responder, decidindo se a conexão continua
            Resposta resposta = executar(requisicao);
            boolean continuar = querManterAberta(requisicao)
                    && atendidas + 1 < configuracao.maximoDeRequisicoesPorConexao()
                    && !encerrando;
            escritor.escrever(resposta, saida, requisicao.metodo() == Metodo.HEAD, !continuar);
            Registro.info(requisicao.metodo() + " " + requisicao.alvo().original() + " → " + resposta.status().codigo());
            ocupada = false;
            if (!continuar) return;
        }
    }

    /** HTTP/1.1 é persistente por padrão; HTTP/1.0 só se o cliente pedir. */
    private static boolean querManterAberta(Requisicao r) {
        String connection = r.cabecalhos().valor("Connection").orElse("").toLowerCase(Locale.ROOT);
        if (r.versao().equals("HTTP/1.0")) return connection.contains("keep-alive");
        return !connection.contains("close");
    }

    private Resposta executar(Requisicao requisicao) {
        try {
            return manipulador.atender(requisicao);
        } catch (Exception e) {
            Registro.erro("o manipulador falhou em " + requisicao.metodo() + " " + requisicao.alvo().original(), e);
            return Resposta.erro(Status.ERRO_INTERNO, "erro interno");
        }
    }

    /** Chamado pelo servidor ao desligar: conexões ociosas fecham já; as ocupadas, depois da resposta. */
    void encerrar() {
        encerrando = true;
        if (!ocupada) fecharAgora();
    }

    void fecharAgora() {
        try {
            soquete.close();   // um read() bloqueado nesta conexão lança SocketException
        } catch (IOException e) {
            // já estava fechada
        }
    }
}

As regras da RFC 9112 para conexões persistentes estão no método querManterAberta. No HTTP/1.1, a conexão é persistente por padrão, e o cliente só pede para fechá-la com Connection: close. No HTTP/1.0, é o contrário: a conexão fecha por padrão, e o cliente pede para mantê-la com Connection: keep-alive. A conexão também fecha depois de um número máximo de requisições, para que um cliente não monopolize uma conexão para sempre e para que os recursos acumulados nela sejam liberados de tempos em tempos. Quando o servidor decide fechar, avisa o cliente na última resposta, com Connection: close, e o cliente sabe que precisa abrir outra conexão para a próxima requisição.

Tudo isso só funciona por causa de uma decisão tomada duas aulas atrás: o leitor lê exatamente Content-Length bytes de corpo, e o escritor calcula o Content-Length de cada resposta. Numa conexão persistente, não existe "fim da conexão" para marcar o fim de uma mensagem, e o tamanho declarado é a única fronteira entre uma requisição e a seguinte. O teste naoLeAlemDoCorpo, de "A Primeira Pedra", era a preparação para este momento.

Um erro de protocolo, por outro lado, sempre fecha a conexão. Depois de uma requisição malformada, não há como saber onde a próxima começa: se o corpo de um 413 não foi lido, os bytes dele ainda estão no socket, e seriam interpretados como uma nova requisição. Já uma falha no manipulador, que vira 500, deixa a conexão aberta: a requisição foi lida por inteiro, e o fluxo de bytes continua sincronizado.

Dois prazos diferentes

Com conexões persistentes, um único prazo de leitura deixa de fazer sentido, porque há dois tipos de silêncio. Uma conexão ociosa é aquela que terminou uma requisição e está esperando a próxima; é normal, e o navegador a mantém aberta justamente para reaproveitá-la, mas ela ocupa um descritor e uma vaga, e deve ser fechada depois de alguns segundos. Uma requisição lenta é aquela que já começou a chegar e parou no meio; isso é anormal, e pode ser um cliente com rede ruim ou um ataque. O Farol usa um prazo para cada caso: 5 segundos para a ociosidade e 10 segundos para completar uma requisição.

Para saber em qual dos dois casos está, a conexão precisa de uma forma de esperar o primeiro byte da próxima requisição sem consumi-lo, e trocar o prazo assim que ele chegar. O BufferedInputStream tem a ferramenta para isso, o par mark e reset, e o EntradaHttp ganha um método:

/**
 * Espera até chegar pelo menos um byte, sem consumi-lo.
 * @return false se a conexão foi encerrada pelo cliente
 */
boolean aguardarDados() throws IOException {
    entrada.mark(1);
    int b = entrada.read();
    if (b == -1) return false;
    entrada.reset();                                 // devolve o byte ao buffer
    return true;
}

O mark(1) marca a posição atual, o read() bloqueia até um byte chegar, e o reset() volta à marca, devolvendo o byte ao buffer para que o leitor o encontre normalmente. O campo entrada do EntradaHttp passou a ser declarado como BufferedInputStream, com o construtor usando o padrão instanceof BufferedInputStream b ? b : new BufferedInputStream(entrada), porque mark e reset só funcionam de forma confiável com o buffer. O LeitorDeRequisicao expõe o método como aguardarProxima(), e a Conexao o chama com o prazo ocioso, troca para o prazo de leitura e só então chama ler().

Quando o prazo ocioso esgota, há uma distinção sutil. Se a conexão nunca enviou nada, o cliente conectou e ficou calado, e o 408 de "A Primeira Pedra" continua sendo a resposta. Se a conexão já atendeu requisições, o silêncio é a ociosidade normal, e o servidor fecha sem responder nada: um 408 enviado a um navegador que só estava mantendo a conexão aberta apareceria para ele como uma resposta a uma requisição que ele ainda não fez.

A configuração

Os novos parâmetros não pertencem à camada de protocolo, que trata de uma requisição de cada vez, e sim ao núcleo, que trata de conexões. Eles ficam num record próprio, que contém os Limites:

package br.com.exemplo.farol.nucleo;

import br.com.exemplo.farol.protocolo.Limites;

import java.time.Duration;
import java.util.Objects;

/** Como o servidor trata as conexões: quantas, por quanto tempo e como desligar. */
public record Configuracao(
        Limites limites,                     // limites de cada requisição (camada de protocolo)
        Duration prazoOcioso,                // conexão sem nova requisição por mais tempo: fechada
        int maximoDeRequisicoesPorConexao,   // depois disso, a conexão é fechada e o cliente abre outra
        int maximoDeConexoes,                // conexões simultâneas; acima disso: 503
        Duration prazoDeDesligamento) {      // quanto esperar as requisições em andamento ao desligar

    public static final Configuracao PADRAO =
            new Configuracao(Limites.PADRAO, Duration.ofSeconds(5), 1000, 10_000, Duration.ofSeconds(10));

    public Configuracao {
        Objects.requireNonNull(limites, "limites");
        if (prazoOcioso.isNegative() || prazoOcioso.isZero()) throw new IllegalArgumentException("prazo ocioso deve ser positivo");
        if (maximoDeRequisicoesPorConexao < 1) throw new IllegalArgumentException("máximo de requisições deve ser >= 1");
        if (maximoDeConexoes < 1) throw new IllegalArgumentException("máximo de conexões deve ser >= 1");
        if (prazoDeDesligamento.isNegative()) throw new IllegalArgumentException("prazo de desligamento negativo");
    }

    public Configuracao comLimites(Limites l) {
        return new Configuracao(l, prazoOcioso, maximoDeRequisicoesPorConexao, maximoDeConexoes, prazoDeDesligamento);
    }
    public Configuracao comPrazoOcioso(Duration d) {
        return new Configuracao(limites, d, maximoDeRequisicoesPorConexao, maximoDeConexoes, prazoDeDesligamento);
    }
    public Configuracao comMaximoDeRequisicoesPorConexao(int n) {
        return new Configuracao(limites, prazoOcioso, n, maximoDeConexoes, prazoDeDesligamento);
    }
    public Configuracao comMaximoDeConexoes(int n) {
        return new Configuracao(limites, prazoOcioso, maximoDeRequisicoesPorConexao, n, prazoDeDesligamento);
    }
    public Configuracao comPrazoDeDesligamento(Duration d) {
        return new Configuracao(limites, prazoOcioso, maximoDeRequisicoesPorConexao, maximoDeConexoes, d);
    }
}

Os métodos com... são os withers de "Contratos à Vista": como o record é imutável, "alterar" um parâmetro produz uma cópia, e as validações do construtor compacto rodam a cada cópia. Com eles, um teste escreve Configuracao.PADRAO.comPrazoOcioso(Duration.ofMillis(300)), que diz exatamente o que difere do padrão, em vez de um construtor com cinco argumentos em que só um importa.

Desligar com ordem

O close() do servidor implementa o desligamento em quatro passos, na ordem certa. Primeiro, fecha o ServerSocket, e nenhuma conexão nova é aceita. Segundo, pede a cada conexão aberta que encerre: as ociosas são fechadas na hora, e o read() bloqueado nelas lança SocketException, que a Conexao trata em silêncio; as ocupadas recebem o aviso pelo campo volatile encerrando, terminam a requisição atual, respondem com Connection: close e saem do laço. Terceiro, espera o executor terminar, com o prazo de desligamento. Quarto, se o prazo esgotar, fecha à força o que sobrou.

Os dois campos volatile da Conexao são a aplicação direta de "O Cadeado e a Memória": a thread que chama close() e a thread virtual da conexão são diferentes, e sem volatile não haveria garantia de que uma visse o que a outra escreveu. Um sinal que uma thread grava e outra lê, sem nenhuma operação composta, é exatamente o caso em que volatile basta.

O Principal conecta esse desligamento ao sistema operacional com um gancho de desligamento (shutdown hook), uma thread que a JVM executa quando recebe o sinal de término, como o Ctrl+C no terminal ou o kill que um orquestrador de contêineres envia:

package br.com.exemplo.farol.app;

import br.com.exemplo.farol.http.Resposta;
import br.com.exemplo.farol.nucleo.Configuracao;
import br.com.exemplo.farol.nucleo.Servidor;
import br.com.exemplo.farol.rotas.Roteador;

import java.time.Duration;

public class Principal {
    public static void main(String[] args) throws Exception {
        int porta = args.length > 0 ? Integer.parseInt(args[0]) : 8080;

        Roteador rotas = new Roteador()
                .get("/", req -> Resposta.html("<h1>Farol</h1><p>Bem-vindo à Ilha do Conhecimento.</p>"))
                .get("/ola/{nome}", req -> Resposta.texto("Olá, " + req.parametro("nome") + "!\n"))
                .get("/lento", req -> {                        // simula uma consulta demorada ao banco
                    Thread.sleep(Duration.ofSeconds(1));
                    return Resposta.texto("demorou, mas veio\n");
                });
        new ApiDeTarefas().registrarEm(rotas);

        Servidor farol = new Servidor(porta, Configuracao.PADRAO, rotas).iniciar();
        // Ctrl+C ou "kill" (SIGTERM): desliga com ordem antes de a JVM terminar
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                farol.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }));
        farol.aguardar();
    }
}

A rota /lento, que dorme um segundo, simula uma consulta demorada a um banco de dados e vai ser útil nas medições. Para ver o desligamento funcionando, vinte clientes pedem /lento e cinco ficam com conexões ociosas abertas; quase um segundo depois, antes de as consultas terminarem, o servidor recebe um kill -TERM:

15:09:28.869 [main] Farol ouvindo em http://localhost:8080
15:09:31.144 [Thread-0] desligando: 25 conexões abertas
15:09:31.512 [Thread-0] Farol desligado

20 clientes x 1 requisições (keep-alive): 20 respostas {200=20}, 0 falhas, 1.14 s, 18 req/s

As 25 conexões estavam abertas no momento do sinal. As cinco ociosas foram fechadas imediatamente; as vinte ocupadas terminaram suas consultas, e os vinte clientes receberam um 200 completo, com Connection: close. Só então, 370 ms depois do sinal, o servidor terminou. Nenhuma requisição foi cortada. Um kill -9, por outro lado, mata o processo sem executar o gancho, e o exercício 5 discute o que isso significa.

Medindo

Para medir o Farol, o curl não basta: ele faz uma requisição de cada vez. O cliente de carga abaixo cria um número de clientes simultâneos, cada um numa thread virtual, cada um fazendo um número de requisições, com ou sem keep-alive, e conta as respostas por código de estado:

import java.io.*;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

// Uso: java Carga.java <clientes> <requisicoesPorCliente> <caminho> <manter|fechar>
static String lerCabecalho(InputStream in) throws IOException {
    ByteArrayOutputStream b = new ByteArrayOutputStream();
    int c, estado = 0;                                   // procura \r\n\r\n
    while ((c = in.read()) != -1) {
        b.write(c);
        estado = (c == '\r' && (estado == 0 || estado == 2)) || (c == '\n' && (estado == 1 || estado == 3)) ? estado + 1 : 0;
        if (estado == 4) return b.toString(StandardCharsets.ISO_8859_1);
    }
    throw new EOFException("conexão fechada no meio da resposta");
}

// Devolve o status; negativo se o servidor avisou que vai fechar a conexão
static int requisitar(InputStream in, OutputStream out, String caminho, boolean manter) throws IOException {
    out.write(("GET " + caminho + " HTTP/1.1\r\nHost: localhost\r\n" + (manter ? "" : "Connection: close\r\n") + "\r\n")
            .getBytes(StandardCharsets.ISO_8859_1));
    out.flush();
    String cabecalho = lerCabecalho(in);
    int tamanho = 0;
    boolean fechando = false;
    for (String linha : cabecalho.split("\r\n")) {
        if (linha.regionMatches(true, 0, "Content-Length:", 0, 15)) tamanho = Integer.parseInt(linha.substring(15).strip());
        if (linha.equalsIgnoreCase("Connection: close")) fechando = true;
    }
    in.readNBytes(tamanho);
    int status = Integer.parseInt(cabecalho.substring(9, 12));
    return fechando ? -status : status;
}

void main(String[] args) throws Exception {
    int clientes = Integer.parseInt(args[0]), porCliente = Integer.parseInt(args[1]);
    String caminho = args[2];
    boolean manter = args[3].equals("manter");
    AtomicInteger ok = new AtomicInteger(), falhas = new AtomicInteger();
    ConcurrentHashMap<Integer, LongAdder> porStatus = new ConcurrentHashMap<>();
    long t0 = System.nanoTime();
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (int c = 0; c < clientes; c++) {
            executor.submit(() -> {
                Socket s = null;
                InputStream in = null;
                OutputStream out = null;
                try {
                    for (int i = 0; i < porCliente; i++) {
                        if (s == null) {
                            s = new Socket("localhost", 8080);
                            in = new BufferedInputStream(s.getInputStream());
                            out = new BufferedOutputStream(s.getOutputStream());
                        }
                        int status = requisitar(in, out, caminho, manter);
                        if (status < 0) {               // o servidor vai fechar: a próxima usa outra conexão
                            s.close();
                            s = null;
                            status = -status;
                        }
                        porStatus.computeIfAbsent(status, k -> new LongAdder()).increment();
                        ok.incrementAndGet();
                    }
                } catch (IOException e) {
                    falhas.incrementAndGet();
                } finally {
                    if (s != null) try { s.close(); } catch (IOException e) { }
                }
            });
        }
    }
    double segundos = (System.nanoTime() - t0) / 1e9;
    IO.println(String.format("%d clientes x %d requisições (%s): %d respostas %s, %d falhas, %.2f s, %.0f req/s",
            clientes, porCliente, manter ? "keep-alive" : "uma conexão por requisição",
            ok.get(), porStatus, falhas.get(), segundos, ok.get() / segundos));
}

O lerCabecalho é uma pequena máquina de estados que procura a sequência \r\n\r\n, contando quantos caracteres dela já apareceram em seguida. O LongAdder por código de estado, dentro de um ConcurrentHashMap, é a combinação recomendada no exercício 5 de "Ferramentas de Precisão" para contagens disputadas.

A primeira versão deste cliente tinha um defeito instrutivo. Ela ignorava o Connection: close das respostas e continuava enviando requisições na mesma conexão. Na primeira medição, cada cliente falhou exatamente na requisição de número 1001, porque o Farol fecha a conexão depois de mil requisições e avisa o cliente na milésima resposta. Um cliente HTTP correto precisa respeitar o aviso e abrir outra conexão, e é o que o valor negativo devolvido por requisitar faz. O protocolo tem dois lados, e os dois precisam cumprir o contrato.

Todas as medições abaixo rodaram com cliente e servidor na mesma máquina, presos aos mesmos dois núcleos com taskset -c 0,1 e disputando os processadores, e com o Farol registrando cada requisição num arquivo. Os números absolutos dizem pouco; as comparações dizem muito.

Muitas requisições lentas ao mesmo tempo. Cinco mil clientes, cada um pedindo /lento, que leva um segundo:

1000 clientes x 1 requisições (keep-alive): 1000 respostas {200=1000}, 0 falhas, 1.34 s, 749 req/s
5000 clientes x 1 requisições (keep-alive): 5000 respostas {200=5000}, 0 falhas, 1.89 s, 2645 req/s

Cinco mil consultas de um segundo, atendidas em 1,9 segundo. Na versão de "A Primeira Pedra", uma conexão por vez, as mesmas cinco mil levariam cinco mil segundos, uma hora e vinte e três minutos. Com um pool de 200 threads de plataforma, levariam 25 segundos, pela mesma conta de "Um Milhão de Threads": 5000 / 200 = 25 levas de um segundo. Cada uma das cinco mil conexões teve sua própria thread virtual, dormindo dentro do Thread.sleep sem ocupar uma thread do sistema operacional, e o código do manipulador é o mais simples possível, um sleep e um return.

Keep-alive contra uma conexão por requisição. Cinquenta clientes, cada um fazendo duas mil requisições rápidas a /ola/Ana:

50 clientes x 2000 requisições (keep-alive): 100000 respostas {200=100000}, 0 falhas, 5.68 s, 17618 req/s
50 clientes x 2000 requisições (uma conexão por requisição): 100000 respostas {200=100000}, 0 falhas, 8.59 s, 11638 req/s
50 clientes x 2000 requisições (keep-alive): 100000 respostas {200=100000}, 0 falhas, 5.81 s, 17200 req/s
50 clientes x 2000 requisições (uma conexão por requisição): 100000 respostas {200=100000}, 0 falhas, 8.49 s, 11776 req/s

Com conexões persistentes, o Farol atende cerca de 17 mil requisições por segundo; abrindo uma conexão para cada requisição, menos de 12 mil. A diferença, de uma vez e meia, vem do custo de abrir e fechar conexões TCP, de aceitá-las e de criar uma thread para cada uma, que o keep-alive elimina. O exercício 3 detalha de onde vem cada parte.

Esse número depende muito do ambiente. Na mesma máquina, sem prender os processos a dois núcleos (ela tem 16 lógicos), a ordem chegou a se inverter: cerca de 17 mil requisições por segundo com keep-alive e 22 mil sem ele. Aqui cliente e servidor conversam pela interface de loopback, onde abrir uma conexão não paga nenhuma viagem pela rede. Entre máquinas de verdade, cada conexão nova custa pelo menos uma ida e volta a mais antes da primeira requisição, e várias em HTTPS, e é aí que o keep-alive faz a maior diferença. É mais um motivo para medir no ambiente real antes de concluir.

O ataque do cliente calado, com cinco mil atacantes. Um programa abre cinco mil conexões que nunca enviam nada, e, com elas abertas, mede o tempo de resposta de um cliente normal:

5000 conexões caladas abertas; medindo um cliente normal:
  HTTP/1.1 200 OK em 25 ms
depois de 5037 ms, o servidor tinha respondido às caladas: {HTTP/1.1 408 Request Timeout=5000}

O cliente normal foi atendido em 25 milissegundos, com cinco mil conexões hostis abertas ao mesmo tempo. Na versão de "A Primeira Pedra", ele teria esperado cinquenta mil segundos. E depois do prazo ocioso de cinco segundos, cada uma das cinco mil recebeu seu 408 e foi fechada. Um despejo de threads em JSON, com o jcmd de "Um Milhão de Threads", tirado enquanto as conexões estavam abertas, mostra as cinco mil threads virtuais do executor de conexões, cada uma parada no read() do seu socket, dentro da aguardarDados, e nenhuma delas ocupando uma thread do sistema operacional.

Os testes

Os testes do servidor foram reescritos para as conexões persistentes. Uma classe interna Cliente mantém uma conexão aberta e lê uma resposta de cada vez pelo Content-Length, como um cliente HTTP de verdade:

package br.com.exemplo.farol.nucleo;

import br.com.exemplo.farol.http.Resposta;
import br.com.exemplo.farol.rotas.Roteador;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

import java.io.*;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.*;

import static org.assertj.core.api.Assertions.assertThat;

class ServidorTest {
    private static final Configuracao RAPIDA = Configuracao.PADRAO
            .comPrazoOcioso(Duration.ofMillis(300))
            .comPrazoDeDesligamento(Duration.ofSeconds(2));

    private Servidor servidor;

    private Servidor subir(Configuracao configuracao) throws IOException {
        Roteador rotas = new Roteador()
                .get("/", req -> Resposta.texto("oi"))
                .get("/quebra", req -> { throw new IllegalStateException("defeito no manipulador"); })
                .get("/lento", req -> { Thread.sleep(500); return Resposta.texto("fim"); });
        servidor = new Servidor(0, configuracao, rotas).iniciar();
        return servidor;
    }

    @AfterEach
    void descer() throws IOException {
        if (servidor != null) servidor.close();
    }

    /** Uma conexão de teste que envia pedidos crus e lê respostas uma a uma, pelo Content-Length. */
    private final class Cliente implements Closeable {
        final Socket soquete = new Socket("localhost", servidor.porta());
        final InputStream entrada = new BufferedInputStream(soquete.getInputStream());

        Cliente() throws IOException {}

        String pedir(String pedido) throws IOException {
            soquete.getOutputStream().write(pedido.getBytes(StandardCharsets.ISO_8859_1));
            ByteArrayOutputStream cabeca = new ByteArrayOutputStream();
            while (!cabeca.toString(StandardCharsets.ISO_8859_1).endsWith("\r\n\r\n")) {
                int b = entrada.read();
                if (b == -1) return cabeca.toString(StandardCharsets.ISO_8859_1);   // o servidor fechou
                cabeca.write(b);
            }
            String texto = cabeca.toString(StandardCharsets.ISO_8859_1);
            int tamanho = texto.lines().filter(l -> l.startsWith("Content-Length: "))
                    .mapToInt(l -> Integer.parseInt(l.substring(16))).findFirst().orElse(0);
            return texto + new String(entrada.readNBytes(tamanho), StandardCharsets.UTF_8);
        }

        /** true se o servidor fechou a conexão (a leitura chega ao fim). */
        boolean fechadaPeloServidor() throws IOException {
            soquete.setSoTimeout(2000);
            return entrada.read() == -1;
        }

        @Override public void close() throws IOException { soquete.close(); }
    }

    private static final String GET = "GET / HTTP/1.1\r\nHost: x\r\n\r\n";

    @Test
    void atendeVariasRequisicoesNaMesmaConexao() throws IOException {
        subir(RAPIDA);
        try (Cliente c = new Cliente()) {
            for (int i = 0; i < 3; i++) {
                String r = c.pedir(GET);
                assertThat(r).startsWith("HTTP/1.1 200 OK");
                assertThat(r).doesNotContain("Connection: close");
            }
        }
    }

    @Test
    void fechaDepoisDoMaximoDeRequisicoes() throws IOException {
        subir(RAPIDA.comMaximoDeRequisicoesPorConexao(2));
        try (Cliente c = new Cliente()) {
            assertThat(c.pedir(GET)).doesNotContain("Connection: close");
            assertThat(c.pedir(GET)).contains("Connection: close\r\n");
            assertThat(c.fechadaPeloServidor()).isTrue();
        }
    }

    @Test
    void conexaoOciosaEntreRequisicoesFechaEmSilencio() throws IOException {
        subir(RAPIDA);
        try (Cliente c = new Cliente()) {
            c.pedir(GET);
            assertThat(c.entrada.readAllBytes().length).isEqualTo(0);   // fechou sem mandar nada
        }
    }

    @Test
    void falhaNoManipuladorViraErro500EAConexaoSegue() throws IOException {
        subir(RAPIDA);
        try (Cliente c = new Cliente()) {
            String r = c.pedir("GET /quebra HTTP/1.1\r\nHost: x\r\n\r\n");
            assertThat(r).startsWith("HTTP/1.1 500 Internal Server Error");
            assertThat(r).doesNotContain("defeito no manipulador");
            assertThat(c.pedir(GET)).startsWith("HTTP/1.1 200");
        }
    }

    @Test
    void acimaDoLimiteDeConexoesResponde503() throws IOException {
        subir(RAPIDA.comMaximoDeConexoes(1).comPrazoOcioso(Duration.ofSeconds(5)));
        try (Cliente primeiro = new Cliente()) {
            assertThat(primeiro.pedir(GET)).startsWith("HTTP/1.1 200");          // ocupa a única vaga
            try (Cliente segundo = new Cliente()) {
                assertThat(segundo.pedir(GET)).startsWith("HTTP/1.1 503 Service Unavailable");
            }
        }
    }

    @Test
    void desligamentoEsperaARequisicaoEmAndamento() throws Exception {
        subir(RAPIDA);
        try (Cliente c = new Cliente(); var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<String> resposta = executor.submit(() -> c.pedir("GET /lento HTTP/1.1\r\nHost: x\r\n\r\n"));
            Thread.sleep(150);                  // a requisição já está no manipulador
            servidor.close();
            String r = resposta.get(2, TimeUnit.SECONDS);
            assertThat(r).startsWith("HTTP/1.1 200 OK");
            assertThat(r).contains("Connection: close\r\n");
            assertThat(r).endsWith("fim");
        }
    }
}

O arquivo completo tem dez testes: além dos seis acima, o Connection: close do cliente respeitado, o HTTP/1.0 que fecha por padrão, o cliente calado que recebe 408 e o erro de protocolo que fecha a conexão. Os testes de concorrência têm uma armadilha conhecida, a de passar por sorte, porque dependem da ordem em que as threads rodam. O teste do desligamento usa uma pausa de 150 ms para garantir que a requisição já chegou ao manipulador, que dorme 500 ms, uma folga generosa dos dois lados; e toda a suíte foi executada várias vezes seguidas, sem nenhuma falha, antes de ser considerada estável. Com todas as classes de teste:

Tests run: 54, Failures: 0, Skipped: 0

O Farol agora é um servidor de verdade: atende milhares de conexões simultâneas com código sequencial, reaproveita conexões, resiste a clientes calados, limita o que é escasso e desliga sem cortar ninguém. Falta o que torna um servidor útil para um site: servir arquivos.

Fontes e leituras recomendadas

• RFC 9112, "HTTP/1.1", seção 9, "Connection Management": as regras de persistência, o cabeçalho Connection, o encerramento e as recomendações sobre prazos e conexões ociosas. https://www.rfc-editor.org/rfc/rfc9112#section-9
• JEP 444, "Virtual Threads": a seção sobre o modelo de uma thread por requisição, que é exatamente o desenho desta aula. https://openjdk.org/jeps/444
• Documentação do método Runtime.addShutdownHook (Java SE 25): quando os ganchos de desligamento rodam, quando não rodam, e os cuidados ao escrevê-los. https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/lang/Runtime.html#addShutdownHook(java.lang.Thread)
• Documentação da classe Socket (Java SE 25), seção sobre threads virtuais e interrupção: o comportamento de operações bloqueantes de socket em threads virtuais, incluindo o fechamento do socket na interrupção. https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/net/Socket.html
• Robert "RSnake" Hansen, "Slowloris HTTP DoS" (2009): o texto original que apresentou o ataque das conexões lentas, com a explicação de por que servidores de uma thread por conexão com pools limitados eram vulneráveis. Fácil de encontrar pelo título; a página da Wikipédia sobre o Slowloris resume bem o ataque e as defesas. https://en.wikipedia.org/wiki/Slowloris_(computer_security)

Exercícios

Exercício 1

A Conexao fecha a conexão depois de qualquer erro de protocolo, mas mantém a conexão aberta depois de um 500. Construa um exemplo, com bytes, que mostre o que aconteceria se ela mantivesse a conexão aberta depois de um 413.

POST /tarefas HTTP/1.1\r\n
Host: x\r\n
Content-Length: 2000000\r\n
\r\n
<2 milhões de bytes de corpo>

Exercício 2

Por que conexões persistentes seriam impossíveis se o escritor não enviasse Content-Length nas respostas? E por que a resposta a um HEAD envia um Content-Length de um corpo que não vai junto?

Exercício 3

No teste de carga, o keep-alive foi cerca de uma vez e meia mais rápido que uma conexão por requisição. Liste os custos que o keep-alive elimina, e explique por que um cliente que abre dezenas de milhares de conexões por segundo pode esgotar as portas da própria máquina.

Exercício 4

Rodando o Farol com o limite de descritores de arquivo do sistema reduzido para 80 (ulimit -n 80) e abrindo 150 conexões caladas, o registro do servidor passou de um milhão e setecentas mil linhas em nove segundos, e um cliente normal não foi atendido. Explique o que acontece no laço do aceitador e corrija-o.

} catch (IOException e) {
    if (soqueteDoServidor.isClosed()) break;
    // Ex.: "Too many open files". Tentar de novo na hora só gira em falso: espera um pouco
    Registro.erro("falha ao aceitar conexão; nova tentativa em 100 ms", e);
    try {
        Thread.sleep(100);
    } catch (InterruptedException ie) {
        Thread.currentThread().interrupt();
        break;
    }
    continue;
}

Exercício 5

O gancho de desligamento faz o Farol terminar as requisições em andamento quando recebe kill (SIGTERM). O que acontece com kill -9 (SIGKILL)? E por que a ordem dos passos do close() importa: o que daria errado se o executor fosse encerrado antes de o ServerSocket ser fechado?