Encadeando o que Ainda Não Aconteceu — CompletableFuture

Como descrever o que fazer com um resultado antes que ele exista: encadear etapas, combinar consultas paralelas, tratar erros e impor prazos com CompletableFuture, sem deixar threads paradas esperando.

"Não Crie Threads, Entregue Tarefas" terminou com uma limitação do Future: a única forma de usar o resultado é chamar get() e esperar. Quando o resultado de uma consulta alimenta a próxima, e o resultado de três consultas precisa ser combinado, o código se transforma numa sequência de esperas, cada uma com uma thread parada. O exercício do impasse por inanição mostrou até onde isso pode levar. O Java 8, em 2014, trouxe a resposta: CompletableFuture, criado por Doug Lea, um futuro que pode ser encadeado. Em vez de esperar o resultado para depois decidir o que fazer, você registra desde já o que deve acontecer quando ele chegar, e o próprio futuro executa essas etapas no momento certo. É o mesmo modelo das Promises do JavaScript, e é a base de boa parte do código assíncrono escrito em Java na última década.

A primeira cadeia

Um CompletableFuture<T> implementa Future<T>, então tudo da aula anterior continua valendo. A diferença está nos métodos de encadeamento, que recebem as interfaces funcionais de "Comportamento como Valor":

import java.util.concurrent.*;

static void dormir(int ms) {
    try { Thread.sleep(ms); } catch (InterruptedException e) { throw new CompletionException(e); }
}
static String onde() { return Thread.currentThread().getName(); }

void main() {
    try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
        CompletableFuture<Integer> tamanho = CompletableFuture
                .supplyAsync(() -> { dormir(200); IO.println("buscando em " + onde()); return "Ilha do Conhecimento"; }, pool)
                .thenApply(texto -> { IO.println("transformando em " + onde()); return texto.toUpperCase(); })
                .thenApply(String::length);
        IO.println("main segue livre; pronto? " + tamanho.isDone());
        tamanho.thenAccept(n -> IO.println("tamanho: " + n + " (em " + onde() + ")"));
        IO.println("join = " + tamanho.join());

        // Se o futuro já terminou, thenApply roda na thread de quem chama
        CompletableFuture.completedFuture("pronto")
                .thenApply(s -> { IO.println("já pronto, roda em " + onde()); return s; });
    }
}

main segue livre; pronto? false
buscando em pool-1-thread-1
transformando em pool-1-thread-1
tamanho: 20 (em pool-1-thread-1)
join = 20
já pronto, roda em main

O supplyAsync recebe um Supplier, executa-o no pool indicado e devolve um CompletableFuture do resultado. Cada thenApply recebe uma Function e devolve um novo futuro, do resultado transformado, sem esperar nada: a linha que monta a cadeia inteira termina imediatamente, e a main segue livre. O thenAccept recebe um Consumer e é a etapa final típica, que usa o resultado sem produzir outro. Existe ainda o thenRun, que recebe um Runnable e nem olha o resultado.

O join() faz o mesmo que o get(): espera e devolve o resultado. A diferença está nas exceções, assunto da seção sobre erros. Neste exemplo, ele só existe para que a main não termine antes da cadeia; num programa real, o ideal é que a cadeia termine num thenAccept que entrega o resultado a quem precisa, e ninguém espere.

A saída mostra uma regra importante: em qual thread cada etapa roda. As etapas sem o sufixo Async rodam na thread que completou a etapa anterior, aqui a pool-1-thread-1. Mas, se o futuro já estiver completo quando a etapa é registrada, ela roda imediatamente na thread que a registrou, como a última linha mostra com a main. O thenAccept do exemplo poderia ter rodado na main, se a cadeia terminasse antes do registro. Na prática, isso significa que etapas sem Async devem ser rápidas e não bloqueantes, porque podem rodar em qualquer uma das duas threads. Cada método tem uma variante Async (thenApplyAsync, thenAcceptAsync e assim por diante) que envia a etapa a um executor, como uma tarefa nova.

