在将数据批量导入到 Elasticsearch（ES）时，使用线程池和 CountDownLatch 可以有效管理并发处理，同时避免内存溢出（Out Of Memory, OOM）问题。下面是一个示例代码，演示如何利用这些工具将数据分批导入到 ES 中。
示例代码说明
- 使用线程池来并发处理数据的导入。
- 使用
CountDownLatch来等待所有任务完成，确保在导入数据之后执行后续操作。 - 数据批处理，避免一次性加载过多数据到内存中。
代码示例
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClientBuilder;import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class DataImporter {private static RestHighLevelClient client; // Elasticsearch 高级客户端private static final int BATCH_SIZE = 100; // 每批处理的大小private static final int THREAD_COUNT = 5; // 线程池线程数量public static void main(String[] args) throws InterruptedException {// 初始化 Elasticsearch 客户端client = createClient(); // 创建客户端的具体实现需要提供// 模拟获取数据，例如从数据库中读取数据List<Data> dataList = fetchData(); // 这里是获取待导入数据的方法// 计算批次int totalBatches = (int) Math.ceil((double) dataList.size() / BATCH_SIZE);ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);CountDownLatch latch = new CountDownLatch(totalBatches);// 开始批量导入数据for (int i = 0; i < totalBatches; i++) {int start = i * BATCH_SIZE;int end = Math.min(start + BATCH_SIZE, dataList.size());List<Data> batch = dataList.subList(start, end);executorService.submit(() -> {try {bulkInsert(batch);} catch (Exception e) {e.printStackTrace(); // 处理异常} finally {latch.countDown(); // 线程完成后减少计数}});}latch.await(); // 等待所有线程完成executorService.shutdown(); // 关闭线程池client.close(); // 关闭 Elasticsearch 客户端}private static void bulkInsert(List<Data> batch) {BulkRequest bulkRequest = new BulkRequest();for (Data data : batch) {// 假设 Data 类定义了某种需要导入的结构// 可以根据需求设置索引、文档类型和数据bulkRequest.add(data.toIndexRequest()); // 将数据添加到 BulkRequest 中}try {BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);if (bulkResponse.hasFailures()) {// 处理失败情况System.out.println("Bulk import failed: " + bulkResponse.buildFailureMessage());}} catch (Exception e) {// 处理异常e.printStackTrace();}}private static RestHighLevelClient createClient() {// 创建和配置 Elasticsearch 高级客户端的代码// 需要提供具体的 ES 服务地址、鉴权等信息return new RestHighLevelClient(RestClientBuilder builder); }private static List<Data> fetchData() {// 模拟读取数据的逻辑// 需要实现从数据库或其他数据源获取数据的代码return List.of(); // 返回待导入数据的列表}private static class Data {// 数据结构，例如与 Elasticsearch 文档相对应的字段public IndexRequest toIndexRequest() {// 生成 IndexRequest 对象，包含必要的索引信息和文档信息return new IndexRequest("your_index_name").source(this); // 具体实现需要根据业务需求}}
}
代码说明
- 线程池 (
ExecutorService)：管理多个线程并发执行导入操作，可以灵活配置线程数量以最佳利用资源。 CountDownLatch：在所有线程完成导入操作后，主线程会等待，确保所有数据批次都被成功处理。- 批处理：通过将数据分成小批次进行处理，避免占用过多内存，降低 OOM 的风险。
- Bulk API：使用 Elasticsearch 的 Bulk API 高效地批量插入数据。
注意事项
- 异常处理：确保在异常发生时能够及时捕获并处理，避免因一处错误导致整个导入流程中断。
- 资源释放：确保在流程结束后关闭线程池和 Elasticsearch 客户端，以释放相关资源。
- 性能调优：根据具体的需求和环境条件对线程数量和批处理大小进行调优，找到适合的平衡点，既提高导入速度又避免 OOM。
通过这样的方式，可以有效地将数据批量导入到 Elasticsearch，同时避免内存溢出，提高导入操作的稳定性和效率。
