java - CompletableFuture：等待完成百分比

问题描述

我正在将相同的数据并行写入分布式系统的 n 个节点。
当这些节点中的 n% 已成功写入时，其余对其他节点的写入并不重要，因为 n% 保证了其他节点之间的复制。

Java 的 CompletableFuture 似乎与我想要的非常接近，例如：

CompletableFuture.anyOf()

（在第一个未来完成时返回） - 避免不必要的等待，但在我需要 n% 完成时返回太快

CompletableFuture.allOf()

（在所有期货完成时返回）- 避免过早返回但不必要地等待 100% 完成

我正在寻找一种在特定百分比的期货完成时返回的方法。例如，如果我提供 10 个期货，当其中 6% 或 60% 成功完成时返回。

例如，Bluebird JS具有此功能

Promise.some(promises, countThatNeedToComplete)

我想知道我是否可以在 java 中使用 TheadExecutor 或 vanilla CompletableFuture 做类似的事情

标签： javaparallel-processingfuturepercentagecompletable-future

我相信您可以仅使用已提供的内容来实现您想要的内容CompletableFuture，但是您必须实施额外的控制以了解已经完成了多少未来任务，并且当您达到所需的数量/百分比时，取消剩余的任务.

下面是一个类来说明这个想法：

public class CompletableSome<T>
{
private List<CompletableFuture<Void>> tasks;
private int tasksCompleted = 0;

public CompletableSome(List<CompletableFuture<T>> tasks, int percentOfTasksThatMustComplete)
{
    int minTasksThatMustComplete = tasks.size() * percentOfTasksThatMustComplete / 100;
    System.out.println(
        String.format("Need to complete at least %s%% of the %s tasks provided, which means %s tasks.",
            percentOfTasksThatMustComplete, tasks.size(), minTasksThatMustComplete));

    this.tasks = new ArrayList<>(tasks.size());
    for (CompletableFuture<?> task : tasks)
    {
        this.tasks.add(task.thenAccept(a -> {
            // thenAccept will be called right after the future task is completed. At this point we'll
            // check if we reached the minimum number of nodes needed. If we did, then complete the 
            // remaining tasks since they are no longer needed.
            tasksCompleted++;
            if (tasksCompleted >= minTasksThatMustComplete)
            {
                tasks.forEach(t -> t.complete(null));
            }
        }));
    }
}

public void execute()
{
    CompletableFuture.allOf(tasks.toArray(new CompletableFuture<?>[0])).join();
}
}

您将使用此类，如下例所示：

public static void main(String[] args)
{
    int numberOfNodes = 4;

    // Create one future task for each node.
    List<CompletableFuture<String>> nodes = new ArrayList<>();
    for (int i = 1; i <= numberOfNodes; i++)
    {
        String nodeId = "result" + i;
        nodes.add(CompletableFuture.supplyAsync(() -> {
            try
            {
                // Sleep for some time to avoid all tasks to complete before the count is checked.
                Thread.sleep(100 + new Random().nextInt(500));
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }

            // The action here is just to print the nodeId, you would make the actual call here.
            System.out.println(nodeId + " completed.");
            return nodeId;
        }));
    }

    // Here we're saying that just 75% of the nodes must be called successfully.
    CompletableSome<String> tasks = new CompletableSome<>(nodes, 75);
    tasks.execute();
}

请注意，使用此解决方案，您最终可能会执行比最低要求更多的任务 - 例如，当两个或多个节点几乎同时响应时，您可能会在第一个节点响应时达到所需的最低计数，但没有时间取消其他任务。如果这是一个问题，那么您将不得不实施更多控制。

java - CompletableFuture：等待完成百分比

问题描述

解决方案

推荐阅读