Подождите, пока не будет выполнено какое-либо из Future <T>

Ответ 1

Насколько я знаю, 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();
    }
}

Ответ 2

просто, зайдите ExecutorCompletionService.

Ответ 3

ExecutorService.invokeAny

Ответ 4

Почему бы просто не создать очередь результатов и не подождать очереди? Или проще, используйте CompletionService, поскольку это то, что это такое: очередь результатов ExecutorService +.

Ответ 5

Это довольно легко с 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");
    }
}

Ответ 6

Так как вам все равно, какой из них заканчивается, почему бы просто не использовать один WaitHandle для всех потоков и ждать по этому поводу? Независимо от того, что заканчивается первым, можно установить дескриптор.

Ответ 7

См. этот параметр:

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();
    }
  }
}