Melhor juntos: Java e a classe Thread. Parte IV — Callable, Future e amigos

Introdução

Na Parte I , revisamos como os encadeamentos são criados. Vamos relembrar mais uma vez. Um thread é representado pela classe Thread, cujo run()método é chamado. Então, vamos usar o compilador Java online Tutorialspoint e executar o seguinte código:

public class HelloWorld {
    
    public static void main(String[] args) {
        Runnable task = () -> {
            System.out.println("Hello World");
        };
        new Thread(task).start();
    }
}

Esta é a única opção para iniciar uma tarefa em um thread?

java.util.concurrent.Callable

Acontece que java.lang.Runnable tem um irmão chamado java.util.concurrent.Callable que veio ao mundo em Java 1.5. Quais são as diferenças? Se você observar atentamente o Javadoc dessa interface, verá que, ao contrário de Runnable, a nova interface declara um call()método que retorna um resultado. Além disso, lança Exception por padrão. Ou seja, evita que tenhamos que try-catchbloquear as exceções verificadas. Nada mal, certo? Agora temos uma nova tarefa em vez de Runnable:

Callable task = () -> {
	return "Hello, World!";
};

Mas o que fazemos com isso? Por que precisamos de uma tarefa em execução em um thread que retorne um resultado? Obviamente, para quaisquer ações realizadas no futuro, esperamos receber o resultado dessas ações no futuro. E temos uma interface com um nome correspondente:java.util.concurrent.Future

java.util.concurrent.Future

A interface java.util.concurrent.Future define uma API para trabalhar com tarefas cujos resultados planejamos receber no futuro: métodos para obter um resultado e métodos para verificar o status. Em relação a Future, estamos interessados ​​em sua implementação na classe java.util.concurrent.FutureTask . Esta é a "Tarefa" que será executada em Future. O que torna essa implementação ainda mais interessante é que ela também implementa Runnable. Você pode considerar isso uma espécie de adaptador entre o antigo modelo de trabalho com tarefas em threads e o novo modelo (novo no sentido de que apareceu no Java 1.5). Aqui está um exemplo:

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class HelloWorld {
    
    public static void main(String[] args) throws Exception {
        Callable task = () -> {
            return "Hello, World!";
        };
        FutureTask<String> future = new FutureTask<>(task);
        new Thread(future).start();
        System.out.println(future.get());
    }
}

Como você pode ver no exemplo, usamos o getmétodo para obter o resultado da tarefa. Observação:quando você obtém o resultado usando o get()método, a execução se torna síncrona! Que mecanismo você acha que será usado aqui? É verdade que não há bloco de sincronização. É por isso que não veremos WAITING no JVisualVM como um monitorou wait, mas como o park()método familiar (porque o LockSupportmecanismo está sendo usado).

Interfaces funcionais

A seguir, falaremos sobre as classes do Java 1.8, portanto faríamos bem em fornecer uma breve introdução. Observe o seguinte código:

Supplier<String> supplier = new Supplier<String>() {
	@Override
	public String get() {
		return "String";
	}
};
Consumer<String> consumer = new Consumer<String>() {
	@Override
	public void accept(String s) {
		System.out.println(s);
	}
};
Function<String, Integer> converter = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return Integer.valueOf(s);
	}
};

Muitos e muitos códigos extras, você não diria? Cada uma das classes declaradas executa uma função, mas usamos um monte de código extra de suporte para defini-la. E é assim que os desenvolvedores Java pensavam. Assim, eles introduziram um conjunto de "interfaces funcionais" ( @FunctionalInterface) e decidiram que agora o próprio Java faria o "pensamento", deixando apenas as coisas importantes para nos preocuparmos:

Supplier<String> supplier = () -> "String";
Consumer<String> consumer = s -> System.out.println(s);
Function<String, Integer> converter = s -> Integer.valueOf(s);

