es之java插入优化（批量插入 代码

以下是使用Java的RestHighLevelClient进行ES批量插入的示例代码，其中包括了上述提到的优化措施：

public class ESInsertDemo {
    private static final String INDEX_NAME = "test_index";
    private static final String TYPE_NAME = "test_type";
    private static final int BULK_SIZE = 1000;
    private static final int THREAD_POOL_SIZE = 10;
    private static final int MAX_RETRY_TIMES = 3;

    private RestHighLevelClient client;

    public ESInsertDemo() {
        // 初始化ES客户端
        RestClient restClient = RestClient.builder(
                new HttpHost("localhost", 9200, "http")).build();
        client = new RestHighLevelClient(restClient);
    }

    public void insertBulk(List<Map<String, Object>> documents) {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                // 批量插入前操作
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                // 批量插入后操作
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                // 批量插入失败操作
            }
        };

        BulkProcessor bulkProcessor = BulkProcessor.builder(
                (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
                listener)
                .setBulkActions(BULK_SIZE)
                .setConcurrentRequests(THREAD_POOL_SIZE)
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), MAX_RETRY_TIMES))
                .build();

        for (Map<String, Object> document : documents) {
            IndexRequest indexRequest = new IndexRequest(INDEX_NAME, TYPE_NAME)
                    .source(document);
            bulkProcessor.add(indexRequest);
        }

        try {
            bulkProcessor.awaitClose(10, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            // 异常处理
        }
    }

    public void close() throws IOException {
        client.close();
    }
}

在上述代码中，我们使用了BulkProcessor来进行批量插入，同时设置了批量操作的文档数量、线程数量和重试次数等参数。在添加文档时，我们使用IndexRequest构建文档，并通过bulkProcessor.add()方法添加到批量请求中。

在实际使用中，我们可以将待插入文档分批加载进内存，然后调用insertBulk()方法进行批量插入。