作者：N个小灰流_701 | 2023-09-05 09:52
我有很少的异步任务在运行,我需要等到它们中至少有一个完成(将来我可能需要等待N个任务中的util M完成).目前它们被呈现为未来,所以我需要类似的东西
/**
* Blocks current thread until one of specified futures is done and returns it.
*/
public static Future waitForAny(Collection> futures)
throws AllFuturesFailedException
有这样的事吗？或类似的东西,对于Future来说不是必需的.目前我循环收集期货,检查一个是否完成,然后睡一段时间再检查一下.这看起来不是最好的解决方案,因为如果我长时间睡眠会增加不必要的延迟,如果我短时间睡眠则会影响性能.
我可以尝试使用
new CountDownLatch(1)
并在任务完成时减少倒计时并执行
countdown.await()
,但我发现只有控制未来的创造才有可能.这是可能的,但需要重新设计系统,因为当前创建任务的逻辑(将Callable发送到ExecutorService)与决定等待哪个Future分开.我也可以覆盖
RunnableFuture AbstractExecutorService.newTaskFor(Callable callable)
并建立RunnableFuture与附加监听能力的自定义实现时通知任务完成,然后附上这样的监听器所需的任务和使用CountDownLatch,但是这意味着我必须重写newTaskFor每一个ExecutorService的我用 - 可能会有执行它不会扩展AbstractExecutorService.我也可以尝试包装ExecutorService用于同样的目的,但是我必须装饰所有生产Futures的方法.
所有这些解决方案都可行,但看起来非常不自然.看起来我错过了一些简单的东西,比如
WaitHandle.WaitAny(WaitHandle[] waitHandles)
在c#中.针对此类问题,是否有任何众所周知的解决方案？
更新:
最初我根本没有访问Future创建,所以没有优雅的解决方案.重新设计系统后,我得到了进入未来创造并能够countDownLatch.countdown()添加到执行过程中,那么我可以countDownLatch.await(),一切工作正常.谢谢其他答案,我不知道ExecutorCompletionService,并将它的确可以在相似的任务有帮助的,但在这个特殊的情况下,它不能被使用,因为一些期货在没有任何遗嘱执行人创造的 - 实际任务是通过网络发送到另一台服务器,远程完成并收到完成通知.
1> 小智..：
很简单,请查看ExecutorCompletionService.
该类的文档,包括如何在第一个任务完成后取消所有其他任务的示例(如果这是您想要做的)可以在http://java.sun.com/javase/6/docs找到/api/java/util/concurrent/ExecutorCompletionService.html
ExecutorCompletionService不能接受期货,AFAICT这不回答原始问题.
2> ykaganovich..：
ExecutorService.invokeAny
几乎.它需要一个`Collection
`而不是`Collection `
3> Alex Miller..：
为什么不创建结果队列并等待队列？或者更简单地说,使用CompletionService,因为它就是这样:ExecutorService +结果队列.
4> Scott Stanch..：
使用wait()和notifyAll()实际上非常简单.
首先,定义一个锁对象.(你可以使用任何类,但我喜欢明确):
package com.javadude.sample;
public class Lock {}
接下来,定义您的工作线程.他完成处理后必须通知该锁定对象.请注意,通知必须位于锁定对象上的同步块锁定中.
package com.javadude.sample;
public class Worker extends Thread {
private Lock lock_;
private long timeToSleep_;
private String name_;
public Worker(Lock lock, String name, long timeToSleep) {
lock_ = lock;
timeToSleep_ = timeToSleep;
name_ = name;
}
@Override
public void run() {
// do real work -- using a sleep here to simulate work
try {
sleep(timeToSleep_);
} catch (InterruptedException e) {
interrupt();
}
System.out.println(name_ + " is done... notifying");
// notify whoever is waiting, in this case, the client
synchronized (lock_) {
lock_.notify();
}
}
}
最后,您可以编写您的客户端:
package com.javadude.sample;
public class Client {
public static void main(String[] args) {
Lock lock = new Lock();
Worker worker1 = new Worker(lock, "worker1", 15000);
Worker worker2 = new Worker(lock, "worker2", 10000);
Worker worker3 = new Worker(lock, "worker3", 5000);
Worker worker4 = new Worker(lock, "worker4", 20000);
boolean started = false;
int numNotifies = 0;
while (true) {
synchronized (lock) {
try {
if (!started) {
// need to do the start here so we grab the lock, just
// in case one of the threads is fast -- if we had done the
// starts outside the synchronized block, a fast thread could
// get to its notification *before* the client is waiting for it
worker1.start();
worker2.start();
worker3.start();
worker4.start();
started = true;
}
lock.wait();
} catch (InterruptedException e) {
break;
}
numNotifies++;
if (numNotifies == 4) {
break;
}
System.out.println("Notified!");
}
}
System.out.println("Everyone has notified me... I'm done");
}
}
京公网安备 11010802040832号