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

Question

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

我有一个现有的 ExecutorService覆盖 newTaskFor .它测试参数的类型并创建 FutureTask 的子类。这自然支持提交以及 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 上的方法.因此，永远不会调用 MyExecutorService 上的 newTaskFor 方法。

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

方案B

在 newTaskFor 中，正常创建 MyRunnableFuture，然后用 ListenableFutureTask 包装它。

问题 1: ListenableFutureTask的工厂方法不接受 RunnableFuture s，它们接受 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);
    }
}

Answer 1

基于这个问题和我最近进行的其他一些讨论，我得出的结论是 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.
}

这行得通吗？