Repare também no dormir, que transforma a InterruptedException numa CompletionException, não verificada. As interfaces Supplier e Function não permitem exceções verificadas, ao contrário do Callable, e esta é uma das pequenas fricções da API. Transformar a interrupção numa exceção que interrompe a cadeia é uma forma honesta de lidar com ela; engoli-la, não.

O executor padrão, e por que evitá-lo

Todos os métodos Async têm uma versão sem o parâmetro de executor. Ela usa o ForkJoinPool.commonPool(), o pool compartilhado pela JVM inteira, o mesmo dos streams paralelos:

IO.println("paralelismo do pool comum: " + ForkJoinPool.commonPool().getParallelism());
IO.println(CompletableFuture.supplyAsync(() -> Thread.currentThread().getName()).join());

paralelismo do pool comum: 1
ForkJoinPool.commonPool-worker-1

Numa máquina de dois núcleos, o pool comum tem uma thread: por padrão, ele usa o número de processadores menos um. Isso faz sentido para cálculos, mas é desastroso para tarefas que esperam por entrada e saída: dez consultas de rede de um segundo, enviadas ao pool comum dessa máquina, levariam dez segundos, e ainda travariam qualquer stream paralelo do programa. Por isso, a regra prática é sempre passar um executor a supplyAsync e runAsync quando a tarefa bloqueia, dimensionado como em "Não Crie Threads, Entregue Tarefas". O pool comum fica para cálculos curtos e puros.

Combinando consultas: thenCompose, thenCombine e allOf

O poder do CompletableFuture aparece quando há várias consultas com dependências entre elas. Imagine montar a página de um produto numa loja: primeiro é preciso buscar o produto, e depois, com ele em mãos, o preço, o estoque e a nota das avaliações, três consultas que dependem do produto mas não umas das outras:

import java.util.*;
import java.util.concurrent.*;

record Produto(int id, String nome) {}
record Pagina(Produto produto, int preco, int estoque, double nota) {}

static void dormir(int ms) {
    try { Thread.sleep(ms); } catch (InterruptedException e) { throw new CompletionException(e); }
}

// Serviços "remotos" simulados
static Produto buscarProduto(int id)   { dormir(200); return new Produto(id, "Caderno"); }
static int buscarPreco(Produto p)      { dormir(300); return 25; }
static int buscarEstoque(Produto p)    { dormir(250); return 12; }
static double buscarNota(Produto p)    { dormir(350); return 4.7; }

static Pagina montarSequencial(int id) {
    Produto p = buscarProduto(id);
    return new Pagina(p, buscarPreco(p), buscarEstoque(p), buscarNota(p));
}

static CompletableFuture<Pagina> montar(int id, Executor pool) {
    return CompletableFuture.supplyAsync(() -> buscarProduto(id), pool)
        .thenCompose(p -> {
            // As três consultas dependem do produto, mas não umas das outras
            var preco = CompletableFuture.supplyAsync(() -> buscarPreco(p), pool);
            var estoque = CompletableFuture.supplyAsync(() -> buscarEstoque(p), pool);
            var nota = CompletableFuture.supplyAsync(() -> buscarNota(p), pool);
            return CompletableFuture.allOf(preco, estoque, nota)
                .thenApply(nada -> new Pagina(p, preco.join(), estoque.join(), nota.join()));
        });
}

