线程池（五）：线程池使用场景问题

线程池（五）：线程池使用场景问题

1 线程池使用场景CountDownLatch、Future

1.1 CountDownLatch

CountDownLatch是Java并发包中的一个同步辅助类。它允许一个或多个线程等待直到在其他线程中执行的一组操作完成。

原理

CountDownLatch内部维护了一个计数器，该计数器的初始值为需要等待完成的操作数量。当一个线程调用countDown()方法时，计数器会减1。而调用await()方法的线程会被阻塞，直到计数器的值变为0，此时所有等待的线程会被释放继续执行。

示例代码

import java.util.concurrent.CountDownLatch;public class CountDownLatchExample {public static void main(String[] args) throws InterruptedException {int numThreads = 5;CountDownLatch latch = new CountDownLatch(numThreads);for (int i = 0; i < numThreads; i++) {new Thread(() -> {try {// 模拟线程执行任务Thread.sleep(1000);System.out.println(Thread.currentThread().getName() + " 任务完成");} catch (InterruptedException e) {e.printStackTrace();} finally {latch.countDown();}}).start();}// 主线程等待所有子线程完成任务latch.await();System.out.println("所有任务已完成，主线程继续执行");}
}

1.2 案例一（es数据批量导入）

需求分析

在将大量数据导入Elasticsearch（ES）时，为了提高导入效率，通常会采用多线程并发导入的方式。但是，在所有数据导入完成之前，可能需要等待所有线程执行完毕，才能进行后续的操作，比如数据校验、通知用户导入完成等。

实现步骤

1. 初始化CountDownLatch：根据要并发导入的任务数量，初始化CountDownLatch的计数器值。例如，如果要分10个线程并发导入数据，计数器初始值就设为10。
2. 创建线程执行导入任务：在每个线程中，编写数据导入ES的逻辑。在导入完成后，调用countDown()方法将计数器减1。
3. 主线程等待：主线程调用await()方法，阻塞等待计数器变为0，即所有数据导入线程都完成任务。

示例代码

import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class EsBulkImport {private static final RestHighLevelClient client = new RestHighLevelClient(); // 假设已正确初始化ES客户端private static final int BATCH_SIZE = 1000; // 每批导入的数据量private static final int THREAD_COUNT = 10; // 并发线程数public static void main(String[] args) throws InterruptedException {List<DataObject> dataList = new ArrayList<>(); // 假设已准备好要导入的数据CountDownLatch latch = new CountDownLatch(THREAD_COUNT);for (int i = 0; i < THREAD_COUNT; i++) {int startIndex = i * BATCH_SIZE;int endIndex = Math.min((i + 1) * BATCH_SIZE, dataList.size());List<DataObject> subList = dataList.subList(startIndex, endIndex);new Thread(() -> {BulkRequest bulkRequest = new BulkRequest();for (DataObject data : subList) {bulkRequest.add(client.prepareIndex().setSource(data.toJson(), XContentType.JSON).request());}try {BulkResponse bulkResponse = client.bulk(bulkRequest);if (bulkResponse.hasFailures()) {// 处理失败情况System.err.println("导入失败: " + bulkResponse.buildFailureMessage());} else {System.out.println("部分数据导入成功");}} catch (IOException e) {e.printStackTrace();} finally {latch.countDown();}}).start();}latch.await();System.out.println("所有数据导入完成");}static class DataObject {// 假设数据对象有相关属性和方法public String toJson() {// 转换为JSON字符串的逻辑return "{}";}}
}

1.3 案例二（数据汇总）

需求分析

在一个系统中，可能需要从多个不同的数据源（如不同的数据库表、不同的API接口等）获取数据，然后将这些数据进行汇总分析。为了提高获取数据的效率，可以使用多线程并发从各个数据源获取数据，最后再进行汇总。

实现步骤

1. 初始化CountDownLatch：根据数据源的数量，设置CountDownLatch的计数器值。比如有5个数据源，计数器就初始化为5。
2. 创建线程获取数据：每个线程负责从一个数据源获取数据。获取完成后，调用countDown()方法。
3. 汇总数据：主线程在调用await()方法等待所有数据获取线程完成后，进行数据汇总操作。

示例代码

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class DataAggregation {public static void main(String[] args) throws InterruptedException {int numSources = 5;CountDownLatch latch = new CountDownLatch(numSources);List<Integer> dataList = new ArrayList<>();for (int i = 0; i < numSources; i++) {new Thread(() -> {try {// 模拟从数据源获取数据int data = getDataSourceData();dataList.add(data);} catch (Exception e) {e.printStackTrace();} finally {latch.countDown();}}).start();}latch.await();int sum = 0;for (int data : dataList) {sum += data;}System.out.println("汇总后的数据: " + sum);}private static int getDataSourceData() {// 模拟从数据源获取数据的逻辑，这里简单返回一个随机数return (int) (Math.random() * 100);}
}

1.4 案例三（异步调用）

需求分析

在一些业务场景中，需要发起多个异步调用（如调用多个远程服务接口），并且在所有异步调用完成后，再进行统一的处理（如合并结果、更新状态等）。使用CountDownLatch可以方便地实现这种等待所有异步操作完成的需求。

实现步骤

1. 初始化CountDownLatch：根据异步调用的数量设置计数器值。例如有3个异步调用，计数器初始值设为3。
2. 发起异步调用：在每个线程中发起异步调用（可以是通过网络请求远程服务等操作）。调用完成后，调用countDown()方法。
3. 统一处理：主线程调用await()方法等待所有异步调用线程完成，然后进行后续的统一处理逻辑。

示例代码

import java.util.concurrent.CountDownLatch;public class AsyncInvocation {public static void main(String[] args) throws InterruptedException {int numCalls = 3;CountDownLatch latch = new CountDownLatch(numCalls);for (int i = 0; i < numCalls; i++) {new Thread(() -> {try {// 模拟异步调用，这里简单睡眠一段时间Thread.sleep(1000);System.out.println(Thread.currentThread().getName() + " 异步调用完成");} catch (InterruptedException e) {e.printStackTrace();} finally {latch.countDown();}}).start();}latch.await();System.out.println("所有异步调用完成，进行统一处理");}
}

2 如何控制某个方法允许并发访问线程的数量

2.1 方法一：使用Semaphore（信号量）

原理

Semaphore是Java并发包中的一个类，它通过控制内部的许可证数量来限制同时访问的线程数量。可以初始化Semaphore并指定许可证的数量，当一个线程想要访问受保护的资源（如某个方法）时，需要先获取许可证（调用acquire()方法），使用完后释放许可证（调用release()方法）。如果没有可用的许可证，线程将被阻塞，直到有其他线程释放许可证。

示例代码

import java.util.concurrent.Semaphore;public class SemaphoreExample {private static final Semaphore semaphore = new Semaphore(3); // 允许同时3个线程访问public static void accessMethod() {try {semaphore.acquire();System.out.println(Thread.currentThread().getName() + " 进入方法");// 模拟方法执行逻辑，这里简单睡眠一段时间Thread.sleep(1000);System.out.println(Thread.currentThread().getName() + " 离开方法");} catch (InterruptedException e) {e.printStackTrace();} finally {semaphore.release();}}public static void main(String[] args) {for (int i = 0; i < 5; i++) {new Thread(() -> accessMethod()).start();}}
}

2.2 方法二：使用线程池（ThreadPoolExecutor）

原理

线程池可以控制线程的创建、复用和销毁，并且可以设置核心线程数、最大线程数等参数。通过合理设置这些参数，可以间接控制并发访问某个方法的线程数量。例如，设置核心线程数和最大线程数为相同的值，就可以限制同时执行任务（访问方法）的线程数量。

示例代码

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class ThreadPoolExample {private static final ExecutorService executorService = Executors.newFixedThreadPool(3); // 固定大小为3的线程池public static void accessMethod() {System.out.println(Thread.currentThread().getName() + " 进入方法");// 模拟方法执行逻辑，这里简单睡眠一段时间try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + " 离开方法");}public static void main(String[] args) {for (int i = 0; i < 5; i++) {executorService.submit(() -> accessMethod());}executorService.shutdown();}
}

2.3 方法三：使用ReentrantLock结合条件变量（Condition）

原理

ReentrantLock是可重入锁，它提供了比synchronized更灵活的锁机制。通过结合条件变量（Condition），可以实现更精细的线程同步控制。可以使用一个计数器来记录当前正在访问方法的线程数量，当达到限制数量时，新的线程通过条件变量等待，当有线程离开方法时，通知等待的线程。

示例代码

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;public class LockConditionExample {private static final ReentrantLock lock = new ReentrantLock();private static final Condition condition = lock.newCondition();private static int activeThreads = 0;private static final int MAX_THREADS = 3;public static void accessMethod() {lock.lock();try {while (activeThreads >= MAX_THREADS) {try {condition.await();} catch (InterruptedException e) {e.printStackTrace();}}activeThreads++;System.out.println(Thread.currentThread().getName() + " 进入方法");// 模拟方法执行逻辑，这里简单睡眠一段时间Thread.sleep(1000);System.out.println(Thread.currentThread().getName() + " 离开方法");activeThreads--;condition.signalAll();} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}public static void main(String[] args) {for (int i = 0; i < 5; i++) {new Thread(() -> accessMethod()).start();}}
}

以上详细介绍了线程池使用场景中涉及的CountDownLatch、Future相关内容，以及控制方法并发访问线程数量的多种方式，并通过丰富的代码示例进行了说明。