Это хороший способ использовать java.util.concurrent.FutureTask?

Прежде всего, я должен сказать, что я совершенно новичок в API java.util.concurrent, так что, возможно, то, что я делаю, совершенно неверно.

Что я хочу делать?

У меня есть приложение Java, которое в основном запускает две отдельные обработки (называемые myFirstProcess, mySecondProcess), но эта обработка должна выполняться одновременно.

Итак, я попытался сделать это:

public void startMyApplication() {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    FutureTask<Object> futureOne = new FutureTask<Object>(myFirstProcess);
    FutureTask<Object> futureTwo = new FutureTask<Object>(mySecondProcess);
    executor.execute(futureOne);
    executor.execute(futureTwo);
    while (!(futureOne.isDone() && futureTwo.isDone())) {
        try {
            // I wait until both processes are finished.
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    logger.info("Processing finished");
    executor.shutdown();
    // Do some processing on results
    ...
}

myFirstProcess и mySecondProcess - это классы, реализующие Callable<Object>, и где вся их обработка выполняется в методе call().

Он работает достаточно хорошо, но я не уверен, что это правильный способ сделать это. Хороший способ делать то, что я хочу? Если нет, можете ли вы дать мне несколько советов по улучшению моего кода (и по-прежнему держать его как можно проще).

Ответ 1

Вам лучше использовать метод get().

futureOne.get();
futureTwo.get();

Оба из них ждут уведомления из потока, который он завершил обработку, это экономит ваш занятый-ожидание с помощью таймера, который вы используете, что неэффективно и не элегантно.

В качестве бонуса у вас есть API get(long timeout, TimeUnit unit), который позволяет вам определить максимальное время, в течение которого поток будет спать, и ждать ответа, и в противном случае продолжает работать.

Подробнее см. Java API.

Ответ 2

Использование FutureTask выше допустимо, но определенно не идиоматично. Фактически вы обертываете дополнительный FutureTask вокруг того, который вы отправили в ExecutorService. Ваш FutureTask рассматривается как Runnable с помощью ExecutorService. Внутри он переносит ваш FutureTask -as- Runnable в новый FutureTask и возвращает его вам как Future<?>.

Вместо этого вы должны отправить свои экземпляры Callable<Object> в CompletionService. Вы отбрасываете два Callable через submit(Callable<V>), затем поворачиваетесь и вызываете CompletionService#take() дважды (один раз для каждого отправленного Callable). Эти вызовы будут блокироваться до тех пор, пока не будут завершены одна, а затем и другие поставленные задачи.

Учитывая, что у вас уже есть Executor, создайте новый ExecutorCompletionService вокруг него и отпустите свои задачи там. Не спинь и не спишь; CompletionService#take() будет заблокирован до тех пор, пока одна из ваших задач не будет завершена (завершена работа или отменена), или поток, ожидающий на take(), прервется.

Ответ 3

Решение Юваля прекрасное. В качестве альтернативы вы также можете сделать это:

ExecutorService executor = Executors.newFixedThreadPool();
FutureTask<Object> futureOne = new FutureTask<Object>(myFirstProcess);
FutureTask<Object> futureTwo = new FutureTask<Object>(mySecondProcess);
executor.execute(futureOne);
executor.execute(futureTwo);
executor.shutdown();
try {
  executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
} catch (InterruptedException e) {
  // interrupted
}

В чем преимущество этого подхода? Там не так много различий, за исключением того, что таким образом вы прекращаете выполнение исполнителем каких-либо задач (вы можете сделать это и наоборот). Я обычно предпочитаю эту идиому этому.

Кроме того, если get() выдает исключение, вы можете оказаться в части вашего кода, который предполагает выполнение обеих задач, что может быть плохо.

Ответ 4

Вы можете использовать метод invokeall (Colelction....)

package concurrent.threadPool;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class InvokeAll {

    public static void main(String[] args) throws Exception {
        ExecutorService service = Executors.newFixedThreadPool(5);
        List<Future<java.lang.String>> futureList = service.invokeAll(Arrays.asList(new Task1<String>(),new Task2<String>()));

        System.out.println(futureList.get(1).get());
        System.out.println(futureList.get(0).get());
    }

    private static class Task1<String> implements Callable<String>{

        @Override
        public String call() throws Exception {
            Thread.sleep(1000 * 10);
            return (String) "1000 * 5";
        }

    }

    private static class Task2<String> implements Callable<String>{

        @Override
        public String call() throws Exception {
            Thread.sleep(1000 * 2);
            int i=3;
            if(i==3)
                throw new RuntimeException("Its Wrong");
            return (String) "1000 * 2";
        }

    }
}

Ответ 5

Вы можете использовать CyclicBarrier, если вы заинтересованы в запуске потоков одновременно или ожидаете их завершения а затем выполнить некоторую дополнительную обработку. См. Javadoc для получения дополнительной информации.

Ответ 6

Если ваши будущие задачи больше 2, рассмотрите [ListenableFuture][1].

Когда несколько операций должны начаться, как только произойдет другая операция начинается - " вентилятор" - ListenableFuture работает только: он запускает все запрошенные обратные вызовы. При немного большей работе мы можем " включить вентилятор" или вызвать ListenableFuture для вычисления, как только несколько других фьючерсы все закончились.