java - 使用 Map of Futures，如何通知()其中的单个元素？

Question

我试图保存一个 Future 的静态列表，并在稍后的时间取消或通知正在进行的 Future。与这些 Future 关联的 Callable 类内部有一个 wait()，因此每个 Future 都必须得到外部源的通知才能继续。但是，我对 notification() 的调用似乎被忽略，因为可调用对象永远不会超过其等待语句。包含 Futures 列表的类看起来像这样:

private static Map <String, Future<Object>> results = new HashMap <String, Future<Object>>();

ExecutorService taskExecutor;

public void doStuff() {
    taskExecutor = Executors.newCachedThreadPool();

    // loop inifinitely - external processes will modify the conditions within
    while(!shutItDown) {

        if (<condition1>) {
            // condition 1 dictates the kick-off of a new callable
            Future<Object> future = taskExecutor.submit(new MyCallable(id));
            results.put(id, future);
        }
        else if (<condition2>) {
            // condition 2 represents a callable in a wait status needs
            // to be notified
            Future<Object> future = results.get(uid);
            if (future != null) {
                synchronized(future) {
                    future.notify();  // this doesn't have the desired effect!
                }
            }
        }
    }

}

Callable 类目前只是一个模型，看起来与此类似:

public class MyCallable implements Callable<Object> {

private String id;

public MyCallable(String id) {
    this.id = id;
}

@Override
public Object call() throws Exception {     

    try {

        // do some work here, then wait on outside notification

        synchronized(this) {
            this.wait();  // never gets past here!!!
        }

        // do some other work here, once this has been notified
    } 
    catch (InterruptedException e) {
        e.printStackTrace();
    }

    return null;
}

调用了notify()方法，但似乎没有效果。 Future 的对象引用看起来有效(即局部变量“future”与静态列表中存储的 future 的引用匹配)。

我可能在这里缺少并发的一些基本概念，但我预计当满足条件2时，我的 Callable 将继续执行 wait() 调用。

请注意，如果我使用cancel()而不是notify()，它会中断我的可运行并导致一个InterruptedException，正如我所期望的。

Answer 1

您需要通知完全相同的同一对象。在您的情况下，您正在通知 Future 对象，但正在等待 MyCallable 对象。不幸的是，我不知道有什么简单的方法可以让您的 MyCallable 对象查看其包装的 Future，因此无需等待 wait()就可以了。

一种解决方案是将锁对象传递到您的 MyCallable 构造函数中，然后将其与关联的 Future 一起保存。像这样的东西:

  private static Map <String, FutureLock> results =
        new HashMap <String, FutureLock>();
  ...
  Object lock = new Object();
  Future<Object> future = taskExecutor.submit(new MyCallable(id, lock));
  results.put(id, new FutureLock(future, lock));
  ...

  public class FutureLock {
      private Future<Object> future;
      private Object lock;
      public FutureLock(Future<Object> future, Object lock) {
         this.future = future;
         this.lock = lock;
      }
      public void notify() {
         synchronized (lock) {
            lock.notify();
         }
      }
      public Object get() throws Exception {
         return future.get();
      }
  }

  public class MyCallable {
     private Object lock;
     public MyCallable(String id, Object lock) {
         this.lock = lock;
         ...
     }
  }