void main() {
    long t0 = System.nanoTime();
    IO.println(montarSequencial(7) + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");
    try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
        t0 = System.nanoTime();
        Pagina pagina = montar(7, pool).join();
        IO.println(pagina + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // thenCombine: juntar exatamente dois resultados
        t0 = System.nanoTime();
        Produto caderno = new Produto(7, "Caderno");
        int valorEmEstoque = CompletableFuture.supplyAsync(() -> buscarPreco(caderno), pool)
                .thenCombine(CompletableFuture.supplyAsync(() -> buscarEstoque(caderno), pool),
                             (preco, qtd) -> preco * qtd)
                .join();
        IO.println("valor em estoque: " + valorEmEstoque + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");
    }
}

Pagina[produto=Produto[id=7, nome=Caderno], preco=25, estoque=12, nota=4.7] em 1115 ms
Pagina[produto=Produto[id=7, nome=Caderno], preco=25, estoque=12, nota=4.7] em 556 ms
valor em estoque: 300 em 301 ms

A versão sequencial soma os quatro tempos: 200 + 300 + 250 + 350 = 1100 ms. A versão encadeada leva o tempo do caminho mais longo: 200 ms do produto, mais a mais lenta das três consultas paralelas, 350 ms, total 550 ms. O código descreve as dependências, e o CompletableFuture extrai todo o paralelismo que elas permitem.

Três métodos fazem esse trabalho. O thenCompose é o encadeamento de uma etapa que também é assíncrona: a função recebe o produto e devolve outro CompletableFuture. Se o código usasse thenApply nesse ponto, o resultado seria um CompletableFuture<CompletableFuture<Pagina>>, um futuro de um futuro. O thenCompose "achata" essa estrutura, exatamente como o flatMap de "Pare de Escrever Laços" achata um stream de streams, e como o flatMap de "Um Valor que Pode Não Existir" achata um Optional de Optional. A regra de escolha é a mesma nos três: se a função devolve um valor simples, use thenApply (ou map); se devolve outro contêiner, use thenCompose (ou flatMap).

O allOf recebe vários futuros e devolve um CompletableFuture<Void> que completa quando todos completam. Ele não devolve os resultados, uma limitação da API herdada do fato de que os futuros podem ter tipos diferentes, então a etapa seguinte os lê com join(). Esses join() não bloqueiam: dentro do thenApply depois de um allOf, os três já estão completos, e o join() só entrega o valor. Por fim, o thenCombine junta exatamente dois futuros independentes com uma BiFunction, e é a forma mais limpa quando só dois resultados precisam ser combinados, como preço e estoque no valor total.

Existe também o anyOf, que completa quando o primeiro dos futuros completar, o equivalente ao invokeAny de "Não Crie Threads, Entregue Tarefas", com uma diferença: o anyOf completa com o primeiro resultado, seja ele um sucesso ou uma falha.

Quando algo dá errado

Numa cadeia de etapas, uma exceção em qualquer ponto faz o futuro daquela etapa completar com falha. As etapas seguintes que dependem do resultado não executam, e a falha segue pela cadeia até alguém tratá-la, como uma exceção sobe pela pilha de chamadas até um catch:

import java.util.concurrent.*;

static void dormir(int ms) {
    try { Thread.sleep(ms); } catch (InterruptedException e) { throw new CompletionException(e); }
}

static int buscarNota(boolean falhar) {
    dormir(100);
    if (falhar) throw new IllegalStateException("serviço de avaliações fora do ar");
    return 5;
}

void main() throws InterruptedException {
    try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
        // 1. A falha pula as etapas seguintes até alguém tratá-la
        CompletableFuture<String> texto = CompletableFuture.supplyAsync(() -> buscarNota(true), pool)
                .thenApply(n -> { IO.println("nunca executa"); return n * 2; })
                .thenApply(n -> "nota " + n);
        try {
            texto.join();
        } catch (CompletionException e) {
            IO.println("join:  " + e.getClass().getSimpleName() + " <- " + e.getCause());
        }
        try {
            texto.get();
        } catch (ExecutionException e) {
            IO.println("get:   " + e.getClass().getSimpleName() + " <- " + e.getCause());
        }

        // 2. exceptionally: valor substituto
        int nota = CompletableFuture.supplyAsync(() -> buscarNota(true), pool)
                .exceptionally(erro -> { IO.println("recuperando de: " + erro); return 0; })
                .join();
        IO.println("nota = " + nota);

        // 3. handle: recebe resultado OU erro, e sempre produz um valor
        for (boolean falhar : new boolean[]{false, true}) {
            String r = CompletableFuture.supplyAsync(() -> buscarNota(falhar), pool)
                    .handle((valor, erro) -> erro == null ? "nota " + valor : "sem nota (" + erro.getCause().getMessage() + ")")
                    .join();
            IO.println("handle: " + r);
        }

        // 4. whenComplete: observa sem alterar
        CompletableFuture.supplyAsync(() -> buscarNota(false), pool)
                .whenComplete((v, e) -> IO.println("log: valor=" + v + ", erro=" + e))
                .join();
    }
}

join:  CompletionException <- java.lang.IllegalStateException: serviço de avaliações fora do ar
get:   ExecutionException <- java.lang.IllegalStateException: serviço de avaliações fora do ar
recuperando de: java.util.concurrent.CompletionException: java.lang.IllegalStateException: serviço de avaliações fora do ar
nota = 0
handle: nota 5
handle: sem nota (serviço de avaliações fora do ar)
log: valor=5, erro=null

A mensagem "nunca executa" de fato nunca aparece. O join() e o get() entregam a mesma falha, embrulhada de formas diferentes: o get() lança a ExecutionException verificada do Future, e o join() lança uma CompletionException, não verificada, o que o torna muito mais cômodo dentro de lambdas, onde exceções verificadas não são permitidas. Em ambos os casos, getCause() leva à exceção original.

Três métodos tratam a falha dentro da própria cadeia. O exceptionally só é chamado em caso de falha, e devolve um valor substituto: aqui, uma nota zero no lugar de uma página inteira que falha. O handle é chamado sempre, com o resultado ou com o erro (o outro vem null), e produz um valor em qualquer caso, útil quando a resposta ao sucesso e à falha é construída no mesmo lugar. O whenComplete também é chamado sempre, mas não altera o resultado: a falha ou o valor seguem adiante inalterados, e ele serve para registrar em log ou medir tempo.

Repare, na saída do exceptionally, que o erro recebido não é a IllegalStateException, e sim uma CompletionException que a embrulha. Isso acontece sempre que a exceção é lançada por uma função da cadeia, como aqui o Supplier do supplyAsync, ou propagada de uma etapa anterior. Mas se a falha nasce diretamente no futuro que está sendo tratado, por exemplo com CompletableFuture.failedFuture(erro) ou com completeExceptionally(erro), ela chega sem embrulho:

erro recebido: java.lang.IllegalStateException: direto | causa: null
erro recebido: CompletionException | causa: java.lang.IllegalStateException: via supplyAsync

O erro.getCause().getMessage() do exemplo com handle funcionaria no segundo caso e lançaria um NullPointerException no primeiro. Código robusto desembrulha a exceção com cuidado, como pede o quarto exercício.

Prazos que não esperam para sempre

Uma consulta a um serviço externo pode simplesmente não responder. Desde o Java 9, o CompletableFuture tem dois métodos de prazo:

import java.util.concurrent.*;

static void dormir(int ms) {
    try { Thread.sleep(ms); } catch (InterruptedException e) { throw new CompletionException(e); }
}
static String buscar(String fonte, int ms) { dormir(ms); return "frete de " + fonte; }

void main() {
    long inicio = System.nanoTime();
    try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
        long t0 = System.nanoTime();
        // completeOnTimeout: se não chegar a tempo, usa um valor padrão
        String frete = CompletableFuture.supplyAsync(() -> buscar("transportadora", 2000), pool)
                .completeOnTimeout("frete padrão: R$ 20", 300, TimeUnit.MILLISECONDS)
                .join();
        IO.println(frete + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // orTimeout: se não chegar a tempo, falha com TimeoutException
        t0 = System.nanoTime();
        String r = CompletableFuture.supplyAsync(() -> buscar("correios", 2000), pool)
                .orTimeout(300, TimeUnit.MILLISECONDS)
                .exceptionally(e -> "falhou: " + e.getClass().getSimpleName())
                .join();
        IO.println(r + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // anyOf: o primeiro que terminar
        t0 = System.nanoTime();
        Object primeiro = CompletableFuture.anyOf(
                CompletableFuture.supplyAsync(() -> buscar("A", 400), pool),
                CompletableFuture.supplyAsync(() -> buscar("B", 150), pool)).join();
        IO.println(primeiro + " em " + (System.nanoTime() - t0) / 1_000_000 + " ms");

        // Completar manualmente: a ponte entre um callback e um futuro
        CompletableFuture<String> resposta = new CompletableFuture<>();
        Thread.ofPlatform().start(() -> { dormir(100); resposta.complete("chegou pelo callback"); });
        IO.println(resposta.join());
    }
    IO.println("pool fechado aos " + (System.nanoTime() - inicio) / 1_000_000 + " ms");
}

frete padrão: R$ 20 em 302 ms
falhou: TimeoutException em 304 ms
frete de B em 151 ms
chegou pelo callback
pool fechado aos 2313 ms

O completeOnTimeout completa o futuro com um valor padrão se o prazo acabar antes do resultado, e o orTimeout o completa com uma TimeoutException, que pode ser tratada como qualquer outra falha. O anyOf devolve um CompletableFuture<Object>, porque os futuros podem ter tipos diferentes, e o resultado precisa de uma conversão para ser usado com o tipo certo.

O último bloco mostra que um CompletableFuture pode ser criado vazio e completado à mão, com complete(valor) ou completeExceptionally(erro). É daí que vem o nome da classe, e é a ponte entre o mundo dos futuros e as bibliotecas antigas que avisam o resultado chamando um método seu, um callback: o callback chama complete, e o resto do programa trabalha com o futuro.

Mas a última linha da saída revela algo importante: o pool só fechou aos 2313 ms, e não logo depois dos 300 ms dos prazos. O prazo completou o futuro, mas não parou o trabalho: as duas buscas de 2 segundos continuaram rodando no pool até o fim, ocupando threads, e o close() esperou por elas. O mesmo acontece com o cancelamento:

CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(2000);
        IO.println("a tarefa terminou mesmo assim aos " + (System.nanoTime() - t0) / 1_000_000 + " ms");
    } catch (InterruptedException e) {
        IO.println("interrompida");
    }
    return "resultado";
}, pool);
Thread.sleep(100);   // garante que a tarefa já começou
IO.println("cancel(true) = " + f.cancel(true) + ", cancelado = " + f.isCancelled());

