Executor服务上的异步进程

Question

我正在使用多线程执行器服务。跨15个线程，当前进程在转移到下一个执行器服务之前完成了15个线程，下一个执行器服务再次跨15个线程。我正在寻找代码帮助，如果步骤1执行器服务中的任何当前线程完成处理，我需要转到下一个执行器服务，它可以开始使用线程，而不是在步骤1中等待完成15个线程来完成。

我想利用完成的线程，然后转到步骤2，即使其他线程在步骤1中执行，只要步骤1中的每个线程完成，步骤2就可以抓取并继续处理。建议请

// Step 1 
ExecutorService executor1 = Executors.newFixedThreadPool(15);
for (int i=0;i<=15;i++) { 
    Runnable worker = new Runnable("Step 1 Insert"); 
    executor1.execute(worker); } 
executor1.shutdown(); 
// Step 2 
ExecutorService executor2 = Executors.newFixedThreadPool(15);
for (int i=0;i<=15;i++) { 
    Runnable worker = new Runnable("Step 2 Insert"); 
    executor2.execute(worker); } 
executor2.shutdown();

Answer 1

您可以使用ExecutorCompletionService。我稍微修改了JavaDoc中的示例

ExecutorService executor1 = Executors.newFixedThreadPool(15);
ExecutorService executor2 = Executors.newFixedThreadPool(15);

....

CompletionService<Result> ecs = new ExecutorCompletionService<Result>(executor1);
for (Callable<Result> s : solvers)
    ecs.submit(s);
int n = solvers.size();
for (int i = 0; i < n; ++i) {
    Result r = ecs.take().get();
    if (r != null)
        executor2.submit(r);
}
executor1.shutdown();
//and shutdown executor2 when you don't need it

Result类只是一个例子，我假设它实现了Callable或Runnable，因此它可以提交到第二个执行器中。

您还可以将ExecutorService的实例初始化为new ThreadPoolExecutor（0,15,10，TimeUnit.SECONDS，new LinkedBlockingQueue（））以终止现在不需要的线程，但您希望以后重用此执行器。

或者你可以简单地使用一个执行者。从ExecutorCompletionService收到已完成的任务后，将其（或其结果）重新提交给同一执行者（感谢@Fildor）。但在这种情况下，您需要确定第一次完成任务的时间。

for (int i = 0; i < n;) {
    Result r = ecs.take().get();
    if (r != null && isCompletedFirstTime(r))//some user defined function
        executor1.submit(r);
    else 
        ++i;//we need to know when to stop, otherwise we'll stuck in `take()`
}

Answer 2

@SpiderPig的答案是IMHO的一个很好的解决方案，但如果您想要解耦步骤，我想给出一个替代方案：

使用一个执行器，在你的情况下，需求似乎是一个有15个线程的固定线程池。

接下来是定义Step-Runnable，就像这样：

class StepX implements Runnable{
    private final State _state; // Reference to the data to work on.
    StepX( State state ){
        _state = state;
    }

    public void run(){ 
        // work on _state
        executor.submit( new StepXplusOne( _state ) ); // reference executor in outer class and schedule next step.
    }
} 

您可以看到我使用了一个State对象，它保存执行步骤和收集结果所需的所有数据。当然，您需要将StepX定义为Step1，Step2，...

在外类中，您只需提交N个Step1-Runnable，它将只使用您的15个线程并完成步骤。

我遗漏了一个方法，当所有步骤都完成时发出信号，因为有很多可能性可以做到这一点，我相信你可以自己选择一个。

Answer 3

为什么不能在同一个Runnable中执行所有步骤？例如

ExecutorService executor = Executors.newFixedThreadPool(15);
for (int i=0;i<=15;i++) { 
    Runnable worker = new Runnable() {
        public void run() {
            doStep1();
            doStep2();
            doStep3();
            ...
        }
    }; 
    executor.execute(worker);
}
executor.shutdown();