Elasticsearch API

boolean model

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
GET /product/_search
{
    "query": {
        "match": {
            "name": {
                "query": "my good",
                "operator": "or",   //可选：or、and  默认是or
                "minimum_should_match": 2, //最少匹配2个此项
                "boost"：1   //相关度算法权重
            }
        }
    }
}

//所有相关查询，ES底层会转换成bool操作
GET /product/_search
{
    "query": {
        "bool": {    
            "should": [  //and的话，这里就是must, 可选：must/must not/should
              {"term":{"name":"my"}},
              {"term":{"name":"good"}}
            ],
            "boost": 1,
            "minimum_should_match": 2
        }
    }
}

JAVA API

Transport Client：TransportClient不推荐使用，而推荐使用Java High Level REST Client，并将在Elasticsearch 8.0中删除。
Java Low Level REST Client：低级别的REST客户端，通过http与集群交互，用户需自己编组请求JSON串，及解析响应JSON串。兼容所有ES版本
Java High Level REST Client：高级别的REST客户端，基于低级别的REST客户端，增加了编组请求JSON串、解析响应JSON串等相关api。使用的版本需要保持和ES服务端的版本一致，否则会有版本问题。

1
2
3
4
5
6
7
8
9
10
11
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>transport</artifactId>
    <version>7.10.1</version>
</dependency>

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.10.1</version>
</dependency>

