java - 由ListeningExecutorService创建的Future的添加/扩展行为

最终目标是根据Callable/Runnable参数的类型向ListenableFuture添加额外的行为。我想为每个Future方法添加额外的行为。 (示例用例可以在AbstractExecutorService's javadoc和Goetz的Java Concurrency in Practice的7.1.7节中找到)
我有一个覆盖ExecutorService的newTaskFor。它测试参数的类型并创建FutureTask的子类。这自然支持Submit以及invokeAny和invokeAll。
对于ListenableFuture返回的ListeningExecutorService，我如何获得相同的效果？
换句话说，我可以在哪里放置这段代码

if (callable instanceof SomeClass) {
   return new FutureTask<T>(callable) {
        public boolean cancel(boolean mayInterruptIfRunning) {
            System.out.println("Canceling Task");
            return super.cancel(mayInterruptIfRunning);
        }
    };
} else {
    return new FutureTask<T>(callable);
}

这样我的客户就可以执行println语句了

ListeningExecutorService executor = ...;
Collection<Callable> callables = ImmutableSet.of(new SomeClass());
List<Future<?>> futures = executor.invokeAll(callables);
for (Future<?> future : futures) {
    future.cancel(true);
}

解决方案失败
这是我已经尝试过的东西以及为什么它们不起作用的列表。
解决方案A
将MyExecutorService传递给MoreExecutors.listeningDecorator。
问题1:不幸的是，生成的ListeningExecutorService(一个AbstractListeningExecutorService)没有委托(delegate)给ExecutorService方法，而是委托(delegate)给execute(Runnable)上的Executor方法。结果，永远不会调用newTaskFor上的MyExecutorService方法。
问题2:AbstractListeningExecutorService通过无法扩展的静态工厂方法创建了Runnable(a ListenableFutureTask)。
解决方案B
在newTaskFor内，正常创建MyRunnableFuture，然后将其包装为ListenableFutureTask。
问题1:ListenableFutureTask的工厂方法不接受RunnableFuture，它们接受Runnable和Callable。如果我将MyRunnableFuture传递为Runnable，则生成的ListenableFutureTask仅调用run()，而不调用任何Future方法(我的行为在其中)。
问题2:即使它确实调用了Future方法，但MyRunnableFuture并非Callable，因此，当我创建ListenableFutureTask时，我必须提供一个返回值……我没有……因此是Callable。
解决方案C
让MyRunnableFuture扩展ListenableFutureTask而不是FutureTask问题:ListenableFutureTask现在是最终版本(从r10/r11开始)。
解决方案D
让MyRunnableFuture扩展ForwardingListenableFuture并实现RunnableFuture。然后将SomeClass参数包装在ListenableFutureTask中，并从delegate()返回
问题:挂起。我对问题的理解不够充分，无法解释，但是此配置导致FutureTask.Sync中出现死锁。
源代码:根据要求，这是挂起的解决方案D的源代码:

import java.util.*;
import java.util.concurrent.*;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.*;

/** See http://stackoverflow.com/q/8931215/290943 */
public final class MyListeningExecutorServiceD extends ThreadPoolExecutor implements ListeningExecutorService {

    // ===== Test Harness =====

    private static interface SomeInterface {
        public String getName();
    }

    private static class SomeClass implements SomeInterface, Callable<Void>, Runnable {
        private final String name;

        private SomeClass(String name) {
            this.name = name;
        }

        public Void call() throws Exception {
            System.out.println("SomeClass.call");
            return null;
        }

        public void run() {
            System.out.println("SomeClass.run");
        }

        public String getName() {
            return name;
        }
    }

    private static class MyListener implements FutureCallback<Void> {
        public void onSuccess(Void result) {
            System.out.println("MyListener.onSuccess");
        }

        public void onFailure(Throwable t) {
            System.out.println("MyListener.onFailure");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println("Main.start");

        SomeClass someClass = new SomeClass("Main.someClass");

        ListeningExecutorService executor = new MyListeningExecutorServiceD();
        Collection<Callable<Void>> callables = ImmutableSet.<Callable<Void>>of(someClass);
        List<Future<Void>> futures = executor.invokeAll(callables);

        for (Future<Void> future : futures) {
            Futures.addCallback((ListenableFuture<Void>) future, new MyListener());
            future.cancel(true);
        }

        System.out.println("Main.done");
    }

    // ===== Implementation =====

    private static class MyRunnableFutureD<T> extends ForwardingListenableFuture<T> implements RunnableFuture<T> {

        private final ListenableFuture<T> delegate;
        private final SomeInterface someClass;

        private MyRunnableFutureD(SomeInterface someClass, Runnable runnable, T value) {
            assert someClass == runnable;
            this.delegate = ListenableFutureTask.create(runnable, value);
            this.someClass = someClass;
        }

        private MyRunnableFutureD(SomeClass someClass, Callable<T> callable) {
            assert someClass == callable;
            this.delegate = ListenableFutureTask.create(callable);
            this.someClass = someClass;
        }

        @Override
        protected ListenableFuture<T> delegate() {
            return delegate;
        }

        public void run() {
            System.out.println("MyRunnableFuture.run");
            try {
                delegate.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            System.out.println("MyRunnableFuture.cancel " + someClass.getName());
            return super.cancel(mayInterruptIfRunning);
        }
    }

    public MyListeningExecutorServiceD() {
        // Same as Executors.newSingleThreadExecutor for now
        super(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        if (runnable instanceof SomeClass) {
            return new MyRunnableFutureD<T>((SomeClass) runnable, runnable, value);
        } else {
            return new FutureTask<T>(runnable, value);
        }
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof SomeClass) {
            return new MyRunnableFutureD<T>((SomeClass) callable, callable);
        } else {
            return new FutureTask<T>(callable);
        }
    }

    /** Must override to supply co-variant return type */
    @Override
    public ListenableFuture<?> submit(Runnable task) {
        return (ListenableFuture<?>) super.submit(task);
    }

    /** Must override to supply co-variant return type */
    @Override
    public <T> ListenableFuture<T> submit(Runnable task, T result) {
        return (ListenableFuture<T>) super.submit(task, result);
    }

    /** Must override to supply co-variant return type */
    @Override
    public <T> ListenableFuture<T> submit(Callable<T> task) {
        return (ListenableFuture<T>) super.submit(task);
    }
}

最佳答案

根据这个问题以及我最近进行的其他几次讨论，我得出的结论是RunnableFuture/FutureTask本质上是误导性的:显然，您提交了Runnable，显然您还获得了Future，并且显然还有底层Thread的需求一个Runnable。但是，为什么一个类应该同时实现Runnable和Future呢？如果可以，它将替换为哪个Runnable？这已经够糟糕的了，但是随后我们引入了多个级别的执行器，事情真的变得一发不可收拾。

如果这里有解决方案，我认为这需要将FutureTask视为AbstractExecutorService的实现细节。我将重点放在将问题分成两部分:

我想有条件地修改返回的Future。

我想有条件地修改由执行程序服务运行的代码。 (我实际上不确定这里是否有此要求，但如果需要的话我会介绍。即使不是，也可能有助于建立Runnable/Future的区别。)

(Markdown提示)

class MyWrapperExecutor extends ForwardingListeningExecutorService {
  private final ExecutorService delegateExecutor;

  @Override public <T> ListenableFuture<T> submit(Callable<T> task) {
    if (callable instanceof SomeClass) {
      // Modify and submit Callable (or just submit the original Callable):
      ListenableFuture<T> delegateFuture =
          delegateExecutor.submit(new MyCallable(callable));
      // Modify Future:
      return new MyWrapperFuture<T>(delegateFuture);
    } else {
      return delegateExecutor.submit(callable);
    }
  }

  // etc.
}

那行得通吗？

关于java - 由ListeningExecutorService创建的Future的添加/扩展行为，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/8931215/