guava - 如何将 PriorityBlockingQueue 与 ListeningExecutorService 一起使用？

Question

由于 Guava 的 ListeningExecutorService 是通过包装现有的 ExecutorService 来实现的，因此它通过拦截 execute() 方法来“装饰”任务。这意味着如果我想在底层 ExecutorService 上使用自定义 PriorityQueue，我的比较器将装饰任务“视为”作为 ListenableFutureTask 对象而不是原始对象。

有没有办法掌握它包装的任务？以便队列的比较器可以使用任务权重来确定排序？

Answer 1

我假设您关心 submit()而不是 execute() ? (见我回复的底部。)

带 ListeningExecutorService来自 MoreExecutors.listeningDecorator (你所指的那种包装器)，你运气不好。 listeningDecorator ，最喜欢 ExecutorService实现，将任何输入包装到 submit在 FutureTask .此问题的正常解决方案是实现 AbstractExecutorService并覆盖 newTaskFor返回一个自定义对象。这也应该在这里工作。您基本上会重新实现 listeningDecorator ，这是一个相当简单的包装 AbstractListeningExecutorService ，它本身就是 AbstractExecutorService 的一个相当简单的包装器.

有两个复杂的问题。 (好吧，可能还有更多。我承认我还没有测试过我建议的方法。)

AbstractListeningExecutorService不允许您覆盖 newTaskFor . (为什么？我可以解释你是否愿意 file a feature request 。)因此，你必须扩展 AbstractExecutorService直接，很大程度上复制(短)AbstractListeningExecutorService执行。

newTaskFor必须返回 ListenableFuture那也是Comparable . ListenableFuture 的明显选择是 ListenableFutureTask ，但那个类(class)是 final ，因此您无法创建实例 Comparable .解决办法是创建一个ListenableFutureTask并将其包裹在 SimpleForwardingListenableFuture 中实现 Comparable .

为什么我认为您正在处理 submit()而不是 execute() ?
listeningDecorator(...).execute()不包装输入任务，如我刚刚写的这个测试所示:

public void testListeningDecorator_noWrapExecuteTask() {
  ExecutorService delegate = mock(ExecutorService.class);
  ListeningExecutorService service = listeningDecorator(delegate);
  Runnable task = new Runnable() {
    @Override
    public void run() {}
  };
  service.execute(task);
  verify(delegate).execute(task);
}