如何继续向执行者提交任务答案

【问题标题】：How to keep on submitting task to an executor如何继续向执行者提交任务
【发布时间】：2017-01-18 16:44:06
【问题描述】：

我一直在寻找以下情况的解决方案：我有一个 Callables 的 HashSet，我正在将此 Set 提交给执行程序以进行并行执行。现在我希望在任何提交的任务完成后，我应该能够将一个新的 Callable 分配给 executor。

我尝试了这段代码，但是有了这个，如果我使用 executor.invoke 则 Executor 会等到所有任务都完成，如果我使用 executor.submit 则任务会按顺序完成。任何帮助将不胜感激。

package poc.threading;

import java.util.HashSet;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ConcurrencyPoC_UsingExecutor_GetFreeThread {

    public static void main(String[] args) {
        // TODO Auto-generated method stub

        executor();
    }


    public static void executor()
    {
        try{

            ExecutorService ex = Executors.newCachedThreadPool();

            //create a set with all callables task
            HashSet<Callable<Object>> callables = new HashSet<>();
            callables.add(task1());
            callables.add(task2());
            callables.add(task3());
            callables.add(task4());

            //executes all task together but executor waits for completion of all tasks

            List<Future<Object>> fu = ex.invokeAll(callables);
            for(int i=0; i<fu.size(); i++)
            {
                System.out.println(fu.get(i).get() + " , " + Thread.currentThread().getName().toString());
            }

            //executes tasks sequentially 
            for(Callable<Object> task : callables)
            {
                Future<Object> future = ex.submit(task);
                System.out.println(future.get() + " , " + Thread.currentThread().getName().toString());
            }
            ex.shutdownNow();
        }
        catch(Exception e)
        {
            e.printStackTrace();
        }
    }


    public static Callable<Object> task1() throws InterruptedException
    {
        return new Callable<Object>() {

            @Override
            public Object call() throws Exception {

                int count = 0;
                while(count < 3)
                {
                    System.out.println("****** SLEEP TASK1 ******* "+count);
                    Thread.sleep(500);
                    count ++;
                }
                return "Sleep Task Of 500 Completed";
            }
        };
    }


    public static Callable<Object> task2() throws InterruptedException
    {
        return new Callable<Object>() {

            @Override
            public Object call() throws Exception {

                int count = 0;
                while(count < 6)
                {
                    System.out.println("****** SLEEP TASK2 ******* "+count);
                    Thread.sleep(300);
                    count ++;
                }
                return "Sleep Task Of 300 Completed";
            }
        };
    }


    public static Callable<Object> task3() throws InterruptedException
    {
        return new Callable<Object>() {

            @Override
            public Object call() throws Exception {

                int count = 0;
                while(count < 2)
                {
                    System.out.println("****** SLEEP TASK3 ******* "+count);
                    Thread.sleep(1000);
                    count ++;
                }
                return "Sleep Task Of 1000 Completed";
            }
        };
    }

    public static Callable<Object> task4() throws InterruptedException
    {
        return new Callable<Object>() {

            @Override
            public Object call() throws Exception {

                int count = 0;
                while(count < 4)
                {
                    System.out.println("****** SLEEP TASK4 ******* "+count);
                    Thread.sleep(600);
                    count ++;
                }
                return "Sleep Task Of 1000 Completed";
            }
        };
    }
}

【问题讨论】：

我不清楚这个问题。如果您想通知代码的某些部分某项任务已完成，您可以使用观察者模式并为该任务添加一个侦听器。

标签： java multithreading threadpool executorservice

【解决方案1】：

您的任务在第二个示例中按顺序运行的原因是因为您在未来调用 get() 之前，在后续任务上调用 submit()。如果您在任何gets 之前执行所有submits，那么它们将并行运行。

如果您正在寻找相互依赖的任务，请查看 CompletableFuture 类。这种类型的未来将允许您在第一个任务开始后开始另一个任务：

CompletableFuture<Object> task1 = CompletableFuture.supplyAsync(() -> task1(), ex);
CompletableFuture<Object> task2 = task1.thenApplyAsync(task1Result -> task2(), ex);

【讨论】：