Transport Client

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public static void main(String[] args) {
    //配置集群名称
    Settings settings = Settings.builder().put("cluster.name", "elasticsearch").build();
    //创建连接
    //TransportClient client = new PreBuiltTransportClient(Settings.EMPTY)
    TransportClient client = new PreBuiltTransportClient(settings)
        .addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9300))//通讯端口  而不是服务端口
        .addTransportAddress(new TransportAddress(InetAddress.getByName("localhost"), 9301));

    //添加数据
    create(client);
    //关闭连接
    client.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private void create(TransportClient client) {
    //查询数据
    List<Product> list = service.list();
    for (Product item : list) {
       	//固定语法
        IndexResponse response = client.prepareIndex("product2", "_doc", item.getId().toString())
            .setSource(XContentFactory.jsonBuilder()
                       .startObject()
                       .field("name", item.getName())
                       .field("desc", item.getDesc())
                       .field("price", item.getPrice())
                       .field("date", item.getCreateTime().toLocalDateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd")))
                       .field("tags", item.getTags().split(","))
                       .endObject())
            .get();
        System.out.println(response.getResult());
    }
}
1
2
3
4
5
6
7
8
9
10
11
//查询一条数据
private void get(TransportClient client) {
    GetResponse response = client.prepareGet("product", "_doc", "1").get();
    String index = response.getIndex();//获取索引名称
    String type = response.getType();//获取索引类型
    String id = response.getId();//获取索引id
    System.out.println("index:" + index);
    System.out.println("type:" + type);
    System.out.println("id:" + id);
    System.out.println(response.getSourceAsString());
}
1
2
3
4
5
6
7
8
9
10
11
//批量查询
private void getAll(TransportClient client) {
    SearchResponse response = client.prepareSearch("product")
        .get();
    SearchHits searchHits = response.getHits();
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String res = hit.getSourceAsString();
        System.out.println("res" + res);
    }
}
1
2
3
4
5
6
7
8
9
10
//增量更新
private void update(TransportClient client) {
    UpdateResponse response = client.prepareUpdate("product", "_doc", "2")
        .setDoc(XContentFactory.jsonBuilder()
                .startObject()
                .field("name", "update name")
                .endObject())
        .get();
    System.out.println(response.getResult());
}
1
2
3
4
5
//删除数据
private void delete(TransportClient client) {
    DeleteResponse response = client.prepareDelete("product", "_doc", "2").get();
    System.out.println(response.getResult());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//多条件查询
public void multiSearch(TransportClient client) {
    //SearchResponse：返回的结果集
    SearchResponse response = client.prepareSearch("product2")
        .setQuery(QueryBuilders.termQuery("name", "xiaomi"))                 // Query
        .setPostFilter(QueryBuilders.rangeQuery("price").from(0).to(4000))
        .setFrom(1).setSize(3)
        .get();
    SearchHits searchHits = response.getHits();
    SearchHit[] hits = searchHits.getHits();
    for (SearchHit hit : hits) {
        String res = hit.getSourceAsString();
        System.out.println("res" + res);
    }
    client.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
//统计每个月份的tags中每个词项term的平均价格
GET /product/_search
{
    "aggs": {
        "group_by_month": {
            "date_histogram": {
                "field": "date",
                "calendar_interval": "month"
            },
            "aggs": {
                "by_tags": {
                    "terms": {
                        "field": "tags.keyword"
                    },
                    "aggs": {
                        "avg_price": {
                            "avg": {
                                "field": "price"
                            }
                        }
                    }
                }
            }
        }
    },
    "size": 0
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
//查询结果
{
    "aggregations" : {
        "group_by_month" : {
            "buckets" : [
                {
                    "key_as_string" : "2020-04-01T00:00:00.000Z",
                    "key" : 1585699200000,
                    "doc_count" : 3,
                    "by_tags" : {
                        "doc_count_error_upper_bound" : 0,
                        "sum_other_doc_count" : 0,
                        "buckets" : [
                            {
                                "key" : "bufangshui",
                                "doc_count" : 1,
                                "avg_price" : {
                                    "value" : 999.0
                                }
                            },
                            {
                                "key" : "fashao",
                                "doc_count" : 1,
                                "avg_price" : {
                                    "value" : 2999.0
                                }
                            }
                        ]
                    }
                },
                {
                    "key_as_string" : "2020-05-01T00:00:00.000Z",
                    "key" : 1588291200000,
                    "doc_count" : 2,
                    "by_tags" : {
                        "doc_count_error_upper_bound" : 0,
                        "sum_other_doc_count" : 0,
                        "buckets" : [
                            {
                                "key" : "fashao",
                                "doc_count" : 2,
                                "avg_price" : {
                                    "value" : 4499.0
                                }
                            },
                            {
                                "key" : "xingjiabi",
                                "doc_count" : 2,
                                "avg_price" : {
                                    "value" : 4499.0
                                }
                            }
                        ]
                    }
                }
            ]
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
//通过Java代码实现上面案例的聚合查询
public void aggSearch(TransportClient client) {
    //region 1->计算并返回聚合分析response对象
    SearchResponse response = client.prepareSearch("product")
        .addAggregation(
        AggregationBuilders.dateHistogram("group_by_month")
        .field("date")
        .calendarInterval(DateHistogramInterval.MONTH)
        //.dateHistogramInterval(DateHistogramInterval.MONTH)
        .subAggregation(
            AggregationBuilders
            .terms("by_tag")
            .field("tags.keyword")
            .subAggregation(
                AggregationBuilders
                .avg("avg_price")
                .field("price")
            )
        )
    )
        .execute().actionGet();  //执行查询

    //region 2->输出结果信息
    SearchHit[] hits = response.getHits().getHits();
    //获取aggregations结果
    Map<String, Aggregation> map = response.getAggregations().asMap();
    //获取group_by_month结果
    Aggregation group_by_month = map.get("group_by_month");
    Histogram dates = (Histogram) group_by_month;
    //转换成分组的桶
    Iterator<Histogram.Bucket> buckets = (Iterator<Histogram.Bucket>) dates.getBuckets().iterator();

    while (buckets.hasNext()) {
        //遍历桶中的数据
        Histogram.Bucket dateBucket = buckets.next();
        System.out.println("\n\n月份：" + dateBucket.getKeyAsString() + "\n计数：" + dateBucket.getDocCount());
        //获取桶里面的桶by_tag
        Aggregation group_by_tag = dateBucket.getAggregations().asMap().get("by_tag");
        StringTerms terms = (StringTerms) group_by_tag;
        Iterator<StringTerms.Bucket> tagsBucket = terms.getBuckets().iterator();

        while (tagsBucket.hasNext()) {

            //遍历by_tag桶，获取桶中的平均价格
            StringTerms.Bucket tagBucket = tagsBucket.next();
            System.out.println("\t标签名称：" + tagBucket.getKey() + "\n\t数量：" + tagBucket.getDocCount());
            Aggregation avg_price = tagBucket.getAggregations().get("avg_price");
            Avg avg = (Avg) avg_price;
            System.out.println("\t平均价格：" + avg.getValue() + "\n");
        }
    }
    //endregion

    client.close();
}

High Level REST Client

1
2
3
4
常见步骤：
1.创建对应的Request对象
2.创建对应的Builder对象/编写对应操作
3.Client执行对应的方法

创建连接

1
2
3
4
5
6
7
//使用 RestHighLevelClient
public void createConnection {
    RestHighLevelClient client = new RestHighLevelClient(
        RestClient.builder(
            new HttpHost("localhost", 9200, "http"),
            new HttpHost("localhost", 9201, "http")));  //使用的是9200服务端口
}

创建索引

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void createIndex(RestHighLevelClient client) {
    CreateIndexRequest request = new CreateIndexRequest("test_index");

    request.settings(Settings.builder()
                     .put("index.number_of_shards", 3)
                     .put("index.number_of_replicas", 2)
                    );
    CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
    if (createIndexResponse.isAcknowledged()) {
        System.out.println("创建index成功!");
    } else {
        System.out.println("创建index失败!");
    }

    client.close();
}

查询索引

1
2
3
4
5
6
7
8
9
public void getIndex(RestHighLevelClient client) {
    GetIndexRequest request = new GetIndexRequest("test_index*");
    GetIndexResponse response = client.indices().get(request, RequestOptions.DEFAULT);
    String[] indices = response.getIndices();
    for (String indexName : indices) {
        System.out.println("index name:" + indexName);
    }
    client.close();
}

删除索引

1
2
3
4
5
6
7
8
9
10
public void delIndex(RestHighLevelClient client) {
    DeleteIndexRequest request = new DeleteIndexRequest("test_index");
    AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT);
    if (response.isAcknowledged()) {
        System.out.println("删除index成功!");
    } else {
        System.out.println("删除index失败!");
    }
    client.close();
}

插入数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public void insertData(RestHighLevelClient client) {
    List<Product> list = service.list();

    //插入数据，index不存在则自动根据匹配到的template创建。index没必要每天创建一个，如果是为了灵活管理，最低建议每月一个 yyyyMM。
    IndexRequest request = new IndexRequest("test_index");
    //最好不要自定义id 会影响插入速度。
    Product product = list.get(0);
    Gson gson = new Gson();
    request.id(product.getId().toString());
    request.source(gson.toJson(product), XContentType.JSON);
    IndexResponse response = client.index(request, RequestOptions.DEFAULT);
    System.out.println(response);
    client.close();
}

批量插入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void batchInsertData(RestHighLevelClient client) {
    //批量插入数据，更新和删除同理
    BulkRequest request = new BulkRequest("test_index");
    Gson gson = new Gson();
    Product product = new Product();
    product.setPrice(3999.00);
    product.setDesc("xioami");
    for (int i = 0; i < 10; i++) {
        product.setName("name" + i);
        request.add(new IndexRequest().source(gson.toJson(product), XContentType.JSON));
    }
    BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
    System.out.println("数量:" + response.getItems().length);
    client.close();
}

查询数据

1
2
3
4
5
6
7
8
9
10
11
12
13
//通过ID查询数据
public void getById(RestHighLevelClient client) {
    //注意 这里查询使用的是别名。
    GetRequest request = new GetRequest("test_index", "PPWhwnIBRwX67j4bTmV1");
    String[] includes = {"name", "price"};
    String[] excludes = {"desc"};
    FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
    //只查询特定字段。如果需要查询所有字段则不设置该项。
    request.fetchSourceContext(fetchSourceContext);
    GetResponse response = client.get(request, RequestOptions.DEFAULT);
    System.out.println(response);
    client.close();
}

删除数据

1
2
3
4
5
6
7
//通过ID删除数据
public void delById(RestHighLevelClient client) throws IOException {
    DeleteRequest request = new DeleteRequest("test_index", "1");
    DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
    System.out.println(response);
    client.close();
}

批量查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public void multiGetById(RestHighLevelClient client) throws IOException {
    //多个根据id查询
    MultiGetRequest request = new MultiGetRequest();
    request.add("test_index", "PPWhwnIBRwX67j4bTmV1");
    //两种写法
    request.add(new MultiGetRequest.Item(
        "test_index",
        "PfWhwnIBRwX67j4bTmV1"));
    MultiGetResponse response = client.mget(request, RequestOptions.DEFAULT);
    for (MultiGetItemResponse itemResponse : response) {
        System.out.println(itemResponse.getResponse().getSourceAsString());
    }
    client.close();
}

查询更新

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//更新查询
public void updateByQuery(RestHighLevelClient client) throws IOException {
    UpdateByQueryRequest request = new UpdateByQueryRequest("test_index");
    //默认情况下，版本冲突会中止 UpdateByQueryRequest 进程，但是你可以用以下命令来代替
    //设置版本冲突继续
    //        request.setConflicts("proceed");
    //设置更新条件
    request.setQuery(QueryBuilders.matchQuery("name","name1 name3"));
    //        //限制更新条数
    //        request.setMaxDocs(10);
    request.setScript(
        new Script(ScriptType.INLINE, "painless", "ctx._source.desc+='#';", Collections.emptyMap()));
    BulkByScrollResponse response = client.updateByQuery(request,RequestOptions.DEFAULT);
    System.out.println(response);
    client.close();
}

分页查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public ResultDto carInfo(String keyword,Integer from,Integer size) {
    SearchRequest searchRequest = new SearchRequest("product");
    
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.matchQuery("name", keyword));
    searchSourceBuilder.from(from);
    searchSourceBuilder.size(size);
    searchRequest.source(searchSourceBuilder);
    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
    //ESClient.getInstance().closeClient();
    
    ResultDto res = new ResultDto();
    res.setData(searchResponse.getHits().getHits());
    return res;
}

Scroll查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public ResultDto scroll(String scrollId) {
    SearchRequest searchRequest = new SearchRequest("product");
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.size(2); //每页两条
    searchRequest.source(searchSourceBuilder);

    searchRequest.scroll(TimeValue.timeValueMinutes(1L));  //scrollID的有效期1分钟

    SearchResponse searchResponse = scrollId == null
        ? client.search(searchRequest, RequestOptions.DEFAULT)
        : client.scroll(new SearchScrollRequest(scrollId), RequestOptions.DEFAULT);
    scrollId = searchResponse.getScrollId();
    SearchHits hits = searchResponse.getHits();
    
    ResultDto res = new ResultDto();
    res.setTag(scrollId);
    res.setData(hits.getHits());
    return res;
}

批量操作

1
2
3
4
5
6
7
8
9
10
11
public ResultDto bulk() {
    SearchRequest searchRequest = new SearchRequest("msb_auto");
    BulkRequest request = new BulkRequest();
    request.add(new DeleteRequest("product", "13"));  //删除doc
    request.add(new UpdateRequest("product", "22")
                .doc(XContentType.JSON, "name", "天籁之音"));  //修改doc一个字段
    request.add(new IndexRequest("product").id("4")
                .source(XContentType.JSON, "myname", "天津一汽")); //修改整doc
    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
    return null;
}

搜索模板

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public ResultDto templateSearch() {

    //region 创建模板并缓存 作用域为整个集群
    Request scriptRequest = new Request("POST", "_scripts/test_template_search");
    scriptRequest.setJsonEntity(
        "{" +
        "  \"script\": {" +
        "    \"lang\": \"mustache\"," +
        "    \"source\": {" +
        "      \"query\": { \"match\" : { \"{{field}}\" : \"{{value}}\" } }," +
        "      \"size\" : \"{{size}}\"" +
        "    }" +
        "  }" +
        "}");
    Response scriptResponse = client.getLowLevelClient().performRequest(scriptRequest);
    
    
    SearchTemplateRequest request = new SearchTemplateRequest();
    request.setRequest(new SearchRequest("msb_auto"));
    //STORED 保存到内存模板
    request.setScriptType(ScriptType.STORED);
    request.setScript("test_template_search");
    // 本地模板
  
    //request.setScriptType(ScriptType.INLINE);
    //    request.setScript(
    //                    "{\n" +
    //                    "  \"from\": {{from}},\n" +
    //                    "  \"size\": {{size}},\n" +
    //                    "  \"query\": {\n" +
    //                    "    \"match\": {\n" +
    //                    "      \"master_brand_name\": \"{{master_brand_name}}\"\n" +
    //                    "    }\n" +
    //                    "  }\n" +
    //                    "}");
    
    Map<String, Object> scriptParams = new HashMap<>();
    scriptParams.put("field", "master_brand_name");
    scriptParams.put("value", "一汽");
    scriptParams.put("size", 5);
    request.setScriptParams(scriptParams);
    
    //查询
    SearchTemplateResponse response = client.searchTemplate(request, RequestOptions.DEFAULT);
    return null;
}

模糊查询

1
2
3
4
5
6
7
8
public SearchHit[] fuzzy(String name) {
    SearchRequest searchRequest = new SearchRequest("product");
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.fuzzyQuery("brand_name.keyword", name).fuzziness(Fuzziness.AUTO));
    searchRequest.source(searchSourceBuilder);
    SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
    return response.getHits().getHits();
}

多语句查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public ResultDto multiSearch() {
    //1.创建一个空的MultiSearchRequest
    MultiSearchRequest request = new MultiSearchRequest();
  
    //2.创建SearchRequest,并填充查询条件
    SearchRequest firstSearchRequest = new SearchRequest("product");
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.matchQuery("series_name", "朗动"));
    firstSearchRequest.source(searchSourceBuilder);
    //3.将SearchRequest装配到MultiSearchRequest中
    request.add(firstSearchRequest);
  
    //4.创建第二个SearchRequest并重复第三步
    SearchRequest secondSearchRequest = new SearchRequest("product");
    searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.matchQuery("series_name", "揽胜运动"));
    secondSearchRequest.source(searchSourceBuilder);
    request.add(secondSearchRequest);

    MultiSearchResponse response = client.msearch(request, RequestOptions.DEFAULT);
    return null;
}

Bool查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public ResultDto boolSearch() {
    MultiSearchRequest request = new MultiSearchRequest();

    SearchRequest searchRequest = new SearchRequest("product");
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query
        (
        QueryBuilders.boolQuery()
        .must(matchPhraseQuery("sale_name", "2018款"))
        .filter(matchQuery("master_brand_name", "大众").analyzer("ik_max_word"))
        .mustNot(matchQuery("series_name", "速腾"))
    );
    searchRequest.source(searchSourceBuilder);
    request.add(searchRequest);
    MultiSearchResponse response = client.msearch(request, RequestOptions.DEFAULT);
    return null;
}

Sniffer 嗅探器

概念：从运行中的Elasticsearch集群自动发现节点并将它们设置为现有RestClient实例（low实例）。

版本：从ES 2.X开始及更高版本支持。

1
2
3
4
5
6
7
8
9
10
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-client-sniffer</artifactId>
    <version>7.10.1</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-client</artifactId>
    <version>7.10.1</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
用法：
1.创建RestClient：RestClientBuilder
2.HTTPS支持：NodesSniffer
3.绑定：Sniffer.builder(RestClient)
4.监听：SniffOnFailureListener

相关设置参数：
1.setSniffIntervalMillis：每隔多久触发一次嗅探，单位毫秒，默认30000（5分钟）。
2.setSniffAfterFailureDelayMillis：嗅探失败时触发嗅探一次,经过设置的时间（默认1min）之后再次嗅探直至正常。若无绑定监听器则无效。
3.setFailureListener：设置用于监听嗅探失败的监听器，当监听到普通嗅探失败，则通知Sniffer实例进行新一轮的嗅探，并更新节点。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
public class ESClient {

    private static ESClient ESClient;
    private String host = "localhost:9200,localhost:9201";
    private RestClientBuilder builder;
    private static Sniffer sniffer;
    private static RestHighLevelClient highClient;

    private ESClient(){
    }

    public static ESClient getInstance() {
        if (ESClient == null) {
            synchronized (ESClient.class) {
                if (ESClient == null) {
                    ESClient = new ESClient();
                    ESClient.initBuilder();
                }
            }
        }
        return ESClient;
    }

    public RestClientBuilder initBuilder() {
        String[] hosts = host.split(",");
        HttpHost[] httpHosts = new HttpHost[hosts.length];
        for (int i = 0; i < hosts.length; i++) {
            String[] host = hosts[i].split(":");
            httpHosts[i] = new HttpHost(host[0], Integer.parseInt(host[1]), "http");
        }

        builder = RestClient.builder(httpHosts);

        //region 在Builder中设置请求头
        //  1.设置请求头
        Header[] defaultHeaders = new Header[]{
            new BasicHeader("Content-Type", "application/json")
        };
        builder.setDefaultHeaders(defaultHeaders);
        //endregion
        //RestClient restClient = builder.build();

        //启用嗅探器
        //SniffOnFailureListener sniffOnFailureListener = new SniffOnFailureListener();
        //builder.setFailureListener(sniffOnFailureListener);
        //sniffer = Sniffer.builder(restClient)
        //        .setSniffIntervalMillis(5000)
        //        .setSniffAfterFailureDelayMillis(10000)
        //        .build();
        //sniffOnFailureListener.setSniffer(sniffer);
        return builder;
    }

    public RestHighLevelClient getHighLevelClient() {
        if (highClient == null) {
            synchronized (ESClient.class) {
                if (highClient == null) {
                    //gighClient是通过lowClient包装产生的
                    highClient = new RestHighLevelClient(builder);
                    //设置嗅探器
                    //十秒刷新并更新一次节点
                    sniffer = Sniffer.builder(highClient.getLowLevelClient())
                        .setSniffIntervalMillis(5000)
                        .setSniffAfterFailureDelayMillis(15000)
                        .build();
                }
            }
        }
        return highClient;
    }
    /**
     *
     * 关闭sniffer client
     */
    public void closeClient() {
        if (null != highClient) {
            try {
                sniffer.close();    //需要在highClient close之前操作
                highClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void snifferTest() {
    RestHighLevelClient client = ESClient.getInstance().getHighLevelClient();
    Iterator<Node> nodes = client.getLowLevelClient().getNodes().iterator();
    while (nodes.hasNext()) {
        Node node = nodes.next();
        System.out.println(node);
    }
    Thread.sleep(5000);
    System.out.println("5秒后:");
    nodes = client.getLowLevelClient().getNodes().iterator();
    while (nodes.hasNext()) {
        Node node = nodes.next();
        System.out.println(node);
    }
    ESClient.getInstance().closeClient();
}

假设有5台机器，IP分别是【9200 - 9204】最后嗅探造成的输出结果是：

1
2
3
4
5
6
7
8
9
先输出:
host=[http://localhost:9200]
host=[http://localhost:9201]
间隔五秒后输出：
host=[http://localhost:9200]
host=[http://localhost:9201]
host=[http://localhost:9202]
host=[http://localhost:9203]
host=[http://localhost:9204]

嗅探器它能够给我们监控动态更新集群的节点数量，避免了程序因节点过多或增加/减少节点造成的切换问题。

最后更新： 2020年12月16日 10:32

原始链接： https://midkuro.gitee.io/2020/12/10/elasticearch-api/

赏
× 请我吃糖~
打赏二维码