¿Por qué el scheduleAtFixedRate - scheduleWithFixedDelay métodos no el uso que se puede llamar<V>

Question

Estoy haciendo algunos experimentos acerca de la concurrencia en Java 8

En el ScheduledThreadPoolExecutor API

Puedo ver las dos siguientes firmas:

schedule(Callable<V> callable, long delay, TimeUnit unit)
schedule(Runnable command, long delay, TimeUnit unit)

Uno de Callable y uno de Runnable

Puedo ver en la API de los dos siguientes también:

scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

Mi pregunta es, ¿por qué no existe el dos equivalentes para Callable

scheduleAtFixedRate(Callable<V> callable, long initialDelay, long period, TimeUnit unit)
scheduleWithFixedDelay(Callable<V> callable, long initialDelay, long delay, TimeUnit unit)

Necesito recuperar un resultado booleano de una operación.

Gracias.

Answer 1

¿Qué sería de esperar que el tipo de retorno de scheduleAtFixedRate(Callable<V>) ser?El tipo de retorno de schedule(Callable<V>) es Future<V>, lo que indica que en algún momento en el futuro, el valor de tipo de V devuelto por la que se puede llamar estará disponible.Usted puede esperar a que este valor esté disponible llamando get() en el Futuro.

El tipo de retorno de scheduleAtFixedRate(Callable<V>) puede ser algo como Future<List<V>>, ya que esto significaría que en algún momento en el futuro, todos valores devueltos por las repetidas llamadas a la exigible están disponibles.Sin embargo, siempre habrá más ejecuciones de las que se puede llamar programada, por lo que esta lista no va a haber.

Lo que usted necesita para algo como esto es el concepto de una secuencia asincrónica de los resultados a los que puede suscribirse de manera que se controla cada resultado en cuanto llegue.Hasta donde yo sé, esto no existe en la biblioteca estándar.Una librería de terceros sé de que contiene una cosa es Netflix es RxJava.Utilizando por ejemplo un ReplaySubject de esa biblioteca, puede crear la secuencia de resultados y manejar cada uno de los resultados como se devuelve:

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import rx.subjects.ReplaySubject;

public class Callables {

    public static void main(String[] args) {
        // Example Callable that returns a boolean result
        Random random = new Random();
        Callable<Boolean> callable = () -> random.nextBoolean();

        // Turn the Callable into a Runnable that adds the last result to the stream of results
        ReplaySubject<Boolean> results = ReplaySubject.create();
        Runnable runnable = () -> {
            try {
                boolean result = callable.call();
                results.onNext(result);
            } catch (Exception e) {
                // Needed since Callable can throw an exception, but Runnable cannot
            }
        };

        // Periodically run the Runnable
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
        executor.scheduleAtFixedRate(runnable, 1, 1, TimeUnit.SECONDS);

        // Handling the results as they arrive
        results.forEach(result -> System.out.println("Result: " + result));

        System.out.println("Waiting for results...");
    }

}

Si usted decide investigar el uso de RxJava, puede ser vale la pena usar más de su API en lugar de utilizar un Ejecutor directamente.Puede utilizar Observable.interval para generar un flujo que emite un número periódicamente, luego de este mapa para llamar a su ejecutable.De esta manera se consigue la misma corriente de los resultados de una manera más concisa:

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import rx.Observable;

public class MoreCallables {

    public static void main(String[] args) throws IOException {
        Observable<Long> periodic = Observable.interval(1, TimeUnit.SECONDS);

        Random random = new Random();
        Observable<Boolean> results = periodic.map(i -> random.nextBoolean());

        results.forEach(result -> System.out.println("Result: " + result));

        System.out.println("Waiting for results...");
        System.in.read();
    }

}

Answer 2

Siguiendo a la solución @Andersschuller, la forma más concisa de ella puede parecer:

Observable.interval(1, TimeUnit.SECONDS)
    .map((Long i) -> "tick " + i)
    .forEach(result -> System.out.println("Result: " + result));