java多线程-限制提交到ExecutorService

Question

我有一个包含数千行的数据文件。我正在读取它们并将它们保存在数据库中。我想以50行的批次多线程处理这个过程。当我在文件中读取时，10行被提交给ExecutorService。

ExecutorService executor = Executors.newFixedThreadPool(5);`

我可以在一段时间内循环执行以下操作，直到行结束。。。。

 Future<Integer> future = executor.submit(callableObjectThatSaves10RowsAtOneTime);

但是，如果处理10行需要时间，我不想将整个文件读入内存。我只想在其中一个线程返回之前提交5个线程，然后提交下一个线程。

假设一个线程需要20秒来保存10条记录，我不希望ExecutorService被输入数千行，因为读取过程正在继续读取并提交给ExecutorService

实现这一目标的最佳方法是什么？

Answer 1

您可以使用链接列表来执行此操作

int threads = 5;
ExecutorService service = Executors.newFixedThreadPool(threads);
LinkedList<Future<?>> futures = new LinkedList<>();

//As long as there are rows to save:
while(moreRowsLeft()){
    //dump another callable onto the queue:
    futures.addLast(service.submit(new RowSavingCallable());

    //if the queue is "full", wait for the next one to finish before
    //reading any more of the file:
    while(futures.size() >= 2*threads) futures.removeFirst().get();
}

//All rows have been submitted but some may still be writing to the DB:
for(Future<?> f : futures) future.get();

//All rows have been saved at this point

您可能想知道为什么我允许未来的数量达到机器上线程数量的两倍-这允许执行器服务线程处理数据库保存，而主线程正在创建更多的工作要做。这有助于在工作线程忙于数据库写入时隐藏与提供更多可调用项以供处理相关的任何I/O开销。