java 操作elasticsearch 之rest API

        7.1.1



     org.apache.httpcomponents
     httpclient
     4.5.7


    org.elasticsearch.client
    elasticsearch-rest-high-level-client
    ${elasticsearch.version}


    org.elasticsearch.client
    elasticsearch-rest-client
    ${elasticsearch.version}

private static RestHighLevelClient restHighLevelClient = null;

    public static RestHighLevelClient createClient(String host, String port, String name, String password) {
        if (restHighLevelClient != null) {
            return restHighLevelClient;
        }
        //创建低级客户端,提供ip,端口,设置超时重试时间
        RestClientBuilder restClient = RestClient.builder(new HttpHost(host, Integer.parseInt(port)));
        // 设置账户密码
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(name, password));
        restClient.setHttpClientConfigCallback(f -> f.setDefaultCredentialsProvider(credentialsProvider));
        //创建高级客户端,传入低级客户端
        restHighLevelClient = new RestHighLevelClient(restClient);
        return restHighLevelClient;
    }

public static void searchAll(RestHighLevelClient restHighLevelClient, String indexName) {
        //创建查询请求对象,提供索引和类型,如果建表时规范的话,可以不提供类型的
        SearchRequest searchRequest = new SearchRequest(indexName).types("_doc");
        //构建query条件
        MatchAllQueryBuilder queryBuilder = new MatchAllQueryBuilder();
        //query条件作为查询条件,size表示返回结果的条数
        SearchSourceBuilder builder = new SearchSourceBuilder().query(queryBuilder).size(10);
        //请求对象携带条件,查询类型,一般默认即可
        searchRequest.source(builder).searchType(SearchType.DEFAULT);
        try {
            //通过高级客户端执行查询请求,返回响应对象
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            //拿到响应的匹配结果,遍历
            for (SearchHit hit : searchResponse.getHits().getHits()) {
                //转为String,也可以getSourceAsMap转为map,后续进行操作
                System.out.println(hit.getSourceAsString());
            }
        } catch (
                IOException e) {
            e.printStackTrace();
        }
    }

public static void insertMessage(RestHighLevelClient restHighLevelClient, String indexName, List> list) throws IOException {

        BulkRequest request = new BulkRequest();
        request.timeout("3m");
        for (Map map : list) {
            String id = String.valueOf(map.get("id"));
            map.remove("id");
            request.add(new IndexRequest(indexName, "_doc").id(id)
                    .opType("create").source(map));

        }
        BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.out.println(failure);
                continue;
            }
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
            if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                    || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                IndexResponse indexResponse = (IndexResponse) itemResponse;
                System.out.println(indexResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                System.out.println(updateResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                System.out.println(deleteResponse);
            }
        }
    }

public static void deleteMessage(RestHighLevelClient restHighLevelClient, String indexName, List list) throws IOException {
        BulkRequest request = new BulkRequest();
        request.timeout("3m");

        for (String id : list) {
            if (id == null || id.isEmpty()) {
                continue;
            }
            request.add(new DeleteRequest(indexName, id));
        }
        BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.out.println(failure);
                continue;
            }
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
            if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                    || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                IndexResponse indexResponse = (IndexResponse) itemResponse;
                System.out.println(indexResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                System.out.println(updateResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                System.out.println(deleteResponse);
            }
        }

    }

更新消息

public static void updateMessage(RestHighLevelClient restHighLevelClient, String indexName, List> list) throws IOException {
        BulkRequest request = new BulkRequest();
        request.timeout("3m");

        for (Map map : list) {
            String id = String.valueOf(map.getOrDefault("id", null));
            if (id == null || id.isEmpty()) {
                continue;
            }
            map.remove("id");
            request.add(new UpdateRequest(indexName, id).doc(new IndexRequest().source(map)));
        }
        BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.out.println(failure);
                continue;
            }
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
            if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                    || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                IndexResponse indexResponse = (IndexResponse) itemResponse;
                System.out.println(indexResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                System.out.println(updateResponse);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                System.out.println(deleteResponse);
            }
        }

    }

public static void searchAll(RestHighLevelClient restHighLevelClient, String indexName) {
        //创建查询请求对象,提供索引和类型,如果建表时规范的话,可以不提供类型的
        SearchRequest searchRequest = new SearchRequest(indexName).types("_doc");
        //构建query条件
        MatchAllQueryBuilder queryBuilder = new MatchAllQueryBuilder();
        //query条件作为查询条件,size表示返回结果的条数
        SearchSourceBuilder builder = new SearchSourceBuilder().query(queryBuilder).size(10);
        //请求对象携带条件,查询类型,一般默认即可
        searchRequest.source(builder).searchType(SearchType.DEFAULT);
        try {
            //通过高级客户端执行查询请求,返回响应对象
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            //拿到响应的匹配结果,遍历
            for (SearchHit hit : searchResponse.getHits().getHits()) {
                //转为String,也可以getSourceAsMap转为map,后续进行操作
                System.out.println(hit.getSourceAsString());
            }
        } catch (
                IOException e) {
            e.printStackTrace();
        }
    }

public static boolean exists(RestHighLevelClient client, String indexName) {
        GetIndexRequest request = new GetIndexRequest(indexName);
        request.local(false);
        request.humanReadable(true);
        request.includeDefaults(false);
        try {
            return client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {

            e.printStackTrace();
            return false;
        }
    }

public static void deleteIndex(RestHighLevelClient client, String indexName) throws IOException {
        boolean exists = exists(client, indexName);
        if (!exists) {
            //不存在就结束
            return;
        }
        //索引存在，就执行删除
        long s = System.currentTimeMillis();
        ;
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        request.timeout(Timevalue.timevalueMinutes(2));
        request.timeout("2m");

        AcknowledgedResponse delete = client.indices().delete(request, RequestOptions.DEFAULT);
        long t = System.currentTimeMillis();
        //计算删除耗时
        System.out.println("删除索引耗时ms: " + (t - s));
    }

public static void createIndex(RestHighLevelClient client, String indexName) throws IOException {

        CreateIndexRequest request = new CreateIndexRequest(indexName);
        request.settings(setting());
        request.mapping(me.getKey(), me.getValue());

        client.indices().create(request, RequestOptions.DEFAULT);

    }

spring boot 操作es ： 点击这里