Mejor juntos: Java y la clase Thread. Parte IV — Llamable, futuro y amigos

Introducción

En la Parte I , revisamos cómo se crean los hilos. Recordemos una vez más. Un subproceso está representado por la clase Subproceso, cuyo run()método se llama. Entonces, usemos el compilador de Java en línea Tutorialspoint y ejecutemos el siguiente código:

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

¿Es esta la única opción para iniciar una tarea en un hilo?

java.util.concurrente.Llamable

Resulta que java.lang.Runnable tiene un hermano llamado java.util.concurrent.Callable que vino al mundo en Java 1.5. ¿Cuáles son las diferencias? Si observa detenidamente el Javadoc para esta interfaz, vemos que, a diferencia de Runnable, la nueva interfaz declara un call()método que devuelve un resultado. Además, lanza Exception por defecto. Es decir, nos evita tener que try-catchbloquear las excepciones comprobadas. No está mal, ¿verdad? Ahora tenemos una nueva tarea en lugar de Runnable:

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

Pero, ¿qué hacemos con él? ¿Por qué necesitamos una tarea ejecutándose en un subproceso que devuelve un resultado? Obviamente, para cualquier acción realizada en el futuro, esperamos recibir el resultado de esas acciones en el futuro. Y tenemos una interfaz con un nombre correspondiente:java.util.concurrent.Future

java.util.concurrente.Futuro

La interfaz java.util.concurrent.Future define una API para trabajar con tareas cuyos resultados planeamos recibir en el futuro: métodos para obtener un resultado y métodos para verificar el estado. En cuanto a Future, estamos interesados ​​en su implementación en la clase java.util.concurrent.FutureTask . Esta es la "Tarea" que se ejecutará en Future. Lo que hace que esta implementación sea aún más interesante es que también implementa Runnable. Puede considerar esto como una especie de adaptador entre el antiguo modelo de trabajo con tareas en subprocesos y el nuevo modelo (nuevo en el sentido de que apareció en Java 1.5). Aquí hay un ejemplo:

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 puede ver en el ejemplo, usamos el getmétodo para obtener el resultado de la tarea. Nota:cuando obtiene el resultado usando el get()método, ¡la ejecución se vuelve sincrónica! ¿Qué mecanismo crees que se utilizará aquí? Es cierto que no hay bloque de sincronización. Es por eso que no veremos WAITING en JVisualVM como monitoro wait, sino como el park()método familiar (porque LockSupportse está utilizando el mecanismo).

Interfaces funcionales

A continuación, hablaremos de las clases de Java 1.8, por lo que haríamos bien en proporcionar una breve introducción. Mira el siguiente 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);
	}
};

Montones y montones de código extra, ¿no crees? Cada una de las clases declaradas realiza una función, pero usamos un montón de código de soporte adicional para definirla. Y así es como pensaron los desarrolladores de Java. En consecuencia, introdujeron un conjunto de "interfaces funcionales" ( @FunctionalInterface) y decidieron que ahora el propio Java haría el "pensamiento", dejando solo las cosas importantes para que nos preocupemos:

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

Un Suppliersuministros. No tiene parámetros, pero devuelve algo. Así es como suministra las cosas. Un Consumerconsume. Toma algo como entrada (un argumento) y hace algo con eso. El argumento es lo que consume. Entonces también tenemos Function. Toma entradas (argumentos), hace algo y devuelve algo. Puede ver que estamos usando genéricos activamente. Si no está seguro, puede refrescarse leyendo " Genéricos en Java: cómo usar corchetes angulares en la práctica ".

CompletableFuturo

Pasó el tiempo y CompletableFutureapareció una nueva clase llamada en Java 1.8. Implementa la Futureinterfaz, es decir, nuestras tareas se completarán en el futuro y podemos llamar get()para obtener el resultado. Pero también implementa la CompletionStageinterfaz. El nombre lo dice todo: esta es una cierta etapa de algún conjunto de cálculos. Puede encontrar una breve introducción al tema en la revisión aquí: Introducción a CompletionStage y CompletableFuture. Vayamos directo al grano. Veamos la lista de métodos estáticos disponibles que nos ayudarán a comenzar: Aquí hay opciones para usarlos:

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";
        });
    }
}

