Elasticsearch-06-Elasticsearch Java API Client

港控/mmm° 2024-02-20 08:54 78阅读 0赞

## 前言 ##

### 简介 ###

在 Elasticsearch7.15版本之后，Elasticsearch官方将它的**高级客户端 RestHighLevelClient**标记为弃用状态。同时推出了全新的 Java API客户端 **Elasticsearch Java API Client**，该客户端也将在 Elasticsearch8.0及以后版本中成为官方推荐使用的客户端。

Elasticsearch Java API Client 支持除 Vector tile search API 和 Find structure API 之外的所有 Elasticsearch API。且支持所有API数据类型，并且不再有原始JsonValue属性。它是针对Elasticsearch8.0及之后版本的客户端，所以我们需要学习新的Elasticsearch Java API Client的使用方法。

### 为什么要抛弃High Level Rest： ###

*  客户端"too heavy"，相关依赖超过 30 MB，且很多都是非必要相关的；api 暴露了很多服务器内部接口
 *  一致性差，仍需要大量的维护工作。
 *  客户端没有集成 json/object 类型映射，仍需要自己借助字节缓存区实现。

### Java API Client最明显的特征： ###

*  支持lambda表达式操作ES
 *  支持Builder建造者模式操作ES，链式代码具有较强可读性.
 *  应用程序类能够自动映射为Mapping.
 *  所有Elasticsearch API的强类型请求和响应。
 *  所有API的阻塞和异步版本
 *  将协议处理委托给http客户端（如Java低级REST客户端），该客户端负责处理所有传输级问题：HTTP连接池、重试、节点发现等。

### 官方地址 ###

https://www.elastic.co/guide/en/elasticsearch/client/java-api-client/7.17/indexing.html

## 简单使用 ##

### 1：导包 ###

这里记住你的elasticsearch-java必须对应你电脑上装的ES版本

<dependency>
          <groupId>co.elastic.clients</groupId>
          <artifactId>elasticsearch-java</artifactId>
          <version>7.17.6</version>
        </dependency>
    
        <dependency>
          <groupId>com.fasterxml.jackson.core</groupId>
          <artifactId>jackson-databind</artifactId>
          <version>2.12.3</version>
        </dependency>
        <dependency>
          <groupId>jakarta.json</groupId>
          <artifactId>jakarta.json-api</artifactId>
          <version>2.0.1</version>
        </dependency>

### 2:开启链接 ###

//创建一个低级的客户端
            final RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
            //创建JSON对象映射器
            final RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
            //创建API客户端
            final ElasticsearchClient client = new ElasticsearchClient(transport);

### 3:关闭链接 ###

client.shutdown();
            transport.close();
            restClient.close();

### 4：完整代码 ###

public class Client {
        
        public static void main(String[] args) throws IOException {
        
            //创建一个低级的客户端
            final RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200)).build();
            //创建JSON对象映射器
            final RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
            //创建API客户端
            final ElasticsearchClient client = new ElasticsearchClient(transport);
            //查询所有索引-------------------------------------------------------------------------------------
            final GetIndexResponse response = client.indices().get(query -> query.index("_all"));
            final IndexState products = response.result().get("products");
            System.out.println(products.toString());
            //关闭
            client.shutdown();
            transport.close();
            restClient.close();
        }
    }

## JsonData类 ##

原始JSON值。可以使用JsonpMapper将其转换为JSON节点树或任意对象。 此类型在API类型中用于没有静态定义类型或无法表示为封闭数据结构的泛型参数的值。 API客户端返回的此类实例保留对客户端的JsonpMapper的引用，并且可以使用to（class）转换为任意类型，而不需要显式映射器

我们一般在ES的DSL范围查询中会使用到!  
核心方法:

*  to:将此对象转换为目标类。必须在创建时提供映射器
 *  from:从读取器创建原始JSON值
 *  of:从现有对象创建原始JSON值，以及用于进一步转换的映射器
 *  deserialize:使用反序列化程序转换此对象。必须在创建时提供映射器

## 高阶使用 ##

### 1：ES配置类 ###

