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

Question

最终的目标是要额外添加行为ListenableFuture小号基础上的类型Callable/ Runnable参数.我想为每个Future方法添加额外的行为.(示例用例可以在AbstractExecutorService的javadoc和Goetz的Java Concurrency in Practice的 7.1.7节中找到)

我有一个现有的ExecutorService来覆盖newTaskFor.它测试参数的类型并创建一个子类FutureTask.这自然支持submit以及invokeAny和invokeAll.

我如何获得了同样的效果ListenableFuture S按返回的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)不委托给ExecutorService的方法,其委托给执行(可运行)上的方法执行程序.结果,永远不会调用newTaskForon方法MyExecutorService.

问题2: 通过静态工厂方法AbstractListeningExecutorService创建Runnable(一个ListenableFutureTask),我无法扩展.

解决方案B.

在里面newTaskFor,MyRunnableFuture正常创建,然后用一个包装ListenableFutureTask.

问题1: ListenableFutureTask的工厂方法不接受RunnableFuture,他们接受Runnable和Callable.如果我MyRunnableFuture作为Runnable 传递,结果ListenableFutureTask只是调用run()而不是任何Future方法(我的行为在哪里).

问题2:即使它确实调用了我的Future方法,MyRunnableFuture也不是a Callable,所以当我创建ListenableFutureTask... 时我必须提供一个返回值...我没有...因此Callable.

解决方案C.

让MyRunnableFuture扩展ListenableFutureTask而不是FutureTask

问题: ListenableFutureTask现在是最终的(从r10/r11开始).

解决方案D.

让我们MyRunnableFuture扩展ForwardingListenableFuture并实现RunnableFuture.然后将SomeClass参数包装在a中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);
    }
}

Answer 1

基于这个问题和我最近进行的其他一些讨论，我得出的结论是RunnableFuture/FutureTask本质上具有误导性：显然你提交了Runnable，显然你得到了返回Future，并且显然潜在的Thread需求是Runnable。但为什么一个类应该同时实现RunnableandFuture呢？如果确实如此，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.
}

那行得通吗？