使用 CompletionService 处理大量任务

Question

我需要在多核机器上处理大量（>1 亿）请求（每个请求是处理数据文件中的一行，并且涉及到与远程系统的一些 I/O。虽然细节确实没关系，具体任务是从一些数据文件中加载分布式 Hazelcast 地图）。执行将通过ThreadPoolExecutor 处理。一个线程将读取文件，然后将数据提交给多个独立线程以将其放入映射中。机器有 32 个核心，所以有足够的可用于并行加载地图。

由于请求数量众多，一般的创建任务并排队到执行器服务的做法是行不通的，因为排队的任务会占用太多的内存。

这带来了ExecutorCompletionService。有了它，将在先前的操作完成时提交任务，这可以通过调用take()（或poll()，如适用）得知。当执行器服务的所有线程都使用时，这将正常工作。但是，“加载所有线程”还没有完成。有两个阶段：

• 填满队列：当池中仍有未使用的线程时，将任务提交给 ExecutorCompletionService，不要等到提交更多

• 喂入队列：一旦线程全部使用完毕，只有在前一个任务完成后才提交任务。因此，将尽可能快地提供行，但不会更快，也不会排队。

上面可以编码，但我想知道上面的逻辑是否已经实现，我不知何故错过了它。我问是因为它看起来很常见。

Answer 1

您可以在创建ThreadPoolExecutor 时指定BlockingQueue 实现。如果您要避免的只是创建过多的Runnable 对象，那么您可以使用有界BlockingQueue，例如ArrayBlockingQueue 有一个线程将项目推送到队列中，当队列满负荷时，该线程将被阻塞。

Answer 2

如果我理解您的要求，（如果我错了，请纠正我）然后您需要一种机制，其中有多个任务，并且您需要最多 n 任务并行执行，其他任务应该在队列中等待，但是一旦你提交了一个任务，你就不想闲逛或让线程提交任务忙，它可以继续它的工作

对于同样的场景，我们混合使用LinkedBlockingQueue和Thread，相信一个简单的函数可以帮助你理解，

private final LinkedBlockingQueue<YourTaskObjType> EnqueuedTasks;

private void initTasksProcessingThreads(int numberOfThreads) 
{
    EnqueuedTasks= new LinkedBlockingQueue<YourTaskObjType>();
    for (int i = 0; i < numberOfThreads; i++) 
    {
        // each thread will run forever and process incoming
        //Change requests
        Thread worker = new Thread(new Runnable() 
        {               
            public void run() 
            {
                while (true) 
                {
                    try 
                    {   
                        YourTaskObjType task = EnqueuedTasks.take(); //This will wait infinitely until tasks are available
                        PerformTask(task); //Your function which will perform the task operation
                    } 
                    catch (InterruptedException e) 
                    {                                                                
                        Thread.currentThread().interrupt();
                        return;
                    } 
                    catch(Exception e)
                    {
                        e.printStackTrace();
                    }
                }
            }
        });         
        worker.start();
    }
}

然后你可以使用一个简单的函数将Tasks添加到LinkedBlockingQueue，

public void AddTask(YourTaskObjType TaskObj)
{
    EnqueuedTasks.put(TaskObj);                         
}