// 配置的前缀
    @ConfigurationProperties(prefix = "elasticsearch") 
    @Configuration
    public class ESClientConfig {
        
    
    	/**
    	 * 多个IP逗号隔开
    	 */
    	@Setter
    	private String hosts;
    
    	/**
    	 * 同步方式
    	 * 
    	 * @return
    	 */
    	@Bean
    	public ElasticsearchClient elasticsearchClient() {
        
    		HttpHost[] httpHosts = toHttpHost();
    		// Create the RestClient 
    		//RestClient restClient = RestClient.builder(httpHosts).build();
    		RestClient restClient = RestClient.builder(httpHosts)
                    .setHttpClientConfigCallback(httpClientBuilder
                            ->httpClientBuilder.setDefaultHeaders(
                                    listOf(new BasicHeader(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.toString())))
                            .addInterceptorLast((HttpResponseInterceptor) (response, context)
                                    -> response.addHeader("X-Elastic-Product", "Elasticsearch"))).build();
    		// Create the transport with a Jackson mapper
    		RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
    		// create the API client
    		return new ElasticsearchClient(transport);
    	}
    
    	/**
    	 * 异步方式
    	 * 
    	 * @return
    	 */
    	@Bean
    	public ElasticsearchAsyncClient elasticsearchAsyncClient() {
        
    		HttpHost[] httpHosts = toHttpHost();
    		RestClient restClient = RestClient.builder(httpHosts).build();
    		RestClientTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
    		return new ElasticsearchAsyncClient(transport);
    	}
    
    	/**
    	 * 解析配置的字符串hosts，转为HttpHost对象数组
    	 *
    	 * @return
    	 */
    	private HttpHost[] toHttpHost() {
        
    		if (!StringUtils.hasLength(hosts)) {
        
    			throw new RuntimeException("invalid elasticsearch configuration. elasticsearch.hosts不能为空！");
    		}
    
    		// 多个IP逗号隔开
    		String[] hostArray = hosts.split(",");
    		HttpHost[] httpHosts = new HttpHost[hostArray.length];
    		HttpHost httpHost;
    		for (int i = 0; i < hostArray.length; i++) {
        
    			String[] strings = hostArray[i].split(":");
    			httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), "http");
    			httpHosts[i] = httpHost;
    		}
    
    		return httpHosts;
    	}
    
    }

### 2：查询所有索引 ###

//省略连接...
            final GetIndexResponse all = client.indices().get(query -> query.index("_all"));
            System.out.println(all.toString());
    //省略关闭...

### 3：查询某个索引 ###

//查询某个索引
            final GetIndexResponse products = client.indices().get(query -> query.index("products"));
            System.err.println(products.toString());

### 4:创建索引 ###

//查询某个索引是否存在
            boolean exists = client.indices().exists(query -> query.index("products")).value();
            System.out.println(exists);
            if (exists) {
        
                System.err.println("索引已存在");
    
            } else {
        
                final CreateIndexResponse products = client.indices().create(builder -> builder.index("products"));
                System.err.println(products.acknowledged());
            }

### 5:删除指定索引 ###

//删除指定索引
            boolean exists = client.indices().exists(query -> query.index("products")).value();
            System.out.println(exists);
            if (exists) {
        
                DeleteIndexResponse response = client.indices().delete(query -> query.index("products"));
                System.err.println(response.acknowledged());
            } else {
        
                System.err.println("索引不存在");
            }

### 6:查询索引的映射 ###

//查询映射信息
            final GetIndexResponse response = client.indices().get(builder -> builder.index("produces"));
            System.err.println(response.result().get("produces").mappings());

### 7:创建索引指定映射 ###

numberOfReplicas(“1”):设置副本  
numberOfShards(“1”):设置分片

//创建索引指定映射,分片和副本信息
            final CreateIndexResponse response = client.indices().create(builder ->
                    builder.settings(indexSetting -> indexSetting.numberOfReplicas("1").numberOfShards("1")).mappings(
                            map -> map
                                    .properties("name", propertyBuilder -> propertyBuilder.keyword(keywordProperty -> keywordProperty))
                                    .properties("price", propertyBuilder -> propertyBuilder.double_(doubleNumProperty -> doubleNumProperty))
                                    .properties("des", propertyBuilder -> propertyBuilder.text(textProperty -> textProperty.analyzer("ik_smart").searchAnalyzer("ik_smart")))
                    ).index("produces")
    
            );

### 8:创建文档 ###

#### 使用HashMap作为数据存储容器 ####

//创建文档
            //1.创建HashMap进行存储数据，文档要对应映射
            final HashMap<String, Object> doc = new HashMap<>();
            doc.put("name","辣条");
            doc.put("age",12);
            doc.put("id","11111");
            //2.将文档存入索引中
            final IndexResponse response = client.index(builder -> builder.index("produces").id(doc.get("id")).document(doc));
            System.err.println(response.version());

#### 使用自定义类作为数据存储容器 ####

实体类

@Data
    @AllArgsConstructor
    @NoArgsConstructor
    @ToString
    public class Produce {
        
        private String id;
        private String name;
        private double age;
    }

//创建文档
            final Produce produce = new Produce("123", "小明", 18);
            final IndexResponse response = client.index(builder -> builder.index("produces").id(produce.getId()).document(produce));
            System.err.println(response.version());

#### 使用外部JSON数据创建 ####

这里要注意我们需要使用StringReader进行读取时使用replace函数将设置的’改为",当然这在真实的业务中肯定不会有,因为真实业务中一定是标准的JSON数据,无需使用replace进行替换了

