这是一个使用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是可以容忍的,但绝对不是惯用的.你实际上是在你提交给你的那个附近包装一个额外 FutureTask的ExecutorService.你FutureTask被视为一个Runnable人ExecutorService.在内部,它将你的FutureTask-as- 包装成Runnable一个新的FutureTask并将它作为一个返回给你Future<?>.

相反,您应该将您的Callable<Object>实例提交给CompletionService.你将两个Callables放入via submit(Callable<V>),然后转身并拨打CompletionService#take()两次(每次提交一次Callable).这些调用将阻塞直到一个,然后其他提交的任务完成.

鉴于你已经拥有了一个Executor,ExecutorCompletionService在它周围构建一个新的并将你的任务放在那里.不要旋转和睡觉等待; CompletionService#take()将阻塞,直到您的任务之一完成(完成运行或取消)或线程等待take()中断.

Answer 3

Yuval的解决方案很好.作为替代方案,您也可以这样做:

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.