Wie weiter Aufgabe an einen Executor weiterreichen

Question

Ich habe nach einer Lösung einer Situation wie gesucht: Ich habe ein HashSet von Callables und ich übergebe dieses Set an einen Executor für die parallele Ausführung. Jetzt möchte ich, sobald eine eingereichte Aufgabe abgeschlossen ist, sollte ich in der Lage sein, eine neue Callable Executor zuweisen.

Ich habe diesen Code versucht, aber mit diesem, wenn ich executor.invoke verwenden, dann Executor wartet, bis alle Aufgaben abgeschlossen sind, und wenn ich executor.submit verwende, dann Aufgaben nacheinander abgeschlossen werden. Jede Hilfe wäre willkommen.

package poc.threading; 

import java.util.HashSet; 
import java.util.List; 
import java.util.concurrent.Callable; 
import java.util.concurrent.ExecutorService; 
import java.util.concurrent.Executors; 
import java.util.concurrent.Future; 

public class ConcurrencyPoC_UsingExecutor_GetFreeThread { 

    public static void main(String[] args) { 
     // TODO Auto-generated method stub 

     executor(); 
    } 


    public static void executor() 
    { 
     try{ 

      ExecutorService ex = Executors.newCachedThreadPool(); 

      //create a set with all callables task 
      HashSet<Callable<Object>> callables = new HashSet<>(); 
      callables.add(task1()); 
      callables.add(task2()); 
      callables.add(task3()); 
      callables.add(task4()); 

      //executes all task together but executor waits for completion of all tasks 

      List<Future<Object>> fu = ex.invokeAll(callables); 
      for(int i=0; i<fu.size(); i++) 
      { 
       System.out.println(fu.get(i).get() + " , " + Thread.currentThread().getName().toString()); 
      } 

      //executes tasks sequentially 
      for(Callable<Object> task : callables) 
      { 
       Future<Object> future = ex.submit(task); 
       System.out.println(future.get() + " , " + Thread.currentThread().getName().toString()); 
      } 
      ex.shutdownNow(); 
     } 
     catch(Exception e) 
     { 
      e.printStackTrace(); 
     } 
    } 


    public static Callable<Object> task1() throws InterruptedException 
    { 
     return new Callable<Object>() { 

      @Override 
      public Object call() throws Exception { 

       int count = 0; 
       while(count < 3) 
       { 
        System.out.println("****** SLEEP TASK1 ******* "+count); 
        Thread.sleep(500); 
        count ++; 
       } 
       return "Sleep Task Of 500 Completed"; 
      } 
     }; 
    } 


    public static Callable<Object> task2() throws InterruptedException 
    { 
     return new Callable<Object>() { 

      @Override 
      public Object call() throws Exception { 

       int count = 0; 
       while(count < 6) 
       { 
        System.out.println("****** SLEEP TASK2 ******* "+count); 
        Thread.sleep(300); 
        count ++; 
       } 
       return "Sleep Task Of 300 Completed"; 
      } 
     }; 
    } 


    public static Callable<Object> task3() throws InterruptedException 
    { 
     return new Callable<Object>() { 

      @Override 
      public Object call() throws Exception { 

       int count = 0; 
       while(count < 2) 
       { 
        System.out.println("****** SLEEP TASK3 ******* "+count); 
        Thread.sleep(1000); 
        count ++; 
       } 
       return "Sleep Task Of 1000 Completed"; 
      } 
     }; 
    } 

    public static Callable<Object> task4() throws InterruptedException 
    { 
     return new Callable<Object>() { 

      @Override 
      public Object call() throws Exception { 

       int count = 0; 
       while(count < 4) 
       { 
        System.out.println("****** SLEEP TASK4 ******* "+count); 
        Thread.sleep(600); 
        count ++; 
       } 
       return "Sleep Task Of 1000 Completed"; 
      } 
     }; 
    } 
} 

Answer 1

Der Grund Ihre Aufgaben nacheinander im zweiten Beispiel ausgeführt werden, weil Sie get() über die Zukunft fordern, bevor Sie submit() bei zukünftigen Aufgaben aufrufen. Wenn Sie alle Ihre submit vor jeder get s, dann werden sie parallel ausgeführt werden.

Wenn Sie nach Aufgaben suchen, die voneinander abhängen, werfen Sie einen Blick auf die Klasse CompletableFuture. Diese Art von Zukunft wird Ihnen ermöglichen, eine andere Aufgabe zu beginnen, sobald die erste beginnt:

CompletableFuture<Object> task1 = CompletableFuture.supplyAsync(() -> task1(), ex); 
CompletableFuture<Object> task2 = task1.thenApplyAsync(task1Result -> task2(), ex);