等到Future <T>的任何一个完成

Question

我有很少的异步任务在运行,我需要等到它们中至少有一个完成(将来我可能需要等待N个任务中的util M完成).目前它们被呈现为未来,所以我需要类似的东西

/**
 * Blocks current thread until one of specified futures is done and returns it. 
 */
public static <T> Future<T> waitForAny(Collection<Future<T>> futures) 
        throws AllFuturesFailedException

有这样的事吗？或类似的东西,对于Future来说不是必需的.目前我循环收集期货,检查一个是否完成,然后睡一段时间再检查一下.这看起来不是最好的解决方案,因为如果我长时间睡眠会增加不必要的延迟,如果我短时间睡眠则会影响性能.

我可以尝试使用

new CountDownLatch(1)

并在任务完成时减少倒计时并执行

countdown.await()

,但我发现只有控制未来的创造才有可能.这是可能的,但需要重新设计系统,因为当前创建任务的逻辑(将Callable发送到ExecutorService)与决定等待哪个Future分开.我也可以覆盖

<T> RunnableFuture<T> AbstractExecutorService.newTaskFor(Callable<T> callable)

并建立RunnableFuture与附加监听能力的自定义实现时通知任务完成,然后附上这样的监听器所需的任务和使用CountDownLatch,但是这意味着我必须重写newTaskFor每一个ExecutorService的我用 - 可能会有执行它不会扩展AbstractExecutorService.我也可以尝试包装ExecutorService用于同样的目的,但是我必须装饰所有生产Futures的方法.

所有这些解决方案都可行,但看起来非常不自然.看起来我错过了一些简单的东西,比如

WaitHandle.WaitAny(WaitHandle[] waitHandles)

在c#中.针对此类问题,是否有任何众所周知的解决方案？

更新:

最初我根本没有访问Future创建,所以没有优雅的解决方案.重新设计系统后,我得到了进入未来创造并能够countDownLatch.countdown()添加到执行过程中,那么我可以countDownLatch.await(),一切工作正常.谢谢其他答案,我不知道ExecutorCompletionService,并将它的确可以在相似的任务有帮助的,但在这个特殊的情况下,它不能被使用,因为一些期货在没有任何遗嘱执行人创造的 - 实际任务是通过网络发送到另一台服务器,远程完成并收到完成通知.

Answer 1

很简单,请查看ExecutorCompletionService.

Answer 2

ExecutorService.invokeAny

Answer 3

为什么不创建结果队列并等待队列？或者更简单地说,使用CompletionService,因为它就是这样:ExecutorService +结果队列.

Answer 4

使用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");
    }
}

Answer 5

据我所知，Java没有类似的WaitHandle.WaitAny方法结构。

在我看来，这可以通过“WaitableFuture”装饰器来实现：

public WaitableFuture<T>
    extends Future<T>
{
    private CountDownLatch countDownLatch;

    WaitableFuture(CountDownLatch countDownLatch)
    {
        super();

        this.countDownLatch = countDownLatch;
    }

    void doTask()
    {
        super.doTask();

        this.countDownLatch.countDown();
    }
}

不过，只有在可以将其插入到执行代码之前时，这才有效，否则执行代码将不会具有新doTask()方法。但如果你无法在执行前以某种方式获得对 Future 对象的控制，我真的认为没有办法在不进行轮询的情况下做到这一点。

或者，如果 future 总是在它自己的线程中运行，并且您可以以某种方式获取该线程。然后，您可以生成一个新线程来加入彼此的线程，然后在加入返回后处理等待机制...这将非常难看，并且会产生大量开销。如果某些 Future 对象未完成，则可能会因为死线程而导致大量线程被阻塞。如果您不小心，这可能会泄漏内存和系统资源。

/**
 * Extremely ugly way of implementing WaitHandle.WaitAny for Thread.Join().
 */
public static joinAny(Collection<Thread> threads, int numberToWaitFor)
{
    CountDownLatch countDownLatch = new CountDownLatch(numberToWaitFor);

    foreach(Thread thread in threads)
    {
        (new Thread(new JoinThreadHelper(thread, countDownLatch))).start();
    }

    countDownLatch.await();
}

class JoinThreadHelper
    implements Runnable
{
    Thread thread;
    CountDownLatch countDownLatch;

    JoinThreadHelper(Thread thread, CountDownLatch countDownLatch)
    {
        this.thread = thread;
        this.countDownLatch = countDownLatch;
    }

    void run()
    {
        this.thread.join();
        this.countDownLatch.countDown();
    }
}