//创建文档
            final StringReader input = new StringReader(
                    "{'name':'农夫三拳','price':3.00,'des':'农夫三拳有点甜'}".replace('\'', '"')
            );
            final IndexResponse response = client.index(builder -> builder.index("produces").id("44514").withJson(input));
            System.err.println(response.version());

### 9: 查询所有文档 ###

final SearchResponse<Object> response = client.search(builder -> builder.index("produces"), Object.class);
            final List<Hit<Object>> hits = response.hits().hits();
            hits.forEach(
                    x-> System.err.println(x)
            );

### 10:根据ID查询文档 ###

使用HashMap对应查询

//查询文档
            final GetResponse<Map> response = client.get(builder -> builder.index("produces").id("116677"), Map.class);
            final Map source = response.source();
            source.forEach((x,y)->{
        
                System.err.println(x+":"+y);
            });

使用自定义类对应查询

final GetResponse<Produce> response1 = client.get(builder -> builder.index("produces").id("aabbcc123"), Produce.class);
            final Produce source1 = response1.source();
            System.err.println(source1.toString());

### 11:删除文档 ###

### 12:修改文档 ###

#### 全覆盖 ####

//修改文档（覆盖）
            final Produce produce = new Produce("ccaabb123", "旺仔摇滚洞", "旺仔摇滚洞乱摇乱滚", 10.23D);
            final UpdateResponse<Produce> response = client.update(builder -> builder.index("produces").id("aabbcc123").doc(produce), Produce.class);
            System.err.println(response.shards().successful());

#### 修改部分文档 ####

区别在于我们需要设置.docAsUpsert(true)表明是修改部分而不是覆盖

//修改文档(部分修改)
    //        final Produce produce = new Produce("ccaabb123", "旺仔摇滚洞", "旺仔摇滚洞乱摇乱滚", 10.23D);
            final Produce produce = new Produce();
            produce.setName("旺仔摇不动");
            final UpdateResponse<Produce> response = client.update(builder -> builder.index("produces").id("aabbcc123").doc(produce).docAsUpsert(true), Produce.class);
            System.err.println(response.shards().successful());

### 13:批量操作 ###

#### 批量新增 ####

produceList.add(produce1);
            produceList.add(produce2);
            produceList.add(produce3);
            //构建BulkRequest
            final BulkRequest.Builder br = new BulkRequest.Builder();
            for (Produce produce : produceList) {
        
                br.operations(op->op.index(idx->idx.index("produces").id(produce.getSku()).document(produce)));
            }
            final BulkResponse response = client.bulk(br.build());

#### 批量删除 ####

List<BulkOperation> bulkOperations = new ArrayList<>();
                // 向集合中添加需要删除的文档id信息
                for (int i = 0; i < dto.getIds().size(); i++) {
        
                    int finalI = i;
                    bulkOperations.add(BulkOperation.of(b -> b
                            .delete((d -> d
                                    .index(dto.getIndex())
                                    .id(dto.getIds().get(finalI))
                            ))
                    ));
                }
                // 调用客户端的bulk方法，并获取批量操作响应结果
                BulkResponse response = client
                        .bulk(e -> e
                                .index(dto.getIndex())
                                .operations(bulkOperations));

## DSL查询 ##

### 1：matchAll查询所有文档 ###

//matchAll
            final SearchResponse<Produce> response = client.search(builder ->
                    builder.index("produces")
                            .query(q ->
                                    q.matchAll(
                                            v->v
                                    )), Produce.class);
    
            System.err.println(response.hits().hits());

### 2:match 根据字段查询 ###

//简单query方式查询
            final SearchResponse<Produce> response = client.search(builder ->
                    builder.index("produces")
                            .query(q ->
                                    q.match(t ->
                                            t.field("name")
                                                    .query("龙虎万精油"))), Produce.class);
            System.err.println(response.hits().hits());

### 3:多id查询 ###

//多ID查询
            final SearchResponse<Produce> response = client.search(builder ->
                    builder.index("produces")
                            .query(q ->
                                    q.ids(sid->sid.values("1000","1001"))), Produce.class);
            System.err.println(response.hits().hits());

### 4:term 不分词查询 ###

//term不分词条件查询
            final SearchResponse<Produce> response = client.search(builder -> builder.index("produces")
            .query(q -> q.term(t -> t.field("name").value("风油精"))), Produce.class);
            System.err.println(response.hits().hits());

### 5:范围查询 ###

//范围查询
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q ->
                                    q.range(r ->
                                            r.field("price").gt(JsonData.of(5D)).lt(JsonData.of(15D)))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 6: 前缀查询 ###

final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q ->q.prefix(p->p.field("name").value("六"))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 7:匹配查询 ###

#### //匹配查询 ####

