Comment traiter la liste de liste avec CompletableFuture dans java?

Question

J'ai un List<List<String>> dans java, je veux Liste de processus dans la liste parente Asynchronously avec le groupe de threads fixe Exemple 3. J'essaie d'utiliser CompletableFuture et Stream dans java 8. Je ne comprends pas comment fusionner ces deux et comment procéder . Code PFB j'ai essayé jusqu'ici. Dans le processeur je l'imprime juste, mais je ferai des opérations de DB.

Donc ici j'essaye de diffuser List<List<String>> et de créer le nombre de thread basé sur la taille de liste mais hoe passer la liste en streaming en tant qu'argument au processeur avec CompletableFuture.

public class CompletableFutureWithList { 
    public static void main(String args[]) { 
     List<List<String>> aList = new ArrayList<>(); 
     aList.add(new ArrayList<>(Arrays.asList("xyz", "abc"))); 
     aList.add(new ArrayList<>(Arrays.asList("qwe", "poi"))); 
     System.out.println("helo..."); 
     ExecutorService executor = Executors.newFixedThreadPool(aList.size()); 
     //aList.stream().flatMap(List::stream). 
     Processor aProcessor = new Processor(); 
     List<String> tempList = new ArrayList<>(); 
     CompletableFuture aComFuture = supplyAsync(() -> aProcessor.processList(tempList), executor); 
     try { 
      aComFuture.get(); 
     } catch (InterruptedException | ExecutionException e) { 
      e.printStackTrace(); 
     } 
    } 
} 
public class Processor { 
    public boolean processList(List<String> tempList) { 
     for (String string : tempList) { 
      System.out.println("Output: " + string); 
     } 
     return true; 
    } 
} 

Answer 1

D'après ce que je comprends que vous devez appeler votre processeur pour chacun de vos List<String> dans votre List<List<String>>

Donc ce que vous pouvez faire est de créer toutes les nouvelles discussions à l'aide CompletableFuture puis attendez tous à terminer et faire un traitement des valeurs renvoyées.

Donc ce que vous pouvez faire est quelque chose comme ça

List<List<String>> aList = new ArrayList<>(); 

//Create all CFs 
List<CompletableFuture<Boolean>> futureList = aList.stream() 
      .map(strings -> CompletableFuture.supplyAsync(() -> processList(strings), executor)) 
      .collect(toList()); 

//Wait for them all to complete 
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).join(); 

//Do processing of the results 
Stream<Boolean> booleanStream = futureList.stream() 
      .map(CompletableFuture::join); 
//Do other stuff you need 

Answer 2

voici comment vous pouvez fusionner des listes de liste et completablefuture.

public static void main(String args[]) { 
    List<List<String>> aList = new ArrayList<>(); 
    aList.add(new ArrayList<>(Arrays.asList("xyz", "abc"))); 
    aList.add(new ArrayList<>(Arrays.asList("qwe", "poi"))); 
    System.out.println("hello..."); 

    Processor aProcessor = new Processor(); 
    List<String> tempList = new ArrayList<>(); 
    CompletableFuture aComFuture = CompletableFuture.supplyAsync(() -> ""); 

    aList.stream() 
      .forEach(list -> aComFuture.thenApply(fn -> aProcessor.processList(list))); 

    aComFuture.join(); 
} 

static class Processor { 
    public boolean processList(List<String> tempList) { 
     for (String string : tempList) { 
      System.out.println("Output: " + string); 
     } 
     return true; 
    } 
}