Um Suppliersuprimentos. Não tem parâmetros, mas retorna algo. É assim que ela fornece as coisas. A Consumerconsome. Ele pega algo como uma entrada (um argumento) e faz algo com isso. O argumento é o que ele consome. Então também temos Function. Ele recebe entradas (argumentos), faz algo e retorna algo. Você pode ver que estamos usando ativamente genéricos. Se você não tiver certeza, pode se atualizar lendo " Genéricos em Java: como usar colchetes angulares na prática ".

CompletableFuture

O tempo passou e uma nova classe chamada CompletableFutureapareceu no Java 1.8. Ele implementa a Futureinterface, ou seja, nossas tarefas serão concluídas no futuro, e podemos chamar get()para obter o resultado. Mas também implementa a CompletionStageinterface. O nome já diz tudo: trata-se de uma determinada etapa de algum conjunto de cálculos. Uma breve introdução ao tópico pode ser encontrada na revisão aqui: Introdução ao CompletionStage e ao CompletableFuture. Vamos direto ao ponto. Vejamos a lista de métodos estáticos disponíveis que nos ajudarão a começar: Aqui estão as opções para usá-los:

import java.util.concurrent.CompletableFuture;
public class App {
    public static void main(String[] args) throws Exception {
        // A CompletableFuture that already contains a Result
        CompletableFuture<String> completed;
        completed = CompletableFuture.completedFuture("Just a value");
        // A CompletableFuture that runs a new thread from Runnable. That's why it's Void
        CompletableFuture<Void> voidCompletableFuture;
        voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("run " + Thread.currentThread().getName());
        });
        // A CompletableFuture that starts a new thread whose result we'll get from a Supplier 
        CompletableFuture<String> supplier;
        supplier = CompletableFuture.supplyAsync(() -> {
            System.out.println("supply " + Thread.currentThread().getName());
            return "Value";
        });
    }
}

Se executarmos esse código, veremos que a criação de um CompletableFuturetambém envolve o lançamento de um pipeline inteiro. Portanto, com certa semelhança com a SteamAPI do Java8, é aqui que encontramos a diferença entre essas abordagens. Por exemplo:

List<String> array = Arrays.asList("one", "two");
Stream<String> stringStream = array.stream().map(value -> {
	System.out.println("Executed");
	return value.toUpperCase();
});

Este é um exemplo da API Stream do Java 8. Se você executar este código, verá que "Executado" não será exibido. Em outras palavras, quando um fluxo é criado em Java, o fluxo não inicia imediatamente. Em vez disso, ele espera que alguém queira um valor dele. Mas CompletableFuturecomeça a executar o pipeline imediatamente, sem esperar que alguém peça um valor. Acho que isso é importante entender. Então, nós temos um CompletableFuture. Como podemos fazer um pipeline (ou cadeia) e que mecanismos temos? Lembre-se das interfaces funcionais sobre as quais escrevemos anteriormente.

• Temos um Functionque pega um A e retorna um B. Tem um único método: apply().
• Temos um Consumerque pega um A e não retorna nada (Void). Tem um único método: accept().
• Temos Runnable, que roda na thread, não pega nada e não retorna nada. Tem um único método: run().

A próxima coisa a lembrar é que CompletableFutureusa Runnable, Consumerse Functionsem seu trabalho. Assim, você sempre pode saber que pode fazer o seguinte com CompletableFuture:

public static void main(String[] args) throws Exception {
        AtomicLong longValue = new AtomicLong(0);
        Runnable task = () -> longValue.set(new Date().getTime());
        Function<Long, Date> dateConverter = (longvalue) -> new Date(longvalue);
        Consumer<Date> printer = date -> {
            System.out.println(date);
            System.out.flush();
        };
        // CompletableFuture computation
        CompletableFuture.runAsync(task)
                         .thenApply((v) -> longValue.get())
                         .thenApply(dateConverter)
                         .thenAccept(printer);
}

