如何在Java中实现一个非阻塞的未来处理器，以便以后检索处理过的结果

Question

我正在使用一个外部库，其中包含下面的代码。我正在发送很多命令，并在结果中插入统计信息，以检查有多少次调用失败，有多少次成功

public Future<CommandResult> sendCommand(Command command) {
    return command.execute();
}
CommandResult can be success or failure

但是，如果我使用client.sendCommand（命令）. get（）；那么，我正在同步等待结果，同时应用程序被阻止。

我只想稍后再检查（30秒后，哪个调用成功，哪个调用失败）。我保证在10秒内得到答案。问题在于，应用程序等待计算完成，然后检索其结果。

我在思考这种基于答案的方法:

List<Future< CommandResult >> futures = new ArrayList<>();
for(Command command: commands) {
   futures.add(client.sendCommand(command)); 
} 

//in a scheduler, 30+ seconds later 
for (Future<Boolean> future : futures) {  
   saveResult(future.get());
} 

Answer 1

如果将< code>Future实例转换为< code>CompletableFuture(参见Panagiotis Bougioukos的回答)是一个选项，那么您可以实现一个简单的助手函数来转换< code >流

public static <T> CompletableFuture<Stream<T>> collect(Stream<CompletableFuture<T>> futures) {
    return futures
        .map(future -> future.thenApply(Stream::of))
        .reduce(
            CompletableFuture.completedFuture(Stream.empty()),
            (future1, future2) ->
            future1
                .thenCompose(stream1 ->
            future2
                .thenApply(stream2 ->
            concat(stream1, stream2)))
        );
}

从本质上讲，这减少了与流的未来平行的期货流。

如果您在字符串的期货流上使用此选项，它将返回一个期货，该期货在单个期货的最后一个期货完成后即完成：

Stream<CompletableFuture<String>> streamOfFutures = ...
CompletableFuture<Stream<String>> futureOfStream = collect(streamOfFutures);

// Prints a list of strings once the "slowest" future completed
System.out.println(futureOfStream.get().toList());

Answer 2

Future是一个遗留的java功能，它不允许反应式非阻塞功能。CompletableFuture是Java中后来的增强功能，以允许这种反应式非阻塞功能。

您可以基于之前的SO答案尝试将您的未来转换为CompletableFuture，然后您将公开利用非阻塞执行的方法。

检查下面的例子，并进行相应的修改。

public class Application {

    public static void main(String[] args) throws ParseException {

        Future future =  new SquareCalculator().calculate(10);
        CompletableFuture<Integer> completableFuture = makeCompletableFuture(future);
        System.out.println("before apply");
        completableFuture.thenApply(s -> {
            System.out.println(s);
            return s;
        });
        System.out.println("after apply method");
    }


    public static <T> CompletableFuture<T> makeCompletableFuture(Future<T> future) {
        if (future.isDone())
            return transformDoneFuture(future);
        return CompletableFuture.supplyAsync(() -> {
            try {
                if (!future.isDone())
                    awaitFutureIsDoneInForkJoinPool(future);
                return future.get();
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            } catch (InterruptedException e) {
                // Normally, this should never happen inside ForkJoinPool
                Thread.currentThread().interrupt();
                // Add the following statement if the future doesn't have side effects
                // future.cancel(true);
                throw new RuntimeException(e);
            }
        });
    }

    private static <T> CompletableFuture<T> transformDoneFuture(Future<T> future) {
        CompletableFuture<T> cf = new CompletableFuture<>();
        T result;
        try {
            result = future.get();
        } catch (Throwable ex) {
            cf.completeExceptionally(ex);
            return cf;
        }
        cf.complete(result);
        return cf;
    }

    private static void awaitFutureIsDoneInForkJoinPool(Future<?> future)
            throws InterruptedException {
        ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
            @Override public boolean block() throws InterruptedException {
                try {
                    future.get();
                } catch (ExecutionException e) {
                    throw new RuntimeException(e);
                }
                return true;
            }
            @Override public boolean isReleasable() {
                return future.isDone();
            }
        });
    }
}

然后类创建一个示例Future

public class SquareCalculator {

    private ExecutorService executor
            = Executors.newSingleThreadExecutor();

    public Future<Integer> calculate(Integer input) {
        return executor.submit(() -> {
            Thread.sleep(1000);
            return input * input;
        });
    }
}

会导致

Answer 3

我只想稍后再检查（30秒后，哪些调用成功，哪些调用失败）。我保证在10秒内得到答案。问题在于，应用程序等待计算完成，然后检索其结果。

如果您想稍后检查结果，那么您的< code >未来解决方案

如果您想在收到结果时获得结果，我会使用ExecutorCompletionService，您可以随时轮询该服务，查看是否有结果。投票是非阻塞的。

// create your thread pool using fixed or other pool
Executor<Result> threadPool = Executors.newFixedThreadPool(5);
// wrap the Executor in a CompletionService
CompletionService<Boolean> completionService =
   new ExecutorCompletionService<>(e);
// submit jobs to the pool through the ExecutorCompletionService
for (Job job : jobs) {
   completionService.submit(job);
}
// after we have submitted all of the jobs we can shutdown the Executor
// the jobs submitted will continue to run
threadPool.shutdown();
...
// some point later you can do
int jobsRunning = jobs.size();
for (int jobsRunning = jobs.size(); jobsRunning > 0; ) {
   // do some processing ...
   // are any results available?
   Boolean result = completionService.poll();
   if (result != null) {
      // process a result if available
      jobsRunning--;
   }
}

请注意，您需要跟踪您向CompletionService提交了多少作业。