如何将 CompletableFuture.supplyAsync 与 PriorityBlockingQueue 一起使用？

Question

我正在尝试通过 CompletableFuture.supplyAsync 将优先级队列添加到使用 ThreadPoolExecutor 和 LinkedBlockingQueue 的现有应用程序。问题是我无法想出一个设计来分配任务优先级，然后我可以在 PriorityBlockingQueue 的比较器中访问它。那是因为我的任务被 CompletableFuture 包装到一个名为 AsyncSupply 的私有内部 class 的实例中，它将原始任务隐藏在私有字段中。 Comparator 然后使用这些转换为 Runnable 的 AsyncSupply 对象被调用，如下所示：

public class PriorityComparator<T extends Runnable> implements Comparator<T> {

    @Override
    public int compare(T o1, T o2) {

        // T is an AsyncSupply object.
        // BUT I WANT SOMETHING I CAN ASSIGN PRIORITIES TOO!
        return 0;
    }
}

我研究了扩展 CompletableFuture 的可能性，因此我可以将它包装在不同的对象中，但是 CompletableFuture 的大部分内容都是封装的并且不可继承。因此，扩展它似乎不是一种选择。也没有用适配器封装它，因为它实现了一个非常广泛的接口。

除了复制整个 CompletableFuture 并修改它之外，我不确定如何解决这个问题。有什么想法吗？

Answer 1

似乎 API 中的一个限制是 CompletableFuture 没有提供使用 PriorityBlockingQueue 的直接方法。幸运的是，我们可以毫不费力地破解它。在 Oracle 的 1.8 JVM 中，它们恰好将所有内部 classes 字段命名为 fn，因此提取我们的优先级感知 Runnables 可以毫不费力地完成：

public class CFRunnableComparator implements Comparator<Runnable> {

    @Override
    @SuppressWarnings("unchecked")
    public int compare(Runnable r1, Runnable r2) {
        // T might be AsyncSupply, UniApply, etc., but we want to
        // compare our original Runnables.
        return ((Comparable) unwrap(r1)).compareTo(unwrap(r2));
    }

    private Object unwrap(Runnable r) {
        try {
            Field field = r.getClass().getDeclaredField("fn");
            field.setAccessible(true);
            // NB: For performance-intensive contexts, you may want to
            // cache these in a ConcurrentHashMap<Class<?>, Field>.
            return field.get(r);
        } catch (IllegalAccessException | NoSuchFieldException e) {
            throw new IllegalArgumentException("Couldn't unwrap " + r, e);
        }
    }
}

这假设您的 Supplier class 是 Comparable，类似于：

public interface WithPriority extends Comparable<WithPriority> {
    int priority();
    @Override
    default int compareTo(WithPriority o) {
        // Reverse comparison so higher priority comes first.
        return Integer.compare(o.priority(), priority());
    }
}

public class PrioritySupplier<T> implements Supplier<T>, WithPriority {
    private final int priority;
    private final Supplier<T> supplier;
    public PrioritySupplier(int priority, Supplier<T> supplier) {
        this.priority = priority;
        this.supplier = supplier;
    }
    @Override
    public T get() {
        return supplier.get();
    }
    @Override
    public int priority() {
        return priority;
    }
}

用法如下：

PriorityBlockingQueue<Runnable> q = new PriorityBlockingQueue<>(11 /*default*/,
        new CFRunnableComparator());
ThreadPoolExecutor pool = new ThreadPoolExecutor(..., q);
CompletableFuture.supplyAsync(new PrioritySupplier<>(n, () -> {
    ...
}), pool);

如果您创建 class 类 PriorityFunction 和 PriorityBiConsumer，您可以使用相同的技术以适当的方式调用 thenApplyAsync 和 whenCompleteAsync 类的方法还有优先事项。

如何将 CompletableFuture.supplyAsync 与 PriorityBlockingQueue 一起使用？

How do I use CompletableFuture.supplyAsync together with PriorityBlockingQueue?

java

executorservice

threadpoolexecutor

java-threads

completable-future