Quiero ejecutar N tareas en paralelo de modo que ninguna tarea individual se ejecute durante más de 2 segundos (podemos marcar tales tareas como fallidas). Como salida, quiero devolver la salida de las tareas exitosas y el estado de las tareas fallidas como fallidas. Además, el tiempo de espera de 1 tarea no debería provocar una interrupción del circuito, es decir, la ejecución de otras tareas no debería detenerse.
NOTA: Estoy restringido a usar JAVA 8.
Hice referencia a este artículo para el procesamiento paralelo. Estoy haciendo un tipo similar de procesamiento paralelo como se muestra en el ejemplo de este artículo:
public void parallelProcessing() { try { ExecutorService executorService = Executors.newWorkStealingPool(10); List<CompletableFuture<Integer>> futuresList = new ArrayList<CompletableFuture<Integer>>(); futuresList.add(CompletableFuture.supplyAsync(()->(addFun1(10,5)), executorService)); futuresList.add(CompletableFuture.supplyAsync(()->(subFun1(10,5)), executorService)); futuresList.add(CompletableFuture.supplyAsync(()->(mulFun1(10,5)), executorService)); CompletableFuture<Void> allFutures = CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[futuresList.size()])); CompletableFuture<List<Integer>> allCompletableFuture = allFutures.thenApply(future -> futuresList.stream().map(completableFuture -> completableFuture.join()) .collect(Collectors.toList())); CompletableFuture<List<Integer>> completableFuture = allCompletableFuture.toCompletableFuture(); List<Integer> finalList = (List<Integer>) completableFuture.get(); } catch (Exception ex) { } } public static Integer addFun1(int a, int b) { System.out.println(Thread.currentThread().getName()); for (int i = 0; i < 10; i++) { System.out.print(Thread.currentThread().getName() + i); } return a + b; } public static Integer subFun1(int a, int b) { System.out.println(Thread.currentThread().getName()); for (int i = 0; i < 10; i++) { System.out.print(Thread.currentThread().getName() + i); } return a - b; } public static Integer mulFun1(int a, int b) { System.out.println(Thread.currentThread().getName()); for (int i = 0; i < 10; i++) { System.out.print(Thread.currentThread().getName() + i); } return a * b; }Esto funciona bien. Pero quiero establecer el tiempo de espera para el hilo individual. Sé que puedo usar la función de obtención sobrecargada en la última línea. Pero eso establecería el tiempo de espera para los futuros combinados, ¿verdad? Por ejemplo, si no quiero que ningún subproceso individual se bloquee durante más de 2 segundos, y si configuro un tiempo de espera de 2 segundos en la última línea, será un tiempo de espera combinado, ¿verdad?
get(long timeout, TimeUnit unit)Esto es lo que quiero lograr como resultado final:
Supongamos que hay 5 subprocesos y 4 completos a tiempo, 1 tiempo de espera (debido a la ejecución de más de 2 segundos). En este caso, quiero enviar la salida de 4 subprocesos y enviar un error para el quinto subproceso como resultado. Mi formato de salida de entrada es de la siguiente manera:
Entrada de muestra: List<Input> cada elemento se ejecuta en un subproceso separado, donde cada Entrada tiene un uniqueIdentifier .
Salida de muestra: List<Output> tal que->
Output :{ uniqueIdentifier: //same as input to map for which input this output was generated result: success/fail // this Field I want to add. currently it's not there data: { // from output eg addFun1 , subFun1 } }Supongamos que desea ejecutar 10 subprocesos y desea obtener un valor devuelto, puede usar la interfaz Callable<Boolean> y enviarla a ExecutorService y luego obtener el resultado usando Future#get return Boolean.
Aquí hay un ejemplo de uso.
final int NUM_THREADS=10; List<Boolean> results=new ArrayList<Boolean>(); List<Callable<Boolean>> callables=new ArrayList<Callable<Boolean>>(); for(int i=0;i<NUM_THREADS;++i) { callables.add(new Callable<Boolean>() { public Boolean call() { //add your task here return isTaskCompleted; } }); } ExecutorService executorService=ExecutorService.newFixedThreadPool(NUM_THREADS); //run 10 threads for(Callable<Boolean> callable:callables) { Future<Boolean> future=executor.submit(callable); try { results.add(future.get(2, TimeUnit.SECONDS)); // timeout 2 seconds and add the result } catch(Exception ex) { results.add(false); //set result to false if task throw TimeOutExeption } }Si quieres más información sobre estas clases puedes leer este libro: O'Reilly - Aprendiendo Java Capítulo 9: Subprocesos.
El siguiente es un mre de un solo archivo (pegue el código completo en RunParallelTasks.java y ejecútelo). Es un prototipo de la estructura que sugerí en mi comentario destinado a lograr la funcionalidad requerida utilizando medios simples:
import java.util.Optional; public class RunParallelTasks { public static void main(String[] args) { new Thread(()->{ long duration = 3000; Callback<Long> cb = new LongTask(duration); Output<Long> output = new TaskExecuter<Long>().work(cb); System.out.println( output); }).start(); new Thread(()->{ long duration = 300; Callback<Long> cb = new LongTask(duration); Output<Long> output = new TaskExecuter<Long>().work(cb); System.out.println( output); }).start(); new Thread(()->{ long duration = 4000; Callback<Long> cb = new LongTask(duration); Output<Long> output = new TaskExecuter<Long>().work(cb); System.out.println( output); }).start(); new Thread(()->{ long duration = 1000; Callback<Long> cb = new LongTask(duration); Output<Long> output = new TaskExecuter<Long>().work(cb); System.out.println( output); }).start(); } } class TaskExecuter<T>{ private static final long TIMEOUT = 2000;//millis private T value = null; public Output<T> work(Callback<T> call){ Thread t = new Thread(()->{ value = call.work(); }); t.start(); try { t.join(TIMEOUT); } catch (InterruptedException ex) { ex.printStackTrace(); } return new Output<>(t.getId(), value == null ? Optional.empty() : Optional.of(value)) ; } } interface Callback<T> { T work(); } class LongTask implements Callback<Long>{ private final long durationInMillis; public LongTask(long durationInMillis) { this.durationInMillis = durationInMillis; } @Override public Long work() { try { Thread.sleep(durationInMillis); } catch (InterruptedException ex) { ex.printStackTrace(); } return durationInMillis; } } class Output<T> { private final long id; private boolean success = false; private T data; public Output(long id, Optional<T> op) { this.id = id; if(!op.isEmpty()) { data = op.get(); success = true; } } //todo add getters @Override public String toString() { return "task "+ id+ (success ? " Completed, returned "+data : " Failed" ); } }Teníamos un requisito similar en el que necesitamos capturar el tiempo de espera de cada subproceso e ignorar los resultados. Java 8 no tiene esto incorporado. Una de las formas en que lo logramos,
List<CompletableFuture<?>> futures = new ArrayList<>(); List<?> results = new ArrayList<>(); // It can be anything you collect futures.add(asyncService.fetchMethod() .acceptEither( timeoutAfter(timeout, TimeUnit.SECONDS), results:add) .handle( (result, ex) -> { //Handle the timeout exception results.add(...); return result }); CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join(); private <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) { CompletableFuture<T> result = new CompletableFuture<>(); // We need a separate executor here scheduledExecutor.schedule( () -> result.completeExceptionally(new TimeoutException()), timeout, unit); ); return result; }