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

Question

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

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

У меня есть приложение Java, которое в основном выполняет 2 отдельные обработки (называемые 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 ().

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

Answer 1

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

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

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

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

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

Answer 2

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

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

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

Answer 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 () генерирует исключение, вы можете оказаться в той части кода, которая предполагает выполнение обеих задач, что может быть плохо.

Answer 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";
        }

    }
}

Answer 5

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

Answer 6

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

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