Si ejecutamos este código, veremos que crear CompletableFuturetambién implica lanzar un pipeline completo. Por lo tanto, con cierta similitud con SteamAPI de Java8, aquí es donde encontramos la diferencia entre estos enfoques. Por ejemplo:

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

Este es un ejemplo de Stream API de Java 8. Si ejecuta este código, verá que no se mostrará "Ejecutado". En otras palabras, cuando se crea una secuencia en Java, la secuencia no se inicia inmediatamente. En cambio, espera a que alguien quiera un valor de él. Pero CompletableFuturecomienza a ejecutar la canalización de inmediato, sin esperar a que alguien le solicite un valor. Creo que esto es importante de entender. Entonces, tenemos un CompletableFuture. ¿Cómo podemos hacer un oleoducto (o cadena) y qué mecanismos tenemos? Recuerde esas interfaces funcionales sobre las que escribimos anteriormente.

• Tenemos a Functionque toma una A y devuelve una B. Tiene un solo método: apply().
• Tenemos un Consumerque toma una A y no devuelve nada (Vacío). Tiene un único método: accept().
• Tenemos Runnable, que se ejecuta en el subproceso, no toma nada ni devuelve nada. Tiene un único método: run().

Lo siguiente que debe recordar es que CompletableFutureusa Runnable, Consumersy Functionsen su trabajo. En consecuencia, siempre puede saber que puede hacer lo siguiente con 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);
}

Los métodos thenRun(), thenApply()y thenAccept()tienen versiones "Async". Esto significa que estas etapas se completarán en un subproceso diferente. Este hilo se tomará de un grupo especial, por lo que no sabremos de antemano si será un hilo nuevo o antiguo. Todo depende de cuán intensivas desde el punto de vista computacional sean las tareas. Además de estos métodos, existen tres posibilidades más interesantes. Para mayor claridad, imaginemos que tenemos un determinado servicio que recibe algún tipo de mensaje de algún lugar, y esto lleva tiempo:

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

Ahora, echemos un vistazo a otras habilidades que CompletableFutureproporciona. Podemos combinar el resultado de a CompletableFuturecon el resultado de otro 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();

Tenga en cuenta que los subprocesos son subprocesos demonio de forma predeterminada, por lo que, para mayor claridad, solemos get()esperar el resultado. No solo podemos combinar CompletableFutures, también podemos devolver un CompletableFuture:

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

Aquí quiero señalar que el CompletableFuture.completedFuture()método se utilizó por razones de brevedad. Este método no crea un nuevo subproceso, por lo que el resto de la canalización se ejecutará en el mismo subproceso donde completedFuturese llamó. También hay un thenAcceptBoth()método. Es muy similar a accept(), pero si thenAccept()acepta un Consumer, thenAcceptBoth()acepta otro CompletableStage+ BiConsumercomo entrada, es decir, un consumerque toma 2 fuentes en lugar de una. Hay otra habilidad interesante que ofrecen los métodos cuyo nombre incluye la palabra "Either": estos métodos aceptan una alternativa CompletableStagey se ejecutan en la CompletableStageque se ejecuta primero. Finalmente, quiero terminar esta revisión con otra característica interesante de CompletableFuture: manejo de errores.

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

Este código no hará nada, porque habrá una excepción y no pasará nada más. Pero al descomentar la declaración "excepcionalmente", definimos el comportamiento esperado. Hablando de eso CompletableFuture, también te recomiendo que veas el siguiente video:

• Introducción a CompletableFuture

En mi humilde opinión, estos se encuentran entre los videos más explicativos de Internet. Deben dejar en claro cómo funciona todo esto, qué conjunto de herramientas tenemos disponible y por qué todo esto es necesario.

Conclusión

Con suerte, ahora está claro cómo puede usar subprocesos para obtener cálculos una vez que se completan. Material adicional:

• Simultaneidad de Java (multiproceso): explicación de CompletableFuture
• Introducción a CompletionStage y CompletableFuture
• Concurrencia y mejoras en Java 8: Parte 2
• Guía para el futuro completable

Mejor juntos: Java y la clase Thread. Parte I — Hilos de ejecución Mejor juntos: Java y la clase Thread. Parte II — Sincronización Mejor juntos: Java y la clase Thread. Parte III — Interacción Mejor juntos: Java y la clase Thread. Parte V — Ejecutor, ThreadPool, Fork/Join Mejor juntos: Java y la clase Thread. Parte VI — ¡Dispara!