我正在使用带有固定线程池的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服务轮换任务？
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);
}
}
当然，在没有 's 方便的工厂方法的帮助下，您必须做更多的工作来自己实例化它ExecutorService，但是构造函数很简单，可以理解。
答案更多地与用于 的实例的工作队列ExecutorService的实现有关。所以，我建议：
First choose an implementation of java.util.concurrent.BlockingQueue (an example) that provides a circular queue functionality. NOTE, the reason BlockingQueue has been chosen is that to wait until the next task is provided to queue; so, in case of circular + blocking queue, you should be careful how to provide the same behavior and functionality.
Instead of using Executors.new... to create a new ThreadPoolExecutor use a direct constructor such as
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)
This way, unless you command the executor to shutdown, it will try to fetch the next task from the queue for execution from its work queue which is a circular container for tasks.
I suggest the following solution which completely uses functionality existing in the standard library concurrency utils. It uses a CyclicBarrier with a task decorator class and a barrier action which re-submits all tasks:
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 );
}
}
}
You could simply check that all the tasks have been executed and resubmit them once it is the case, like this for example:
List<Future> futures = new ArrayList<>();
for (Runnable task : myTasks) {
futures.add(threadPoolService.submit(task));
}
//wait until completion of all tasks
for (Future f : futures) {
f.get();
}
//restart
......
EDIT
It seems you want to resubmit a task as soon as it gets completed. You could use an ExecutorCompletionService which enables you to retrieve tasks as and when they get executed, - see below a simple example with 2 tasks that get resubmitted a few times as soon as they are completed. Sample output:
Task 1 submitted pool-1-thread-1
Task 2 submitted pool-1-thread-2
Task 1 completed pool-1-thread-1
Task 1 submitted pool-1-thread-3
Task 2 completed pool-1-thread-2
Task 1 completed pool-1-thread-3
Task 2 submitted pool-1-thread-4
Task 1 submitted pool-1-thread-5
Task 1 completed pool-1-thread-5
Task 2 completed pool-1-thread-4
public class Test1 {
public final ConcurrentMap<String, String> concurrentMap = new ConcurrentHashMap<>();
public final AtomicInteger retries = new AtomicInteger();
public final Object lock = new Object();
public static void main(String[] args) throws InterruptedException, ExecutionException {
int count = 0;
List<Runnable> myTasks = new ArrayList<>();
myTasks.add(getRunnable(1));
myTasks.add(getRunnable(2));
ExecutorService threadPoolService = Executors.newFixedThreadPool(10);
CompletionService<Runnable> ecs = new ExecutorCompletionService<Runnable>(threadPoolService);
for (Runnable task : myTasks) {
ecs.submit(task, task);
}
//wait until completion of all tasks
while(count++ < 3) {
Runnable task = ecs.take().get();
ecs.submit(task, task);
}
threadPoolService.shutdown();
}
private static Runnable getRunnable(final int i) {
return new Runnable() {
@Override
public void run() {
System.out.println("Task " + i + " submitted " + Thread.currentThread().getName() + " ");
try {
Thread.sleep(500 * i);
} catch (InterruptedException ex) {
System.out.println("Interrupted");
}
System.out.println("Task " + i + " completed " + Thread.currentThread().getName() + " ");
}
};
}
}