55

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

/**
 * 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()

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

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

并创建 RunnableFuture 的自定义实现,能够附加侦听器以在任务完成时收到通知,然后将此类侦听器附加到所需的任务并使用 CountDownLatch,但这意味着我必须为我使用的每个 ExecutorService 覆盖 newTaskFor - 并且可能会有实现它不扩展 AbstractExecutorService。我也可以尝试包装给定的 ExecutorService 用于相同的目的,但是我必须装饰所有产生 Futures 的方法。

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

WaitHandle.WaitAny(WaitHandle[] waitHandles)

在 C# 中。这类问题有没有众所周知的解决方案?

更新:

最初我根本无法访问 Future 创建,因此没有优雅的解决方案。重新设计系统后,我可以访问 Future 创建并能够将 countDownLatch.countdown() 添加到执行过程中,然后我可以 countDownLatch.await() 并且一切正常。感谢其他答案,我不知道 ExecutorCompletionService ,它确实可以在类似的任务中有所帮助,但在这种特殊情况下,它无法使用,因为某些 Futures 是在没有任何执行程序的情况下创建的 - 实际任务通过网络发送到另一台服务器,远程完成并收到完成通知。

4

8 回答 8

55

很简单,看看ExecutorCompletionService

于 2008-09-22T23:42:28.200 回答
9

ExecutorService.invokeAny

于 2008-09-23T03:45:41.597 回答
7

为什么不直接创建一个结果队列并在队列中等待呢?或者更简单地说,使用 CompletionService,因为它就是:ExecutorService + 结果队列。

于 2008-09-23T01:42:18.070 回答
6

使用 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");
    }
}
于 2008-09-22T23:08:26.583 回答
4

据我所知,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 对象没有完成,您可能会有很多阻塞线程,具体取决于死线程。如果您不小心,这可能会泄漏内存和系统资源。

/**
 * 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();
    }
}
于 2008-09-22T21:18:03.853 回答
1

如果您可以使用CompletableFutures 代替,那么您可以使用CompletableFuture.anyOf它,只需在结果上调用 join :

CompletableFuture.anyOf(futures).join()

您可以CompletableFuture通过调用CompletableFuture.supplyAsyncorrunAsync方法将 s 与执行程序一起使用。

于 2020-10-28T08:03:18.880 回答
0

既然您不在乎哪个完成,为什么不为所有线程设置一个 WaitHandle 并等待呢?谁先完成就可以设置手柄。

于 2008-09-22T21:18:44.957 回答
-1

看到这个选项:

public class WaitForAnyRedux {

private static final int POOL_SIZE = 10;

public static <T> T waitForAny(Collection<T> collection) throws InterruptedException, ExecutionException {

    List<Callable<T>> callables = new ArrayList<Callable<T>>();
    for (final T t : collection) {
        Callable<T> callable = Executors.callable(new Thread() {

            @Override
            public void run() {
                synchronized (t) {
                    try {
                        t.wait();
                    } catch (InterruptedException e) {
                    }
                }
            }
        }, t);
        callables.add(callable);
    }

    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(POOL_SIZE);
    ExecutorService executorService = new ThreadPoolExecutor(POOL_SIZE, POOL_SIZE, 0, TimeUnit.SECONDS, queue);
    return executorService.invokeAny(callables);
}

static public void main(String[] args) throws InterruptedException, ExecutionException {

    final List<Integer> integers = new ArrayList<Integer>();
    for (int i = 0; i < POOL_SIZE; i++) {
        integers.add(i);
    }

    (new Thread() {
        public void run() {
            Integer notified = null;
            try {
                notified = waitForAny(integers);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
            System.out.println("notified=" + notified);
        }

    }).start();


    synchronized (integers) {
        integers.wait(3000);
    }


    Integer randomInt = integers.get((new Random()).nextInt(POOL_SIZE));
    System.out.println("Waking up " + randomInt);
    synchronized (randomInt) {
        randomInt.notify();
    }
  }
}
于 2009-01-08T10:02:17.807 回答