java - 如何管理 M 个线程(每个任务 1 个)确保同时只有 N 个线程。 N < M. 在 Java 中

Question

我在 java 中有一个任务队列。这个队列在数据库中的一个表中。

我需要:

• 每个任务仅 1 个线程
• 同时运行的线程不超过 N 个。这是因为线程与数据库交互，我不想打开一堆数据库连接。

我想我可以这样做:

final Semaphore semaphore = new Semaphore(N);
while (isOnJob) {
    List<JobTask> tasks = getJobTasks();
    if (!tasks.isEmpty()) {
        final CountDownLatch cdl = new CountDownLatch(tasks.size());
        for (final JobTask task : tasks) {
            Thread tr = new Thread(new Runnable() {

                @Override
                public void run() {
                    semaphore.acquire();
                    task.doWork();
                    semaphore.release();
                    cdl.countDown();
                }

            });
        }
        cdl.await();
    }
}

我知道存在一个 ExecutorService 类，但我不确定是否可以将其用于此目的。

那么，您认为这是最好的方法吗？或者您能否向我说明 ExecutorService 的工作原理以解决此问题？

最终解决方案:

我认为最好的解决方案是这样的:

while (isOnJob) {
    ExecutorService executor = Executors.newFixedThreadPool(N);
    List<JobTask> tasks = getJobTasks();
    if (!tasks.isEmpty()) {
        for (final JobTask task : tasks) {
            executor.submit(new Runnable() {

                @Override
                public void run() {
                    task.doWork();
                }

            });
        }
    }
    executor.shutdown();
    executor.awaitTermination(Long.MAX_VALUE, TimeUnit.HOURS);
}

非常感谢提供遮阳篷。顺便说一句，我正在使用连接池，但对数据库的查询非常繁重，我不想同时拥有不受控制的任务数量。

Answer 1

您确实可以使用 ExecutorService。例如，使用 newFixedThreadPool 创建一个新的固定线程池方法。这样，除了缓存线程外，您还可以保证不超过 n 个线程同时运行。

沿着这些线的东西:

private static final ExecutorService executor = Executors.newFixedThreadPool(N);
// ...
while (isOnJob) {
    List<JobTask> tasks = getJobTasks();
    if (!tasks.isEmpty()) {
        List<Future<?>> futures = new ArrayList<Future<?>>();
        for (final JobTask task : tasks) {
                Future<?> future = executor.submit(new Runnable() {    
                        @Override
                        public void run() {
                                task.doWork();
                        }
                });
                futures.add(future);
        }
        // you no longer need to use await
        for (Future<?> fut : futures) {
          fut.get();
        }
    }
}

请注意，您不再需要使用锁存器，因为 get 将在必要时等待计算完成。