使用期货图谱，我如何通知（）一个单一的元素？

Question

我试图持有期货的静态列表，并在稍后时间取消（）或通知（）正在进行的期货。与这些期货相关的Callable类在其中有一个wait（），所以每一个必须由外部来源通知才能继续。但是，我的notify（）调用似乎被忽略，因为可调用对象从未超过其等待语句。与期货的列表中的类看起来是这样的：

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! 
       } 
      } 
     } 
    } 

} 

可调用的类只是现在一个样机，类似于这样：

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”与存储在静态列表中的未来引用相匹配）。

我可能在这里错过了一些并发性的基本概念，但我期望当condition2满足时，我的Callable将继续执行wait（）调用。

请注意，如果我使用cancel（）而不是notify（），它会中断我的runnable并导致InterruptedException，如我所料。

Answer 1

您需要notify确切相同的对象。在您的情况下，您正在通知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; 
     ... 
    } 
    } 

Answer 2

你似乎什么要努力实现（虽然我可能是错的）是有点像SettableFuture。

不确定您是否需要Callable计算能力到Executor中，但SettableFuture应该像创建它并设置准备就绪，而其他线程坐在那里并等待另一个线程设置。

似乎有两条路线，你正在接近这个问题。

一个与执行者。

Executor e = Executors.newSingleThreadExecutor(); 
Future f = null; 

线程1：

f = e.submit(new Callable(){ 
    public Object call(){ 
    return new Object(); 
    } 
}); 

线程2：

f.get(); //will suspend Thread 2 until the Executor submits the callable 

另一种情况是用SettableFuture

final SettableFuture f = new SettableFuture(); 

线程1：

f.set(new Object()); 

线程2：

f.get(); //will suspend Thread 2 until Thread 1 set's the Future. 

双方将实现相同类型的等待力学的区别是线程1在第一个例子将提交到单独的线程来创建对象。线程2仍然会等到另一个线程完成操作。第二个例子将有线程2等待，直到线程1完成并设置Future。