cancel(true) = true, cancelado = true
a tarefa terminou mesmo assim aos 2004 ms

Ao contrário do Future de um ExecutorService, o cancel(true) de um CompletableFuture não interrompe a thread: a documentação avisa que o parâmetro "não tem efeito". O futuro passa a ser considerado cancelado, e quem espera por ele recebe uma CancellationException na hora, mas a tarefa segue até o fim. O motivo é de desenho: um CompletableFuture não sabe qual thread, se alguma, está calculando o seu valor, porque ele pode ser completado por qualquer código. Na prática, isso significa que prazos e cancelamentos protegem quem espera, mas não liberam recursos. Tarefas longas que precisam ser realmente paradas devem verificar um sinal próprio ou usar o Future de um executor, e o modelo de "Tarefas em Família" resolve esse problema de forma estrutural: quando uma tarefa falha ou o prazo acaba, as tarefas irmãs são interrompidas de verdade.

Assíncrono não é sinônimo de rápido

O CompletableFuture dominou o código Java assíncrono por quase dez anos, e bibliotecas inteiras, como o HttpClient do Java 11, devolvem futuros dele. Mas o estilo tem custos que pesam em sistemas grandes. A lógica fica fragmentada em lambdas encadeadas, e um if ou um laço que envolva etapas assíncronas vira uma composição difícil de ler. Uma exceção lançada numa etapa mostra na pilha de chamadas a thread do pool que a executou, e não o código que montou a cadeia, o que torna a depuração penosa. E, como a seção anterior mostrou, cancelar de verdade é difícil.

