É uma boa maneira de usar java.util.concurrent.FutureTask?

Question

Em primeiro lugar, devo dizer que eu sou muito novo para o java.util.concurrent API, talvez por isso que eu estou fazendo é completamente errado.

O que eu quero fazer?

Eu tenho um aplicativo Java que basicamente é executado 2 processamento separado (chamado myFirstProcess , mySecondProcess ), mas estes processamento deve ser executado ao mesmo tempo.

Então, eu tentei fazer isso:

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 e mySecondProcess são classes que implementa Callable<Object>, e onde todo o seu processamento é feito no método call ().

Ele está trabalhando muito bem, mas não estou certo de que é a maneira correta de fazer isso. É uma boa maneira de fazer o que eu quero? Se não, você pode me dar algumas dicas para melhorar o meu código (e ainda mantê-lo tão simples quanto possível).

Answer 1

Você seria melhor fora de usar o método get().

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

Tanto que esperar para a notificação do thread que o processamento terminar, esta poupa o busy-wait-com-timer que você está usando agora que não é eficiente nem elegante.

Como um bônus, você tem a get(long timeout, TimeUnit unit) API que permite que você defina um tempo máximo para o segmento de sono e esperar por uma resposta, e outra continua em execução.

Veja a Java API para mais informações.

Answer 2

Os usos de FutureTask acima são toleráveis, mas definitivamente não idiomática. Você está realmente passando um adicional FutureTask torno do que você submetido ao ExecutorService. Seu FutureTask é tratado como um Runnable pela ExecutorService. Internamente, ele envolve seu FutureTask-as-Runnable em uma nova FutureTask e retorna para você como um Future<?>.

Em vez disso, você deve enviar suas instâncias Callable<Object> a um CompletionService . Você deixa cair dois Callables na via submit(Callable<V>), em seguida, virar e CompletionService#take() chamada duas vezes (uma vez para cada Callable submetido). Essas chamadas serão bloqueadas até que um e, em seguida, as outras tarefas apresentadas estão completos.

Uma vez que você já tem uma Executor na mão, construir um novo ExecutorCompletionService em torno dele e soltar suas tarefas lá. Não rodada e espera o sono; CompletionService#take() irá bloquear até que uma das suas tarefas estão completas (ou correr acabado ou cancelada) ou o fio à espera de take() é interrompida.

Answer 3

A solução da Yuval é bom. Como alternativa, você também pode fazer isso:

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
}

Qual é a vantagem dessa abordagem? Não há muita diferença realmente, exceto que desta forma você parar o executor de aceitar qualquer mais tarefas (você pode fazer isso de outra maneira também). I tendem a preferir este idioma para que embora.

Além disso, se qualquer get () lança uma exceção, você pode acabar em uma parte do seu código que assume ambas as tarefas são feitas, o que pode ser ruim.

Answer 4

Você pode usar invokeall (Coleção ....) Método

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

Você pode querer usar um CyclicBarrier se você estiver interessado em iniciar os threads ao mesmo tempo, ou esperando por eles para terminar e, em seguida, fazer algum processamento adicional. Consulte o javadoc para mais informações.

Answer 6

Se seus futureTasks são mais do que 2, por favor considere [ListenableFuture][1].

Quando várias operações devem começar assim que outra operação começa - " fan-out " - ListenableFuture simplesmente funciona: ele aciona todos os retornos de chamada solicitados. Com um pouco mais de trabalho, nós podemos " fan-in ", ou desencadear uma ListenableFuture para se computado assim como vários outros futuro tem tudo acabado.