如何实现ExecutorService以轮流执行任务？

Question

我使用java.util.concurrent.ExecutorService和固定线程池来执行任务列表。我的任务列表通常在80-150个左右，我已经将任何时候运行的线程数限制为10个，如下所示：

ExecutorService threadPoolService = Executors.newFixedThreadPool(10);

for ( Runnable task : myTasks ) 
{     
    threadPoolService.submit(task); 
}

我的用例要求即使完成的任务也应该重新提交给ExecutorService，但是只有当所有已经提交的任务都被服务/完成时，才应该再次执行/获取。也就是说，提交的任务基本上应该轮流执行。因此，在这种情况下不会有threadPoolService.shutdown（）或threadPoolService.shutdownNow（）调用。

我的问题是，如何实现ExecutorService服务轮换基础任务？

Answer 1

我建议以下解决方案完全使用标准库并发utils中存在的功能。它使用一个带有任务装饰器类的CyclicBarrier和一个重新提交所有任务的障碍动作：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Rotation {

    private static final class RotationDecorator implements Runnable {
        private final Runnable          task;
        private final CyclicBarrier barrier;


        RotationDecorator( Runnable task, CyclicBarrier barrier ) {
            this.task = task;
            this.barrier = barrier;
        }


        @Override
        public void run() {
            this.task.run();
            try {
                this.barrier.await();
            } catch(InterruptedException e) {
                ; // Consider better exception handling
            } catch(BrokenBarrierException e) {
                ; // Consider better exception handling
            }
        }
    }


    public void startRotation( List<Runnable> tasks ) {
        final ExecutorService threadPoolService = Executors.newFixedThreadPool( 10 );
        final List<Runnable> rotatingTasks = new ArrayList<Runnable>( tasks.size() );
        final CyclicBarrier barrier = new CyclicBarrier( tasks.size(), new Runnable() {
            @Override
            public void run() {
                Rotation.this.enqueueTasks( threadPoolService, rotatingTasks );
            }
        } );
        for(Runnable task : tasks) {
            rotatingTasks.add( new RotationDecorator( task, barrier ) );
        }
        this.enqueueTasks( threadPoolService, rotatingTasks );
    }


    private void enqueueTasks( ExecutorService service, List<Runnable> tasks ) {
        for(Runnable task : tasks) {
            service.submit( task );
        }
    }

}

Answer 2

答案与用于ExecutorService实例的工作队列的实现更相关。因此，我建议：

>

不要使用Executors.new…来创建新的ThreadPoolExecutor使用直接构造函数，例如

公共线程池执行器（int corePoolsize，int最大池大小，long保留AliveTime，TimeUnit单位，BlockingQueue

这样，除非您命令执行程序关闭，否则它将尝试从其工作队列（任务的循环容器）中获取要执行的下一个任务。

Answer 3

ThreadPoolExecutor为afterExecution提供了一个扩展点，您可以将作业放回队列末尾。

public class TaskRepeatingThreadPoolExecutor extends ThreadPoolExecutor {

    public TaskRepeatingThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        this.submit(r);
    }
}

当然，您需要做更多的工作来自己实例化它，而无需借助ExecutorService的便捷工厂方法，但是构造函数非常简单，可以进行摸索。