O motivo de todo esse esforço era não deixar threads de plataforma paradas esperando, porque elas são caras e escassas. A pergunta que o Java 21 fez foi: e se threads paradas não custassem quase nada? A resposta são as threads virtuais, com as quais o código volta a ser escrito de forma sequencial, com get() e join() que bloqueiam à vontade, e ainda assim escala para milhões de tarefas simultâneas. O CompletableFuture continua útil, especialmente para combinar resultados de APIs que já o devolvem, e para os casos em que um fluxo de dados realmente é uma cadeia de transformações, mas deixa de ser a única forma de escrever código concorrente eficiente.

Fontes e leituras recomendadas

• Documentação da classe CompletableFuture (Java SE 25): todas as etapas, as regras de qual thread executa cada uma, o aviso sobre cancel e as políticas de exceção. https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/CompletableFuture.html
• Documentação da interface CompletionStage (Java SE 25): o contrato de encadeamento que CompletableFuture implementa, com a explicação das variantes síncronas e Async. https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/concurrent/CompletionStage.html
• JEP 155, "Concurrency Updates": a proposta que trouxe CompletableFuture, LongAdder e as melhorias do ConcurrentHashMap ao Java 8. https://openjdk.org/jeps/155
• JEP 266, "More Concurrency Updates": a proposta do Java 9 que acrescentou orTimeout, completeOnTimeout, failedFuture e delayedExecutor. https://openjdk.org/jeps/266
• Raoul-Gabriel Urma, Mario Fusco e Alan Mycroft, Modern Java in Action (Manning, 2018), capítulos 15 e 16: uma introdução didática à programação assíncrona com CompletableFuture, com um exemplo de loja muito parecido com o desta aula.

