如何持续向执行者提交任务

Question

我一直在寻找这样一种情况的解决方案：我有一个调用项的哈希集，并且我要将这个集提交给执行器进行并行执行。现在我想只要任何提交的任务完成，我应该能够分配一个新的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";
            }
        };
    }
}

Answer 1

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

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

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