java - CompletableFuture 与 PriorityBlockingQueue

Question

我正在尝试编写一个返回 CompletableFuture 的单线程执行器当任务被调度并基于 PriorityBlockingQueue 执行任务时.

我的任务如下所示:

  public interface DriverTask<V> {

    V call(WebDriver driver) throws Throwable;

    default TaskPriority getPriority() {
      return TaskPriority.LOW;
    }
  }

  public enum TaskPriority {
    HIGH,
    MEDIUM,
    LOW
  }

现在我的问题是当我使用 CompletableFuture.supplyAsync 时方法 Executor 只得到一个 Runnable而且我不知道如何让我的 Executor 知道原始任务的优先级。

是否有不同的方式来创建 CompletableFuture以便我根据优先级执行它们？

Answer 1

方法原理如supplyAsync是新建CompletableFuture实例一个设置和异步作业，最终将 complete future 。

您可以对重要的设置执行相同的操作:

private static WebDriver currentDriver() {
    …
}
private static final ExecutorService BACKEND
    = new ThreadPoolExecutor(1, 1, 1, TimeUnit.MINUTES, new PriorityBlockingQueue<>());

public static <V> CompletableFuture<V> runAsync(DriverTask<V> dt) {
    CompletableFuture<V> result = new CompletableFuture<>();
    class Job implements Runnable, Comparable<Job>,
                         CompletableFuture.AsynchronousCompletionTask {
        public void run() {
            try {
                if(!result.isDone()) result.complete(dt.call(currentDriver()));
            }
            catch(Throwable t) { result.completeExceptionally(t); }
        }
        private TaskPriority priority() { return dt.getPriority(); }
        public int compareTo(Job o) { return priority().compareTo(o.priority()); }
    }
    BACKEND.execute(new Job());
    return result;
}

注意实现 CompletableFuture.AsynchronousCompletionTask没有必要;这只是标记那些的约定 Runnable目的是完成一个 CompletableFuture 的实现.

自己实现逻辑的另一个优点是第一阶段不需要将异常包装在 CompletionException 中。 .所以当调用者链接一个 exceptionally ，它将看到原始的、未包装的异常。另外， join的来电者会得到一个 CompletionException反射(reflect) join的代码位置以原始异常为原因调用，携带更多有用的信息。
if(!result.isDone())的目的如果 CompletableFuture，在实际完成尝试之前跳过它在队列中等待时已被取消(或以其他方式完成)。一旦完成尝试开始，取消不会中断它。这是 CompletableFuture的一般行为.