Os executors do Java rodam suas tarefas num pool fixo de threads e devolvem um Future. Aprenda submit, invokeAll, cancelamento, cadeias de CompletableFuture, timeouts, semáforos e blocking queues, com saída que nunca varia.
Criar uma thread à mão para cada tarefa funciona num programa pequeno. Num servidor, não, porque cada tarefa paga por uma thread nova e nada impede que uma rajada de trabalho crie milhares delas. O pacote java.util.concurrent oferece executors no lugar disso: um conjunto fixo de threads que pegam tarefas de uma fila e devolvem para você um resultado pelo qual esperar.
Este post cobre ExecutorService, Future, exceções e cancelamento em tarefas, tamanho do pool, CompletableFuture, CountDownLatch, Semaphore e duas coleções concorrentes. Todo programa abaixo rodou no Java 25, e a saída foi colada da execução. Para rodar um deles, salve como Main.java e rode java Main.java. Como na parte sobre threads e estado compartilhado, os programas concorrentes forçam o timing com latches, então imprimem a mesma coisa em toda execução.
Por que executors: um pool de threads e uma fila
Uma thread por tarefa tem dois custos. Cada thread de plataforma é uma thread do sistema operacional com a própria stack, reservada com 1 MB por padrão no Linux de 64 bits, e criar uma dá trabalho de verdade. E o número não tem limite: dez mil requisições chegando ao mesmo tempo significam dez mil threads.
Um executor resolve os dois. Executors.newFixedThreadPool(4) cria quatro threads e fica com elas. As tarefas que você entrega esperam numa fila até uma das quatro ficar livre. Este programa envia dez tarefas que ficam todas bloqueadas num portão, e depois olha dentro do pool:
void main() throws InterruptedException {
var running = new AtomicInteger();
var mostAtOnce = new AtomicInteger();
var fourStarted = new CountDownLatch(4);
var gate = new CountDownLatch(1);
var done = new AtomicInteger();
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
mostAtOnce.accumulateAndGet(running.incrementAndGet(), Math::max);
fourStarted.countDown();
awaitQuietly(gate);
running.decrementAndGet();
done.incrementAndGet();
});
}
fourStarted.await();
var queue = ((ThreadPoolExecutor) pool).getQueue();
IO.println("tasks running: " + running.get());
IO.println("tasks waiting in the queue: " + queue.size());
gate.countDown();
} // close() waits for every task to finish
IO.println("tasks finished: " + done.get());
IO.println("most tasks running at once: " + mostAtOnce.get());
}
void awaitQuietly(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}Ele imprime:
tasks running: 4
tasks waiting in the queue: 6
tasks finished: 10
most tasks running at once: 4Quatro tarefas começaram e ficaram presas no portão. As outras seis ficaram na fila, porque não havia thread livre para rodá-las. Quando main abriu o portão, as quatro threads foram esvaziando a fila, e em nenhum momento mais de quatro tarefas rodaram.
execute entrega ao pool um Runnable e não retorna nada. O cast para ThreadPoolExecutor só está ali para espiar a fila. Você não vai precisar dele em código real.
O bloco try é um try-with-resources. ExecutorService é AutoCloseable desde o Java 19 (verificado: javac --release 18 rejeita este programa), e close() faz o pool parar de aceitar tarefas novas e espera as que estão na fila terminarem. Sem isso, as threads do pool mantêm a JVM viva depois que main retorna.
Um pool fixo de quatro. Quatro tarefas rodam, seis esperam na fila, e quem chamou guarda um Future para cada tarefa que enviou. As threads são criadas uma vez e reaproveitadas para todas as tarefas.
Explicado como se você tivesse dez anos
Um thread pool é a cozinha de um restaurante com um número fixo de cozinheiros e um trilho de pedidos. O restaurante não contrata um cozinheiro novo para cada pedido. Você entrega a comanda, e ela vai para o trilho. Quando um cozinheiro fica livre, ele pega a próxima comanda do trilho e prepara aquele prato.
Em troca da comanda, você recebe um pager. Esse é o Future. Você pode sentar e fazer outra coisa, e o pager acende quando a comida fica pronta. Se preferir ficar de pé no balcão esperando, isso é o get().
Na hora de fechar, o gerente pode parar de aceitar comandas e deixar os cozinheiros terminarem o trilho. Ou pode tirar todas as comandas do trilho e mandar os cozinheiros pararem agora.
A versão precisa
newFixedThreadPool(n) retorna um ThreadPoolExecutor com n threads worker e uma LinkedBlockingQueue de tarefas. Cada worker fica num laço: pega uma tarefa da fila, roda, repete. submit embrulha sua tarefa num FutureTask, que é ao mesmo tempo a coisa que o worker roda e o Future que você segura. Quando a tarefa retorna ou lança uma exceção, o FutureTask guarda o resultado ou a exceção, e qualquer thread bloqueada em get() acorda.
Onde a analogia falha: o trilho de newFixedThreadPool não tem limite de tamanho, então uma enxurrada de tarefas se acumula na memória em vez de ser recusada. O pager também guarda o desfecho, inclusive uma falha, não só um sinal. E um cozinheiro mandado parar pode ignorar a ordem: no Java, parar uma tarefa em execução é um pedido que a tarefa precisa atender, como mostra a seção sobre cancelamento.
submit, invokeAll e invokeAny
submit recebe um Callable que retorna um valor, ou um Runnable, e devolve um Future. invokeAll envia uma lista de tarefas e espera todas. invokeAny espera a primeira que der certo:
void main() throws Exception {
try (var pool = Executors.newFixedThreadPool(3)) {
Future<Integer> length = pool.submit(() -> "executor".length());
IO.println("submit: get() returned " + length.get());
var thirdDone = new CountDownLatch(1);
List<Callable<String>> jobs = List.of(
() -> {
thirdDone.await(); // finish last on purpose
return "first";
},
() -> "second",
() -> {
thirdDone.countDown();
return "third";
});
var results = new ArrayList<String>();
for (Future<String> f : pool.invokeAll(jobs)) {
results.add(f.get());
}
IO.println("invokeAll: " + results);
List<Callable<String>> mirrors = List.of(
() -> { throw new IOException("mirror A is down"); },
() -> "downloaded from mirror B",
() -> { throw new IOException("mirror C is down"); });
IO.println("invokeAny: " + pool.invokeAny(mirrors));
}
}Ele imprime:
submit: get() returned 8
invokeAll: [first, second, third]
invokeAny: downloaded from mirror Bget() bloqueia até a tarefa terminar e depois retorna o valor dela.
A primeira tarefa espera a terceira terminar, então as tarefas terminam fora de ordem. Mesmo assim, invokeAll retorna os futures na ordem em que você enviou as tarefas, então lê-los num laço dá uma ordem fixa, não importa como as threads rodaram.
invokeAny retornou o resultado do mirror B, porque as outras duas tarefas lançaram exceção. Quando uma tarefa dá certo, ele cancela as demais. Se todas falharem, ele lança ExecutionException carregando uma das falhas.
Parando um pool: shutdown e shutdownNow
Um pool para de um de dois jeitos, e eles diferem no que acontece com a fila. Aqui cada pool tem uma thread, uma tarefa rodando bloqueada num portão e três tarefas na fila atrás dela:
void main() throws InterruptedException {
IO.println("shutdown(): " + finishedTasks(false));
IO.println("shutdownNow(): " + finishedTasks(true));
}
String finishedTasks(boolean now) throws InterruptedException {
var pool = Executors.newFixedThreadPool(1);
var started = new CountDownLatch(1);
var gate = new CountDownLatch(1);
var finished = new AtomicInteger();
pool.execute(() -> {
started.countDown();
try {
gate.await();
finished.incrementAndGet();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // interrupted: give up
}
});
for (int i = 0; i < 3; i++) {
pool.execute(finished::incrementAndGet);
}
started.await(); // one task running, three in the queue
int neverStarted = 0;
if (now) {
neverStarted = pool.shutdownNow().size();
} else {
pool.shutdown();
}
String rejected = "";
try {
pool.execute(finished::incrementAndGet);
} catch (RejectedExecutionException e) {
rejected = "new task rejected, ";
}
gate.countDown();
boolean terminated = pool.awaitTermination(5, TimeUnit.SECONDS);
return rejected + "finished " + finished.get() + ", never started " + neverStarted
+ ", terminated " + terminated;
}Ele imprime:
shutdown(): new task rejected, finished 4, never started 0, terminated true
shutdownNow(): new task rejected, finished 0, never started 3, terminated trueOs dois métodos fazem o pool rejeitar tarefas novas com RejectedExecutionException.
shutdown() deixa a tarefa em execução e a fila inteira terminarem, então as quatro terminaram. shutdownNow() tirou as três tarefas da fila, devolveu-as numa lista e interrompeu a que estava rodando. Essa tarefa capturou a interrupção e desistiu, então nada terminou.
awaitTermination espera, até um timeout, as threads do pool acabarem, e retorna se elas acabaram. close() é mais ou menos shutdown() seguido de awaitTermination num laço.
Exceções dentro de tarefas
Uma exceção lançada dentro de uma tarefa não chega na hora ao código que a enviou. Para onde ela vai depende de como você entregou a tarefa. Este programa dá ao pool uma thread factory que instala um handler de exceções não capturadas, para a gente ver o que chega até ele:
void main() throws InterruptedException {
var handlerSaw = new LinkedBlockingQueue<String>();
var threadsMade = new AtomicInteger();
ThreadFactory withHandler = task -> {
threadsMade.incrementAndGet();
var thread = new Thread(task);
thread.setUncaughtExceptionHandler((t, e) -> handlerSaw.add(e.getMessage()));
return thread;
};
try (var pool = Executors.newFixedThreadPool(1, withHandler)) {
Future<Integer> parsed = pool.submit(() -> Integer.parseInt("forty-two"));
try {
parsed.get();
} catch (ExecutionException e) {
IO.println("submit: get() threw " + e.getClass().getSimpleName());
IO.println(" cause: " + e.getCause());
}
pool.execute(() -> Integer.parseInt("seven"));
IO.println("execute: the handler got " + handlerSaw.take());
IO.println(" threads the pool has made: " + threadsMade.get());
Future<?> forgotten = pool.submit(() -> Integer.parseInt("nine"));
while (!forgotten.isDone()) {
Thread.sleep(1);
}
IO.println("submit, no get(): state is " + forgotten.state());
IO.println(" and the handler got " + handlerSaw.poll());
}
}Ele imprime:
submit: get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "forty-two"
execute: the handler got For input string: "seven"
threads the pool has made: 2
submit, no get(): state is FAILED
and the handler got nullCom submit, o Future captura a exceção e a guarda. get() lança ExecutionException, e getCause() é a NumberFormatException original.
Com execute, não existe Future para guardá-la. A exceção escapa da tarefa e mata a thread worker, então o handler de exceções não capturadas da thread a recebe, e o pool cria uma thread substituta. Sem um handler próprio, o handler padrão imprime a stack trace na saída de erro, e main nunca fica sabendo.
O terceiro caso é o que perde erros. Uma tarefa enviada com submit falhou, e ninguém chamou get(). O handler não recebeu nada, e nada foi impresso em lugar nenhum. Future.state(), adicionado no Java 19, mostra FAILED, mas só se alguém perguntar. Quando você enviar uma tarefa, guarde o Future dela e chame get() nele.
Timeouts e cancelamento
get tem uma versão com timeout, e cancel(true) interrompe uma tarefa que já está rodando. Aqui a tarefa espera um latch que nunca abre, então o timeout de 50 ms com certeza vai estourar:
void main() throws Exception {
var neverOpens = new CountDownLatch(1);
var started = new CountDownLatch(1);
var sawInterrupt = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(1)) {
Future<String> slow = pool.submit(() -> {
started.countDown();
try {
neverOpens.await();
return "finished";
} catch (InterruptedException e) {
sawInterrupt.countDown();
throw e;
}
});
started.await();
try {
slow.get(50, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
IO.println("get(50 ms) threw TimeoutException");
}
IO.println("the task is still running: " + !slow.isDone());
boolean cancelled = slow.cancel(true);
sawInterrupt.await();
IO.println("cancel(true) returned " + cancelled);
IO.println("the task saw the interrupt and stopped waiting");
IO.println("state: " + slow.state());
try {
slow.get();
} catch (CancellationException e) {
IO.println("get() now throws CancellationException");
}
}
}Ele imprime:
get(50 ms) threw TimeoutException
the task is still running: true
cancel(true) returned true
the task saw the interrupt and stopped waiting
state: CANCELLED
get() now throws CancellationExceptionTimeoutException só quer dizer que você parou de esperar. A tarefa continuou rodando. Depois, cancel(true) interrompeu a thread dela, await() lançou InterruptedException, e a tarefa acabou. A partir daí, get() lança CancellationException.
cancel(false) não interrompe. Ele ainda marca o future como cancelado, mas uma tarefa que já está rodando segue até o fim. Qualquer um dos dois tipos de cancel impede que uma tarefa na fila chegue a começar.
Interrupção é um pedido
Uma interrupção não para uma thread. Ela liga uma flag na thread, e métodos bloqueantes como await, sleep e BlockingQueue.take percebem a flag e lançam InterruptedException. Quando lançam, eles limpam a flag. Então uma tarefa que captura a exceção e segue em frente apagou o pedido.
Estas duas tarefas são idênticas, exceto por uma linha no catch:
void main() throws InterruptedException {
var neverOpens = new CountDownLatch(1);
var bothStarted = new CountDownLatch(2);
var politeStopped = new CountDownLatch(1);
var rudeIgnoredIt = new CountDownLatch(1);
var pool = Executors.newFixedThreadPool(2, Thread.ofPlatform().daemon().factory());
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // put the flag back
}
}
politeStopped.countDown();
});
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
rudeIgnoredIt.countDown(); // swallowed: the flag stays clear
}
}
});
bothStarted.await();
pool.shutdownNow(); // interrupts both tasks
politeStopped.await();
rudeIgnoredIt.await();
IO.println("the task that restored the flag stopped");
IO.println("the task that swallowed the interrupt went back to waiting");
IO.println("pool terminated within 200 ms: "
+ pool.awaitTermination(200, TimeUnit.MILLISECONDS));
}Ele imprime:
the task that restored the flag stopped
the task that swallowed the interrupt went back to waiting
pool terminated within 200 ms: falseA primeira tarefa religou a flag com Thread.currentThread().interrupt(), então a condição do laço a viu e a tarefa acabou. A segunda engoliu a exceção, a flag continuou limpa, e o laço voltou direto a esperar. shutdownNow mandou uma interrupção e não tem uma segunda para mandar, então o pool não consegue terminar. As threads dele são daemon aqui só para o programa conseguir sair.
A regra: ou deixe InterruptedException se propagar, ou capture e chame Thread.currentThread().interrupt(). Um laço longo sem chamadas bloqueantes deve verificar Thread.currentThread().isInterrupted() por conta própria. O método estático Thread.interrupted() também lê a flag, mas também a limpa, então use-o só quando você for tratar a interrupção ali mesmo.
Quantas threads?
O tamanho certo do pool depende de como as tarefas gastam o tempo, e não existe uma resposta exata para consultar.
Tarefas CPU-bound, como parsing, compressão ou cálculo pesado, precisam de um núcleo o tempo todo. Mais threads que núcleos faz as threads se revezarem, e alternar entre elas custa tempo. Comece perto de Runtime.getRuntime().availableProcessors().
Tarefas IO-bound, como chamadas a um banco de dados ou a outro serviço, passam a maior parte do tempo esperando. Uma thread esperando não usa CPU, então mais threads que núcleos pode ajudar.
A regra prática clássica, do Java Concurrency in Practice, é:
threads = núcleos × (1 + tempo de espera / tempo de cálculo)
Uma tarefa que espera 90 ms e calcula 10 ms numa máquina de 8 núcleos dá 8 × (1 + 9) = 80 threads. Trate isso como um primeiro palpite, não como resposta. O banco de dados pode aceitar só 20 conexões, e aí 80 threads só fazem fila na frente dele. Meça sob carga real e ajuste.
Para trabalho IO-bound, as virtual threads do Java 21 costumam eliminar a pergunta. A próxima parte trata delas.
CompletableFuture: encadeando etapas
Um Future dá a você um jeito de usar o resultado, que é bloquear em get(). Um CompletableFuture deixa você anexar a próxima etapa, e ela roda quando o valor chega:
record User(int id, String name) {}
void main() {
try (var pool = Executors.newFixedThreadPool(4)) {
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> findUser(7), pool);
CompletableFuture<String> name = user.thenApply(User::name);
IO.println("thenApply: " + name.join());
CompletableFuture<CompletableFuture<Integer>> nested =
user.thenApply(u -> cartTotal(u, pool));
CompletableFuture<Integer> total = user.thenCompose(u -> cartTotal(u, pool));
IO.println("thenApply, nested: " + nested.join().join());
IO.println("thenCompose: " + total.join());
CompletableFuture<Integer> shipping = CompletableFuture.supplyAsync(() -> 5, pool);
CompletableFuture<Integer> toPay = total.thenCombine(shipping, Integer::sum);
IO.println("thenCombine: " + toPay.join());
}
}
User findUser(int id) {
return new User(id, "Ana"); // stands in for a database call
}
CompletableFuture<Integer> cartTotal(User u, Executor pool) {
return CompletableFuture.supplyAsync(() -> 37, pool); // stands in for another service
}Ele imprime:
thenApply: Ana
thenApply, nested: 37
thenCompose: 37
thenCombine: 42supplyAsync roda o supplier no executor que você passar. thenApply transforma o valor quando ele fica pronto, como map num stream.
thenApply e thenCompose se diferenciam quando a próxima etapa também retorna um future. Com thenApply, você recebe um future de um future, CompletableFuture<CompletableFuture<Integer>>, e precisa de duas chamadas a join(). thenCompose achata isso num único CompletableFuture<Integer>, como flatMap. thenCombine espera dois futures independentes e junta os valores deles.
join() espera como get(), mas não lança exceções checadas, então cabe dentro de lambdas.
Passe um executor. Sem ele, supplyAsync usa ForkJoinPool.commonPool(). Esse pool é compartilhado com parallel streams e com todo o resto da JVM, e é dimensionado para trabalho de CPU, com uma thread a menos que o número de núcleos da máquina. Chamadas bloqueantes ali podem deixar código sem relação nenhuma sem threads. As threads dele também são daemon (verificado: Thread.currentThread().isDaemon() retornou true dentro de uma), então a JVM pode sair enquanto sua tarefa ainda está rodando.
Esperando vários com allOf
CompletableFuture.allOf retorna um future que completa quando todos os futures que você passou tiverem completado. Ele não carrega valores, então você lê cada um depois. Aqui cada previsão do tempo espera a anterior, então elas terminam na ordem inversa:
void main() {
var finishOrder = new ConcurrentLinkedQueue<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
CompletableFuture<?> nothing = CompletableFuture.completedFuture(null);
// each forecast waits for the one before it, so they finish in reverse
var beijing = forecast("Beijing", nothing, finishOrder, pool);
var madrid = forecast("Madrid", beijing, finishOrder, pool);
var lisbon = forecast("Lisbon", madrid, finishOrder, pool);
CompletableFuture.allOf(lisbon, madrid, beijing).join();
IO.println("finished: " + finishOrder);
for (var f : List.of(lisbon, madrid, beijing)) {
IO.println(f.join());
}
}
}
CompletableFuture<String> forecast(
String city, CompletableFuture<?> after, Queue<String> finishOrder, Executor pool) {
return CompletableFuture.supplyAsync(() -> {
after.join();
finishOrder.add(city);
int celsius = switch (city) {
case "Lisbon" -> 24;
case "Madrid" -> 31;
default -> 28;
};
return city + ": " + celsius + "C";
}, pool);
}Ele imprime:
finished: [Beijing, Madrid, Lisbon]
Lisbon: 24C
Madrid: 31C
Beijing: 28CAs previsões terminaram com Beijing primeiro. Depois de allOf(...).join(), todos os futures estão prontos, então cada join() no laço retorna na hora. O laço lê os futures na ordem que você escolheu, então a ordem da saída é fixa.
Quando uma etapa falha
Um CompletableFuture embrulha uma falha de um jeito diferente conforme a forma como você espera por ela:
void main() throws InterruptedException {
try (var pool = Executors.newFixedThreadPool(2)) {
CompletableFuture<Integer> port =
CompletableFuture.supplyAsync(() -> Integer.parseInt("eighty"), pool);
try {
port.join();
} catch (CompletionException e) {
IO.println("join() threw CompletionException");
IO.println(" cause: " + e.getCause());
}
try {
port.get();
} catch (ExecutionException e) {
IO.println("get() threw ExecutionException");
IO.println(" cause: " + e.getCause());
}
int withDefault = port.exceptionally(ex -> 8080).join();
IO.println("exceptionally: " + withDefault);
String report = port
.thenApply(p -> "listening on " + p)
.handle((value, ex) -> {
if (ex == null) {
return value;
}
Throwable real = ex instanceof CompletionException ? ex.getCause() : ex;
return "handle got " + ex.getClass().getSimpleName()
+ ", real problem: " + real.getMessage();
})
.join();
IO.println(report);
}
}Ele imprime:
join() threw CompletionException
cause: java.lang.NumberFormatException: For input string: "eighty"
get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "eighty"
exceptionally: 8080
handle got CompletionException, real problem: For input string: "eighty"join() embrulha a falha numa CompletionException não checada. get() embrulha numa ExecutionException checada, igual a Future.get(). Nos dois casos, getCause() é a exceção real.
exceptionally só roda em caso de falha e fornece um valor substituto. handle roda nos dois casos e recebe o valor ou a exceção, um deles null.
Isto nos surpreendeu: handle recebeu uma CompletionException, não a NumberFormatException. Uma falha repassada de uma etapa anterior chega embrulhada. Também vimos que exceptionally chamado direto no future de supplyAsync recebeu o embrulho. Então desembrulhe antes de examinar, como o programa faz.
Timeouts num CompletableFuture
orTimeout faz o future falhar depois de um tempo, e completeOnTimeout o completa com um valor de reserva. As duas consultas aqui esperam um latch que nunca abre:
void main() {
var neverOpens = new CountDownLatch(1);
var slowTaskEnded = new CountDownLatch(2);
try (var pool = Executors.newFixedThreadPool(2)) {
Supplier<String> slowLookup = () -> {
try {
neverOpens.await();
return "live price";
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return "interrupted";
} finally {
slowTaskEnded.countDown();
}
};
CompletableFuture<String> strict = CompletableFuture
.supplyAsync(slowLookup, pool)
.orTimeout(50, TimeUnit.MILLISECONDS);
try {
strict.join();
} catch (CompletionException e) {
IO.println("orTimeout: " + e.getCause().getClass().getSimpleName());
}
String price = CompletableFuture
.supplyAsync(slowLookup, pool)
.completeOnTimeout("cached price", 50, TimeUnit.MILLISECONDS)
.join();
IO.println("completeOnTimeout: " + price);
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
neverOpens.countDown(); // let them finish, or close() would wait forever
}
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
}Ele imprime:
orTimeout: TimeoutException
completeOnTimeout: cached price
slow lookups that have ended: 0
slow lookups that have ended: 2orTimeout fez o future falhar com TimeoutException, embrulhada em CompletionException por join(). completeOnTimeout deu o preço em cache.
Nenhum dos dois parou o trabalho. Depois dos dois timeouts, nenhuma consulta lenta tinha acabado: as threads delas continuavam bloqueadas. main teve que abrir o latch, senão close() teria esperado para sempre. cancel(true) também não ajuda. A documentação dele diz que a flag de interrupção não tem efeito, e nós verificamos: a tarefa nunca viu uma interrupção. Quando main abriu o latch, as duas consultas acabaram, que é a última linha. Um timeout num CompletableFuture para a espera, não a tarefa.
Ferramentas de coordenação
java.util.concurrent também tem ferramentas pequenas para fazer threads esperarem umas pelas outras. CountDownLatch, que a parte sobre threads usou, deixa uma thread esperar até outras terminarem alguma coisa:
void main() throws InterruptedException {
var services = List.of("search", "cache", "database");
var allReady = new CountDownLatch(services.size());
var ready = new ConcurrentSkipListSet<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
for (String service : services) {
pool.execute(() -> {
ready.add(service); // stands in for slow start-up work
allReady.countDown();
});
}
allReady.await();
IO.println("every service is up: " + ready);
}
}Ele imprime:
every service is up: [cache, database, search]O latch começa em 3, cada serviço faz a contagem regressiva uma vez, e await() retorna quando ele chega a 0. Um ConcurrentSkipListSet mantém os elementos ordenados, então a ordem impressa é fixa.
Um Semaphore guarda um número de permissões. acquire() pega uma, ou espera se não sobrar nenhuma, e release() devolve. Ele limita quantas threads fazem algo ao mesmo tempo, como chamar um serviço que aceita duas conexões. Aqui seis tarefas dividem duas permissões, e um latch garante que duas estejam mesmo lá dentro juntas:
void main() throws InterruptedException {
var permits = new Semaphore(2);
var inside = new AtomicInteger();
var mostInside = new AtomicInteger();
var twoInside = new CountDownLatch(2);
var gate = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(6)) {
for (int i = 0; i < 6; i++) {
pool.execute(() -> {
try {
permits.acquire();
try {
mostInside.accumulateAndGet(inside.incrementAndGet(), Math::max);
twoInside.countDown();
gate.await();
inside.decrementAndGet();
} finally {
permits.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
twoInside.await();
while (permits.getQueueLength() < 4) {
Thread.sleep(1);
}
IO.println("holding a permit: " + inside.get());
IO.println("waiting for one: " + permits.getQueueLength());
gate.countDown();
}
IO.println("most inside at once: " + mostInside.get());
}Ele imprime:
holding a permit: 2
waiting for one: 4
most inside at once: 2As duas primeiras tarefas pegaram as permissões e esperaram no portão. As outras quatro esperaram em acquire(). O pico é 2. O latch twoInside forçou duas tarefas com permissão lá dentro juntas, e o semáforo deixou uma terceira do lado de fora. Libere no finally, como com um lock, senão uma tarefa que falhou fica com a permissão para sempre.
Um CyclicBarrier faz um número fixo de threads esperarem até todas chegarem ao mesmo ponto, e aí deixa todas seguirem juntas.
Coleções concorrentes, de novo
A parte sobre threads mostrou que uma chamada a ConcurrentHashMap é segura sozinha, mas check-then-act em duas chamadas não é. Os métodos compute colocam a atualização inteira dentro de uma única chamada:
void main() {
var stock = new ConcurrentHashMap<String, Integer>();
stock.put("tea", 20);
stock.put("cake", 3);
var requests = new ConcurrentHashMap<String, Integer>();
var basket = List.of("tea", "cake", "tea", "soup", "cake", "tea");
try (var pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 4; i++) {
pool.execute(() -> {
for (String item : basket) {
requests.merge(item, 1, Integer::sum);
// take one; the last one removes the entry
stock.computeIfPresent(item, (name, left) -> left == 1 ? null : left - 1);
}
});
}
}
IO.println("requests: " + new TreeMap<>(requests));
IO.println("stock: " + new TreeMap<>(stock));
}Ele imprime:
requests: {cake=8, soup=4, tea=12}
stock: {tea=8}Quatro tarefas fizeram 24 pedidos. merge contou todos sem perder nenhuma atualização. computeIfPresent tirou um item de cada vez, e retornar null removeu a entrada, então o bolo esgotou depois de três e sumiu. A sopa nunca esteve em estoque, então a função nunca rodou para ela.
Uma BlockingQueue passa trabalho de uma thread para outra. put espera enquanto a fila está cheia, e take espera enquanto ela está vazia:
void main() throws Exception {
var queue = new ArrayBlockingQueue<String>(2);
var received = new ArrayList<String>();
try (var pool = Executors.newFixedThreadPool(2)) {
Future<?> producer = pool.submit(() -> {
for (int i = 1; i <= 5; i++) {
queue.put("order-" + i); // waits while the queue is full
}
queue.put("DONE");
return null;
});
Future<?> consumer = pool.submit(() -> {
String order = queue.take(); // waits while the queue is empty
while (!order.equals("DONE")) {
received.add(order);
order = queue.take();
}
return null;
});
producer.get();
consumer.get();
}
IO.println("consumer received: " + received);
}Ele imprime:
consumer received: [order-1, order-2, order-3, order-4, order-5]Com capacidade 2, o produtor não consegue passar à frente do consumidor por mais de dois pedidos. O valor "DONE" avisa o consumidor para parar. Com um produtor e um consumidor, os pedidos saem na ordem em que entraram. Com vários de qualquer um dos lados, cada fila continua segura, mas a ordem entre threads não é fixa.
As duas tarefas são lambdas Callable, já que terminam com return null. Isso deixa put e take lançarem InterruptedException para dentro do Future sem um try.
O que lembrar
- Um executor reaproveita um conjunto fixo de threads e põe as tarefas extras na fila. Abra com try-with-resources, para
close()esperar o trabalho terminar. submitretorna umFuture. Chameget()nele, senão a exceção de uma tarefa que falhou se perde em silêncio.invokeAllretorna os futures na ordem de envio.get()lançaExecutionException, ejoin()lançaCompletionException. A exceção real é a causa.- Um timeout para a espera, não a tarefa. Interrupção é um pedido, então religue a flag quando capturar
InterruptedException. - Dimensione um pool para trabalho de CPU perto do número de núcleos. Para trabalho de IO, use a regra prática de espera por cálculo como palpite inicial, e depois meça.
- Dê ao
CompletableFutureo seu próprio executor. UsethenComposequando a próxima etapa retorna um future, eallOfseguido dejoinnuma ordem fixa. - Um
Semaphorelimita quantas threads fazem algo ao mesmo tempo. UmaBlockingQueuepassa trabalho entre threads em ordem.
Entregue as tarefas a um pool, guarde cada future e decida de antemão o que acontece quando uma tarefa falha ou demora demais.