final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q ->q.wildcard(w->w.field("name").value("风*"))),
                    Produce.class);
            System.err.println(response.hits().hits());

#### ?单字符匹配 ####

//匹配查询
            final SearchResponse<Produce> response = client.search(builder ->
            builder.index("produces").query(q ->q.wildcard(w->w.field("name").value("风?精"))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 8:模糊查询 ###

//模糊查询
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q ->q.fuzzy(f->f.field("name").value("六仙花露水"))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 9:多条件查询 ###

使用bool关键字配合must,should,must\_not

*  must:所有条件必须同时成立
 *  must\_not:所有条件必须同时不成立
 *  should:所有条件中成立一个即可

//多条件
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q ->
                                    q.bool(b ->
                                            b.must(t ->
                                                    t.term(v ->
                                                            v.field("name")
                                                                    .value("旺仔摇不动")))
                                             .must(t2 ->
                                                     t2.term(v2 ->
                                                            v2.field("price")
                                                                     .value(0.0D))))),
                    Produce.class);
            System.err.println(response.hits().hits());

或者创建BoolQuery.Builder，以便进行业务判断是否增加查询条件

List<FieldValue> fieldValues = new ArrayList<>();
            fieldValues.add(FieldValue.of(10));
            fieldValues.add(FieldValue.of(100));
            BoolQuery.Builder boolQuery = new BoolQuery.Builder();
    
            boolQuery.must(t->
                    t.terms(v->
                            v.field("label")
                                    .terms(term->
                                            term.value(fieldValues))));
            boolQuery.must(t->
                    t.match(f->
                            f.field("name")
                                    .query("旺仔")));
    
    
            SearchResponse<Object> search = elasticsearchClient.search(builder -> builder.index("my_test_index")
                    .query(q->
                            q.bool(boolQuery.build())),Object.class);

### 10：多字段查询-multiMatch ###

//多字段查询
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces").query(q->q.multiMatch(qs->qs.query("蚊虫叮咬 辣眼睛").fields("name","des"))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 11:高亮显示 ###

我们注意要设置前缀和后缀

//高亮显示
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces")
                                    .query(q -> q.match(v -> v.field("name").query("风油精")))
                                    .highlight(h -> h.preTags("<span>").postTags("<span>").fields("name", hf -> hf)),
                    Produce.class);
            System.err.println(response.toString());

### 12:分页查询 ###

我们使用match\_all进行全部搜索的时候使用size关键字设置每一页的大小,使用from关键字设置页码  
from的计算公式:(页码-1)\*size

//分页查询
                    final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces")
                                    .query(q->q.matchAll(v->v)).size(2).from(0),
                    Produce.class);
            System.err.println(response.hits().hits());

### 13:排序 ###

使用sort关键字指定需要进行排序的字段设置排序类型即可,我们这里会使用到SortOrder枚举类来进行指定排序方式

desc:降序  
asc:升序

//排序
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces")
                                    .query(q->q.matchAll(v->v))
                                    .sort(builder1 -> builder1.field(f->f.field("price").order(SortOrder.Asc))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 14:指定字段查询 ###

使用\_source关键字在数组中设置需要展示的字段  
值得注意的是在source方法中需要我们写filter去指定是include包含或是exclude去除xx字段

//指定字段查询
            final SearchResponse<Produce> response = client.search(builder ->
                            builder.index("produces")
                                    .query(q->q.matchAll(v->v))
                                    .source(s->s.filter(v->v.includes("price","des"))),
                    Produce.class);
            System.err.println(response.hits().hits());

### 15:聚合查询-求最大值 ###

SearchResponse<Object> search = elasticsearchClient.search(builder ->
                    builder.index("my_test_index")
                            .from(0)
                            .size(1)
                            .aggregations("aa", t ->
                                    t.max(f->
                                            f.field("type"))), Object.class);
            EsResult esResult = EsUtils.searchAnalysis(search);
            Aggregate aa = esResult.getAggregations().get("aa");
            LongTermsAggregate lterms = aa.lterms();
            Buckets<LongTermsBucket> buckets = lterms.buckets();
            List<LongTermsBucket> array = buckets.array();

### 16:桶聚合查询-劣势 group by ###

SearchResponse<JSONObject> search = elasticsearchClient.search(builder ->
                        builder.index(EsIndexConstants.article_info)
                                .query(t->
                                        t.range(f->
                                                f.field("create_time")
                                                        .gte(JsonData.of(startDate))
                                                        .lte(JsonData.of(endDate))))
                                .from(0)
                                .size(1)
                                .aggregations("countValue", t ->
                                        t.terms(f -> f.field("ata_type.keyword")))
                                , JSONObject.class);
    Aggregate countValue = search .getAggregations().get("countValue");
    List<StringTermsBucket> array = countValue.sterms().buckets().array();