如何使用固定数量的线程在Java中运行某些数量的任务

Question

我有一个特定数量的线程THREAD_POOL_SIZE和一堆可以超过线程数的任务。我想使用这些k线程来运行我的所有任务。我在BlockingQueue中有这些任务，每个线程返回一个应该稍后聚合的结果。

在这里，我写了一个简单的程序，其中任务是从1到100的数字，我试图计算所有数字的总和。每个线程将从阻塞队列中选择一个数字并返回它。我正在使用Future来收集我的结果并在以后总结。

使用BlockingQueue的原因是因为我试图解决一个更大的问题，我可以在阻塞队列中拥有任务，并且我有一定数量的线程来运行这些任务。

我想知道如何修复下面的代码，使k线程继续处理阻塞队列中的条目？

static class Consumer implements Callable<Integer> {
    private BlockingQueue<Integer> sharedQueue;

    public Consumer(BlockingQueue<Integer> sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public Integer call() {
        while(true){
            try {
                return sharedQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public static void main(String[] args) throws Exception {
    int THREAD_POOL_SIZE = 10;
    int BLOCKING_QUEUE_SIZE = 100;
    BlockingQueue<Integer> sharedQueue = new ArrayBlockingQueue<>(BLOCKING_QUEUE_SIZE);
    ExecutorService execService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    List<Future<Integer>> futures = new ArrayList<>();
    for (int i = 0; i < BLOCKING_QUEUE_SIZE; i++) {
        sharedQueue.add(i);
    }

    for (int i = 0; i < THREAD_POOL_SIZE; i++) {
        futures.add(execService.submit(new Consumer(sharedQueue)));
    }

    int total = 0;
    for (Future<Integer> future : futures) {
        try {
            total += future.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    System.out.println("Total count: " + total);
    execService.shutdown();
}

谢谢你的帮助！

Answer 1

您需要向执行者添加100个期货，而不是10个：

for (int i = 0; i < THREAD_POOL_SIZE; i++) {

应该：

for (int i = 0; i < 100; i++) {

---

很奇怪您认为队列在这里真正添加了什么。没有它，您可以大规模简化代码。

int THREAD_POOL_SIZE = 10;
ExecutorService execService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
List<Future<Integer>> futures = new ArrayList<>();

for (int i = 0; i < 100; i++) {
    final int j = i;
    futures.add(execService.submit(() -> j));
}

int total = 0;
for (Future<Integer> future : futures) {
    try {
        total += future.get();
    } catch (Exception e) {
        e.printStackTrace();
    }
}
System.out.println("Total count: " + total);
execService.shutdown();

Answer 2

即使只看一眼您的代码，也会发现您的代码是一种过度杀伤的方式。有一些java类可以为你处理这些东西。请查看ExecutorService类和Executors类。 ExecutorService及其特定实现提供了托管线程池。您需要做的就是调用类newFixedThreadPool的方法Executors，然后将所有任务提交到该池。您不需要任何队列，您的任务将由线程池管理。您可以收集Futuresand监视器以查看任务完成的时间。你想的很简单