Exercícios

Exercício 1

Explique a diferença entre thenApply e thenCompose e diga qual seria o tipo do resultado se montar usasse thenApply no lugar de thenCompose. Relacione com o flatMap de streams e de Optional.

Exercício 2

Na montagem da página, os tempos das consultas eram 200 ms (produto), 300 ms (preço), 250 ms (estoque) e 350 ms (nota). (a) Explique os tempos de 1115 ms e 556 ms. (b) Qual seria o tempo se o pool tivesse só 2 threads? (c) E se o serviço de nota passasse a exigir o preço como parâmetro?

Exercício 3

Escreva um método genérico <T> CompletableFuture<T> comRetentativa(Supplier<CompletableFuture<T>> acao, int tentativas) que execute a ação e, em caso de falha, tente de novo até o número de tentativas, devolvendo a última falha se todas falharem. Use exceptionallyCompose, do Java 12.

static <T> CompletableFuture<T> comRetentativa(Supplier<CompletableFuture<T>> acao, int tentativas) {
    CompletableFuture<T> primeira = acao.get();
    if (tentativas <= 1) return primeira;
    return primeira.exceptionallyCompose(erro -> {
        IO.println("falhou (" + erro.getCause().getMessage() + "); restam " + (tentativas - 1));
        return comRetentativa(acao, tentativas - 1);
    });
}

Exercício 4

Escreva um método Throwable causaReal(Throwable erro) que devolva a exceção original de uma falha recebida em handle ou exceptionally, funcionando tanto para falhas embrulhadas numa CompletionException quanto para as que chegam diretamente, como as de failedFuture.

static Throwable causaReal(Throwable erro) {
    while ((erro instanceof CompletionException || erro instanceof ExecutionException)
            && erro.getCause() != null) {
        erro = erro.getCause();
    }
    return erro;
}

Exercício 5

No exemplo de prazos, o completeOnTimeout devolveu o frete padrão em 302 ms, mas o pool só fechou aos 2313 ms. Explique por quê, e explique por que o cancel(true) de um CompletableFuture não interrompeu a tarefa. Num servidor que recebe mil pedidos por segundo, todos com essa consulta de frete de 2 segundos e um prazo de 300 ms, qual seria a consequência?