ElasticSearch 7.x 版本使用 BulkProcessor 实现批量添加数据

import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
 
public class ElasticSearchBulkProcessorExample {
 
    public static void main(String[] args) {
        // 假设 client 已经创建并配置好
        RestHighLevelClient client = new RestHighLevelClient(...);
 
        BulkProcessor bulkProcessor = BulkProcessor.builder(
                client,
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                        // 在执行 bulk 请求前可以进行一些操作
                    }
 
                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                        // 在执行 bulk 请求后可以进行一些操作
                    }
 
                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                        // 如果 bulk 请求失败，可以进行一些错误处理操作
                    }
                })
                .setBulkActions(1000) // 每 1000 个请求执行一次 bulk 操作
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)) // 每 5MB 数据执行一次 bulk 操作
                .setFlushInterval(TimeValue.timeValueSeconds(5)) // 每 5 秒执行一次 bulk 操作
                .setConcurrentRequests(1) // 设置并发执行 bulk 请求的数量
                .build();
 
        // 添加数据到 bulk 进行批量处理
        for (int i = 0; i < 10000; i++) {
            String json = "{\"index\":{\"_id\":\"" + i + "\"}}";
            String data = "{\"data\":\"value" + i + "\"}";
            bulkProcessor.add(new IndexRequest("index").source(json, XContentType.JSON).source(data, XContentType.JSON));
        }
 
        // 关闭 bulkProcessor 以确保所有未处理的请求都被处理完毕
        bulkProcessor.close();
 
        // 关闭 client
        try {
            client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

这段代码展示了如何创建一个 BulkProcessor 对象，并设置了一些参数，如每批索引的数量、批量大小和刷新间隔。然后，它循环添加索引请求到 BulkProcessor，并在循环结束后关闭 \`BulkProcesso