Elasticsearch的Java客户端--文档操作--批量请求

官方文档：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-document-bulk.html

@Test
    public void TestExecutionStep() throws IOException {
        // 1.创建客户端
        RestClientBuilder clientBuilder = RestClient.builder(new HttpHost("127.0.0.1", 9200, "http"));
        RestHighLevelClient client = new RestHighLevelClient(clientBuilder);

        // 2.创建批量操作请求请求
        BulkRequest request = new BulkRequest();

        // 3.往批量操作请求请求中添加请求
        Map map = new HashMap();
        map.put("name", "jack");
        map.put("age", "23");
        //添加索引请求
        request.add(new IndexRequest("posts").id("1").source(map));
        //添加更新请求
        request.add(new UpdateRequest("posts", "1")
                .doc(XContentType.JSON, "other", "text"));
        //添加删除请求
        request.add(new DeleteRequest("posts", "1"));

        // 4.执行请求，返回响应
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            DocWriteResponse response = bulkItemResponse.getResponse();

            // 根据请求类型强转相应对象
            switch (bulkItemResponse.getOpType()){
                case INDEX:
                case CREATE:
                    IndexResponse indexResponse = (IndexResponse) response;
                    break;
                case UPDATE:
                    UpdateResponse updateResponse = (UpdateResponse) response;
                    break;
                case DELETE:
                    DeleteResponse deleteResponse = (DeleteResponse) response;
                    break;
            }
        }
    }

BulkProcess，用于在批量请求执行前，执行后，出错后额外执行一些操作

@Test
    public void TestListener() throws IOException, InterruptedException {
        // 1.创建客户端
        RestClientBuilder clientBuilder = RestClient.builder(new HttpHost("127.0.0.1", 9200, "http"));
        RestHighLevelClient client = new RestHighLevelClient(clientBuilder);

        // 2.创建批量请求监听器
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {

            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                System.out.println("n执行请求前。。。。");
                System.out.println(executionId);
                System.out.println(request.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                System.out.println("n执行请求后。。。。");
                System.out.println(executionId);
                System.out.println(request.numberOfActions());
                BulkItemResponse[] items = response.getItems();
                for (BulkItemResponse item : items) {
                    DocWriteRequest.OpType opType = item.getOpType();
                    System.out.println(opType);
                    RestStatus status = item.status();
                    System.out.println(status.getStatus());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                System.out.println("请求抛出异常。。。。");
                System.out.println(executionId);
                System.out.println(failure.toString());
            }
        };

        //执行添加的请求
        BulkProcessor.Builder builder = BulkProcessor.builder((request, bulkListener) ->
                        client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
                listener, "bulk-processor-name");
        BulkProcessor bulkProcessor = builder.build();

        // 3.往批量操作请求请求中添加请求
        Map map = new HashMap();
        map.put("name", "jack");
        map.put("age", "23");
        bulkProcessor.add(new IndexRequest("posts").id("1").source(map));
        bulkProcessor.add(new UpdateRequest("posts", "1")
                .doc(XContentType.JSON, "other", "text"));
        bulkProcessor.add(new DeleteRequest("posts", "1"));

        // 关闭 bulkprocessor 实例，指定等待时间
        boolean result = bulkProcessor.awaitClose(30L, TimeUnit.SECONDS);
        System.out.println(result);
    }