Os métodos thenRun(), thenApply()e thenAccept()têm versões "Async". Isso significa que esses estágios serão concluídos em um thread diferente. Este thread será retirado de um pool especial — então não saberemos com antecedência se será um thread novo ou antigo. Tudo depende de quão intensivas são as tarefas computacionalmente. Além desses métodos, existem mais três possibilidades interessantes. Para ficar mais claro, vamos imaginar que temos um determinado serviço que recebe algum tipo de mensagem de algum lugar — e isso leva tempo:

public static class NewsService {
	public static String getMessage() {
		try {
			Thread.currentThread().sleep(3000);
			return "Message";
		} catch (InterruptedException e) {
			throw new IllegalStateException(e);
		}
	}
}

Agora, vamos dar uma olhada em outras habilidades que CompletableFuturefornece. Podemos combinar o resultado de um CompletableFuturecom o resultado de outro CompletableFuture:

Supplier newsSupplier = () -> NewsService.getMessage();
        
CompletableFuture<String> reader = CompletableFuture.supplyAsync(newsSupplier);
CompletableFuture.completedFuture("!!")
				 .thenCombine(reader, (a, b) -> b + a)
				 .thenAccept(result -> System.out.println(result))
				 .get();

Observe que as threads são daemon threads por padrão, portanto, para maior clareza, usamos get()para aguardar o resultado. Não apenas podemos combinar CompletableFutures, mas também podemos retornar um CompletableFuture:

CompletableFuture.completedFuture(2L)
				.thenCompose((val) -> CompletableFuture.completedFuture(val + 2))
                               .thenAccept(result -> System.out.println(result));

Aqui, quero observar que o CompletableFuture.completedFuture()método foi usado por brevidade. Este método não cria uma nova thread, então o restante do pipeline será executado na mesma thread onde completedFuturefoi chamado. Existe também um thenAcceptBoth()método. É muito parecido com accept(), mas se thenAccept()aceita um Consumer, thenAcceptBoth()aceita outro CompletableStage+ BiConsumercomo entrada, ou seja, um consumerque leva 2 fontes ao invés de uma. Existe outra habilidade interessante oferecida por métodos cujo nome inclui a palavra "Qualquer um": Esses métodos aceitam uma alternativa CompletableStagee são executados no CompletableStageque deve ser executado primeiro. Por fim, quero encerrar esta revisão com outro recurso interessante de CompletableFuture: tratamento de erros.

CompletableFuture.completedFuture(2L)
				 .thenApply((a) -> {
					throw new IllegalStateException("error");
				 }).thenApply((a) -> 3L)
				 //.exceptionally(ex -> 0L)
				 .thenAccept(val -> System.out.println(val));

Este código não fará nada, porque haverá uma exceção e nada mais acontecerá. Mas ao descomentar a declaração "excepcionalmente", definimos o comportamento esperado. Falando nisso CompletableFuture, também recomendo que você assista ao seguinte vídeo:

• Introdução ao CompletableFuture

Na minha humilde opinião, esses estão entre os vídeos mais explicativos da Internet. Eles devem deixar claro como tudo isso funciona, qual kit de ferramentas temos disponível e por que tudo isso é necessário.

Conclusão

Esperançosamente, agora está claro como você pode usar threads para obter cálculos depois que eles forem concluídos. Material adicional:

• Simultaneidade Java (multithreading) — Explicação do CompletableFuture
• Introdução ao CompletionStage e ao CompletableFuture
• Simultaneidade e Melhorias em Java 8: Parte 2
• Guia para CompletableFuture

Melhor juntos: Java e a classe Thread. Parte I — Threads de execução Melhor juntos: Java e a classe Thread. Parte II — Sincronização melhor juntos: Java e a classe Thread. Parte III — Interação melhor juntos: Java e a classe Thread. Parte V — Executor, ThreadPool, Fork/Join Better juntos: Java e a classe Thread. Parte VI — Atire!