Encontré un problema en el entorno de producción con respecto a CompletableFuture.supplyAsync() Tenemos un método de procesamiento por lotes como el siguiente:
import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; public class CompletableFutureProblem { public void batchOperation(){ List<String> stringList = new ArrayList<>(); stringList.add("task1"); stringList.add("task2"); List<CompletableFuture<String>> futures = new ArrayList<>(); stringList.parallelStream().forEach(str -> { CompletableFuture<String> response = restApiCall(str); futures.add(response); }); //futures.add(null); CompletableFuture<Void> result = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])); CompletableFuture<List<String>> convertedResult = result.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()) ); try { List<String> finishedTask = convertedResult.get(); System.out.println(finishedTask.toString()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } public CompletableFuture<String> restApiCall(String str){ return CompletableFuture.supplyAsync(() -> { return "Complete-" + str; }); } public static void main(String[] args) { CompletableFutureProblem problem = new CompletableFutureProblem(); problem.batchOperation(); } }Cuando todo funcione bien, se imprimirá: [Complete-task2, Complete-task1]
Sin embargo, a veces arroja una excepción como la siguiente en producción:
Exception in thread "main" java.lang.NullPointerException at java.util.concurrent.CompletableFuture.andTree(CompletableFuture.java:1320) at java.util.concurrent.CompletableFuture.allOf(CompletableFuture.java:2238) at third.concurrent.CompletableFutureProblem.batchOperation(CompletableFutureProblem.java:20) at third.concurrent.CompletableFutureProblem.main(CompletableFutureProblem.java:40) Investigué el código fuente de CompletableFuture.allOf() y encontré que si la lista de futuros contiene un valor nulo, por ejemplo, futures.add(null) , se lanzará la excepción, pero realmente no sé en qué escenarios estará CompletableFuture.supplyAsync() en el método restApiCall devuelve null ?
Gracias por su paciente por leer la larga publicación.
varios subprocesos están escribiendo futures , ya que está consumiendo stringList con una secuencia paralela. Sin embargo, futures es un ArrayList , que no es seguro para subprocesos.
Por lo tanto, no puede estar seguro de que cada elemento agregado desde un subproceso diferente será visible sin la sincronización adecuada. Cuando lo transforma en una matriz, habrá problemas de visibilidad de la memoria, que no son deterministas, por lo que a veces funciona como se esperaba.
Para solucionar este problema, normalmente se usaría una colección concurrente. Sin embargo, en este caso, no tiene sentido paralelizar CompletableFuture.supplyAsync() , ya que es una llamada sin bloqueo. Por lo tanto, la mejor solución es simplemente recorrer la lista:
stringList.forEach(str -> { Además, la matriz preasignada en toArray() debe estar vacía:
futures.toArray(new CompletableFuture[0])