如何實現ExecutorService以執行批處理任務

Question

我正在尋找一種方法來執行java中的批處理任務。這個想法是基於線程池的ExecutorService，這將允許我從main線程的不同線程中傳播一組Callable。這個類應該提供一個waitForCompletion方法，它將使main線程進入休眠狀態，直到執行完所有任務。然後應喚醒線程，它將執行一些操作並重新提交一組任務。

此過程將重複多次，所以我想使用ExecutorService.shutdown，因爲這需要創建ExecutorService的多個實例。

目前我已經在使用AtomicInteger以下方式來實現它，和Lock/Condition：

public class BatchThreadPoolExecutor extends ThreadPoolExecutor { 
    private final AtomicInteger mActiveCount; 
    private final Lock   mLock; 
    private final Condition  mCondition; 

    public <C extends Callable<V>, V> Map<C, Future<V>> submitBatch(Collection<C> batch){ 
    ... 
    for(C task : batch){ 
     submit(task); 
     mActiveCount.incrementAndGet(); 
    } 
    } 

    @Override 
    protected void afterExecute(Runnable r, Throwable t) { 
    super.afterExecute(r, t); 
    mLock.lock(); 
    if (mActiveCount.decrementAndGet() == 0) { 
     mCondition.signalAll(); 
    } 
    mLock.unlock(); 
    } 

    public void awaitBatchCompletion() throws InterruptedException { 
    ... 
    // Lock and wait until there is no active task 
    mLock.lock(); 
    while (mActiveCount.get() > 0) { 
     try { 
     mCondition.await(); 
     } catch (InterruptedException e) { 
     mLock.unlock(); 
     throw e; 
     } 
    } 
    mLock.unlock(); 
    } 
} 

請不，我不一定會從一批提交所有的任務一次完成，因此CountDownLatch做似乎不是一種選擇。

這是一個有效的方法嗎？有沒有更高效/優雅的方式來實現呢？

感謝

Answer 1

我認爲ExecutorService本身將能夠執行您的要求。

致電invokeAll([...])並重復執行所有任務。如果您可以遍歷所有期貨，所有任務都已完成。

Answer 2

我同意@ckuetbach的默認Java Executors應該爲您提供執行「批處理」作業所需的所有功能。

如果我是你，我只需提交一堆工作，等待他們完成ExecutorService.awaitTermination()，然後啓動一個新的ExecutorService。要做到這一點，以節省「線程創作」是不成熟的優化，除非你每秒做這個100秒的事情。

如果你真的被困在使用相同的ExecutorService爲每個批次，那麼你可以自己分配一個ThreadPoolExecutor，並在一個循環看ThreadPoolExecutor.getActiveCount()。喜歡的東西：

BlockingQueue jobQueue = new LinkedBlockingQueue<Runnable>(); 
ThreadPoolExecutor executor = new ThreadPoolExecutor(NUM_THREADS, NUM_THREADS, 
    0L, TimeUnit.MILLISECONDS, jobQueue); 
// submit your batch of jobs ... 
// need to wait a bit for the jobs to start 
Thread.sleep(100); 
while (executor.getActiveCount() > 0 && jobQueue.size() > 0) { 
    // to slow the spin 
    Thread.sleep(1000); 
} 
// continue on to submit the next batch 

Answer 3

至於其他的答案指出，似乎沒有爲你的使用情況中的任何一部分需要自定義的ExecutorService。

在我看來，你需要做的就是提交一個批處理，等待它們全部完成，同時忽略主線程中的中斷，然後根據第一批結果提交另一批。我相信這只是一個問題：

ExecutorService service = ...; 

    Collection<Future> futures = new HashSet<Future>(); 
    for (Callable callable : tasks) { 
     Future future = service.submit(callable); 
     futures.add(future); 
    } 

    for(Future future : futures) { 
     try { 
      future.get(); 
     } catch (InterruptedException e) { 
      // Figure out if the interruption means we should stop. 
     } 
    } 

    // Use the results of futures to figure out a new batch of tasks. 
    // Repeat the process with the same ExecutorService.