SpringBoot-ElasticSearch8_胡安民_springboot整合elasticsearch8

未知 02-07 8392

前沿

elasticsearch-rest-high-level-client在 7.15.0 中已弃用。不推荐使用高级 REST 客户端，取而代之的是 Java API 客户端。 spring-boot-starter-data-elasticsearch 也不推荐,虽然基础操作简化了很多,但是一旦使用了es高级特性,那么就如同进入了地狱,同时elasticsearch更新太快了spring-boot-starter-data-elasticsearch的版本根本就赶不上,导致升级会出现很多问题

现在在es官网推荐我们现在使用 Elasticsearch Java API 客户端这个是相当于直接使用elasticsearch自身的接口Api所以不存在升级不兼容的问题

Java API 客户端官网地址

elasticsearch Api 操作大全 rest-api

需要的Maven

官网推荐依赖经过各种的依赖冲突的解决,终于能跑起来了…

<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.4.5</version> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>  <dependency> <groupId>co.elastic.clients</groupId> <artifactId>elasticsearch-java</artifactId> <version>8.1.2</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.12.3</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-annotations</artifactId> <version>2.12.3</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.12.3</version> </dependency> <dependency> <groupId>jakarta.json</groupId> <artifactId>jakarta.json-api</artifactId> <version>2.0.1</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.75</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> </dependencies> 创建客户端 RestClient restClient = RestClient.builder( new HttpHost("localhost", 9200)).build(); ElasticsearchTransport transport = new RestClientTransport( restClient, new JacksonJsonpMapper()); //es 客户端 ElasticsearchClient client = new ElasticsearchClient(transport); 创建索引和映射分片等…

官方文档没有关于怎么设置映射和索引的配置的,自能自己研究,经过千辛万苦看底层源代码(一点注释都没有),了解代码编写的逻辑,经过我不懈的努力终于实现了以下的基础配置,其他的高级配置可以进行参考下配置进行扩展

//构建索引,并且起一个别名 CreateIndexResponse createResponse = client.indices() .create(c -> c .index("my-index") //创索引 .aliases("main-index", a -> a //创别名 .isWriteIndex(true) ) // 映射字段属性 .mappings((m)-> m .properties("name",Property.of(p->p.text(TextProperty.of(p1->p1.analyzer("ik_max_word").searchAnalyzer("ik_smart"))))) .properties("sku" ,Property.of(p->p.text(TextProperty.of(p1->p1.analyzer("ik_max_word").searchAnalyzer("ik_smart"))))) .properties("price",Property.of(p->p.integer(IntegerNumberProperty.of(p1->p1))) ) ) //设置分片 ,numberOfShards分片 ,Replicas副本 .settings((s)->s.numberOfShards("1").numberOfReplicas("2")) ); System.out.println(createResponse.index()); 自己进行二次封装

把官方文档看完后,我只想说一句,真难用, 经过3天的代码封装,基本能解决es的百分之90%的使用场景,代码压缩和简化程度至少优化了50%,同时保留了原有的代码特性, 特殊场景提供了对外客户端自己也可以进行二次开发,特殊的操作需要自己通过es的api接口手动进行修操作,在后端操作es一般就是增删改查数据,以及聚合,高亮等

结构

application.yml es: port: 9200 hostname: localhost alals: true # 是否使用别名的方式访问注解 @Documented @Target({ElementType.FIELD}) @Retention(RetentionPolicy.RUNTIME) @Component public @interface DocId { } @Documented @Target({ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) @Component public @interface EsClass { /** * 索引: 前缀index + 默认类名( 自动转小写) * 默认类名可以修改为我们自定义 * @return */ String index() default ""; /** * 别名: 前缀alias + 默认类名( 自动转小写) *默认类名可以修改为我们自定义 * @return */ String alias() default "" ; /* * 分片 */ int shards()default 1; /** * 分片副本 * @return */ int replicas()default 1; } @Documented @Target({ElementType.FIELD}) @Retention(RetentionPolicy.RUNTIME) @Component public @interface EsField { //默认属性名 String name()default "" ; //数据类型 EsDataType type(); String analyzer()default ""; //分词 String searchAnalyzer()default ""; //搜索分词 } 枚举

不一定都能用,有些可能随着es版本的变化,去掉了,需要啥自行修改

public enum EsDataType { TEXT("text"), KEYWORD("keyword"), FLOAT("float"), LONG("long"), INTEGER("integer"), SHORT("short"), DOUBLE("double"), HALF_FLOAT("half_float"), SCALED_FLOAT("scaled_float"), BYTE("byte"), DATE("date"), BOOLEAN("boolean"), RANGE("rang"), BINARY("binary"), ARRAY("array"), OBJECT("object"), NESTED("nested"), GEO_POINT("geo_point"), GEO_SHAPE("geo_shape"), IP("ip"), COMPLETION("completion"), TOKEN_COUNT("token_count"), ATTACHMENT("attachment"), PERCOLATOR("percolator"); private String type; EsDataType(String type) { this.type = type; } public String getType() { return type; } } 自定义es客户端工具类 package com.es8.utli; import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient; import co.elastic.clients.elasticsearch.ElasticsearchClient; import co.elastic.clients.elasticsearch._types.aggregations.Aggregate; import co.elastic.clients.elasticsearch._types.aggregations.Aggregation; import co.elastic.clients.elasticsearch._types.query_dsl.Query; import co.elastic.clients.elasticsearch.cat.aliases.AliasesRecord; import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord; import co.elastic.clients.elasticsearch.core.*; import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; import co.elastic.clients.elasticsearch.core.search.Hit; import co.elastic.clients.elasticsearch.indices.Alias; import co.elastic.clients.elasticsearch.indices.CreateIndexRequest; import co.elastic.clients.json.jackson.JacksonJsonpMapper; import co.elastic.clients.transport.ElasticsearchTransport; import co.elastic.clients.transport.rest_client.RestClientTransport; import co.elastic.clients.util.ObjectBuilder; import com.alibaba.fastjson.JSONObject; import com.es8.an.DocId; import com.es8.an.EsClass; import com.es8.an.EsField; import com.es8.esenum.EsDataType; import com.reflect.ClassUtls; import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.util.Assert; import java.io.IOException; import java.io.Reader; import java.io.StringReader; import java.lang.reflect.Field; import java.util.*; import java.util.function.Function; import java.util.stream.Collectors; @Component("es8Client") public class Es8Client implements InitializingBean { private static final Logger log = LoggerFactory.getLogger(Es8Client.class); @Value("${es.port}") private int prot; @Value("${es.hostname}") private String hostname; private ElasticsearchClient client; private ElasticsearchAsyncClient asyncClient; private final String indexPrefix = "index-"; private final String aliasPrefix = "alias-"; // 同步客户端 public ElasticsearchClient getClient() { return client; } // 异步客户端 public ElasticsearchAsyncClient getAsyncClient() { return asyncClient; } @Override public void afterPropertiesSet() throws Exception { RestClient restClient = RestClient.builder(new HttpHost(hostname, prot)).build(); ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper()); // es 客户端 this.client = new ElasticsearchClient(transport); this.asyncClient = new ElasticsearchAsyncClient(transport); } /** * 如果索引或者别名已经存在,那么就不会在创建了 * @return 是否成功 * @throws Exception */ public <T> boolean createIndexSettingsMappings(Class<T> tClass) throws Exception { EsClass annotation = tClass.getAnnotation(EsClass.class); String index = getClassIndex(tClass); String alias = getClassAlias(tClass); List<String> indexs = indexs(); List<String> aliases = aliases(); if(indexs.contains(index)||aliases.contains(alias)){ return false; } int shards = annotation.shards(); int replicas = annotation.replicas(); StringBuilder stringBuilder = new StringBuilder("{"); stringBuilder.append( "\"settings\": {\n" + " \"number_of_shards\": " + shards + ",\n" + " \"number_of_replicas\": " + replicas + "\n" + " },"); stringBuilder.append("\"mappings\": {\n" + " \"properties\": "); JSONObject jsonObject = new JSONObject(); for (Field declaredField : tClass.getDeclaredFields()) { declaredField.setAccessible(true); JSONObject jsonObject1 = new JSONObject(); DocId DocId = declaredField.getAnnotation(DocId.class); if (DocId != null) { jsonObject1.put("type", EsDataType.LONG.getType()); jsonObject.put(declaredField.getName(), jsonObject1); continue; } EsField annotation1 = declaredField.getAnnotation(EsField.class); if (annotation1 != null) { String name = annotation1.name(); name = "".equals(name) ? declaredField.getName() : name; EsDataType type = annotation1.type(); String analyzer = annotation1.analyzer(); String searchAnalyzer = annotation1.searchAnalyzer(); jsonObject1.put("type", type.getType()); if (!"".equals(analyzer)) { jsonObject1.put("analyzer", analyzer); } if (!"".equals(searchAnalyzer)) { jsonObject1.put("search_analyzer", searchAnalyzer); } jsonObject.put(name, jsonObject1); } } Assert.isTrue(jsonObject.size() > 0, "请添加es相关注解"); stringBuilder.append(jsonObject); stringBuilder.append("}}"); Reader queryJson = new StringReader(stringBuilder.toString()); String finalIndex = index; String finalAlias = alias; CreateIndexRequest req = CreateIndexRequest.of( b -> b.index(finalIndex) .aliases(finalAlias, Alias.of(a -> a.isWriteIndex(true))) .withJson(queryJson)); return client.indices().create(req).acknowledged(); } // 查询全部索引 public List<String> indexs() { List<IndicesRecord> indices = null; try { indices = client.cat().indices().valueBody(); } catch (IOException e) { e.printStackTrace(); } assert indices != null; return indices.stream().map(IndicesRecord::index).collect(Collectors.toList()); } // 查询全部别名 public List<String> aliases() { List<AliasesRecord> aliasesRecords = null; try { aliasesRecords = client.cat().aliases().valueBody(); } catch (IOException e) { e.printStackTrace(); } assert aliasesRecords != null; return aliasesRecords.stream().map(AliasesRecord::alias).collect(Collectors.toList()); } // 查询全部数据 public <T> List<T> querys(Class<T> tClass) throws IOException { List<T> list = new ArrayList<>(); SearchResponse<T> search = client.search(s -> s.query(q -> q.matchAll(m -> m)), tClass); for (Hit<T> hit : search.hits().hits()) { list.add(hit.source()); Map<String, List<String>> highlight = hit.highlight(); log.info("querys", highlight); } return list; } // 添加数据 /** * @param o 数据源 * @param async 是否异步 true异步 ,false同步 ,如果是异步那么永远返回null * @return */ public <T> String addData(T o, boolean async) { Object id = null; for (Field declaredField : o.getClass().getDeclaredFields()) { declaredField.setAccessible(true); DocId annotation = declaredField.getAnnotation(DocId.class); if (annotation != null) { try { id = declaredField.get(o); } catch (IllegalAccessException e) { e.printStackTrace(); } } } if (id == null) { id = UUID.randomUUID().toString(); } IndexResponse response = null; try { IndexRequest.Builder<T> indexReqBuilder = new IndexRequest.Builder<>(); indexReqBuilder.index(getClassAlias(o.getClass())); indexReqBuilder.id(String.valueOf(id)); indexReqBuilder.document(o); if (async) { asyncClient.index(indexReqBuilder.build()); return null; } response = client.index(indexReqBuilder.build()); } catch (IOException e) { e.printStackTrace(); } assert response != null; return response.id(); } // 批量添加 public <T> void addDatas(List<T> list, boolean async) { BulkRequest.Builder br = new BulkRequest.Builder(); for (T o : list) { Object id = null; for (Field declaredField : o.getClass().getDeclaredFields()) { declaredField.setAccessible(true); DocId annotation = declaredField.getAnnotation(DocId.class); if (annotation != null) { try { id = declaredField.get(o); } catch (IllegalAccessException e) { e.printStackTrace(); } } } if (id == null) { id = UUID.randomUUID().getMostSignificantBits(); } Object finalId = id; br.operations( op -> op.index( idx -> idx.index(getClassAlias(o.getClass())).id(String.valueOf(finalId)).document(o))); } if (async) { asyncClient.bulk(br.build()); return; } BulkResponse result = null; try { result = client.bulk(br.build()); } catch (IOException e) { e.printStackTrace(); } // Log errors, if any assert result != null; if (result.errors()) { log.error("Bulk had errors"); for (BulkResponseItem item : result.items()) { if (item.error() != null) { log.error(item.error().reason()); } } } } public <T> T getDocId(String DocId, Class<T> clazz) { GetResponse<T> response = null; try { response = client.get(g -> g.index(getClassAlias(clazz)).id(DocId), clazz); } catch (IOException e) { e.printStackTrace(); } assert response != null; return response.source(); } public <T> List<T> complexQuery(Query query, Class<T> clazz) throws IOException { List<T> list = new ArrayList<>(); SearchResponse<T> response = client.search(s -> s.index(getClassAlias(clazz)).query(query), clazz); List<Hit<T>> hits = response.hits().hits(); for (Hit<T> hit : hits) { list.add(hit.source()); } return list; } public <T> Aggregate complexQueryAggregations( Query query, Function<Aggregation.Builder, ObjectBuilder<Aggregation>> fn, Class<T> clazz) throws IOException { SearchResponse<T> response = client.search( s -> s.index(getClassAlias(clazz)) .size(0) // 不需要显示数据 ,只想要聚合结果 .query(query) .aggregations("aggregations", fn), clazz); return response.aggregations().get("aggregations"); } //获取类的索引名称(没有指定默认类名首字母小写, 前缀+索引) private <T> String getClassIndex(Class<T> clazz){ EsClass annotation = clazz.getAnnotation(EsClass.class); String index = annotation.index(); index = "".equals(index) ? Objects.requireNonNull(ClassUtls.getClassName(clazz)).toLowerCase() : index.toLowerCase(); return indexPrefix +index; } private <T> String getClassAlias(Class<T> clazz){ EsClass annotation = clazz.getAnnotation(EsClass.class); String alias = annotation.alias(); alias = "".equals(alias) ? Objects.requireNonNull(ClassUtls.getClassName(clazz)).toLowerCase() : alias.toLowerCase(); return aliasPrefix +alias; } } 测试实体类 // 只能使用包装类型 ,不能使用基础类型 ,否则会导致一些高级特性有问题 @EsClass @Data public class UserEsEneity { @DocId private Long id; @Field(type = EsDataType.KEYWORD ) private String name; @Field(type =EsDataType.INTEGER) private Integer age; @Field(type = EsDataType.TEXT,analyzer = "ik_max_word",searchAnalyzer = "ik_smart") private String dec; @Field(type =EsDataType.KEYWORD) private String sku; @Field(type =EsDataType.DOUBLE) private Double price; } 测试Junit类 import co.elastic.clients.elasticsearch._types.aggregations.Aggregate; import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket; import co.elastic.clients.elasticsearch._types.aggregations.StringTermsBucket; import co.elastic.clients.elasticsearch._types.query_dsl.Query; import com.es8.eneity.UserEsEneity; import com.es8.utli.Es8Client; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Map; @RunWith(SpringRunner.class) @SpringBootTest(classes = Es8Application.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) public class Es8ApplictionTest { @Autowired private Es8Client es8Client; @Test public void createIndexSettingsMappings() throws Exception { es8Client.createIndexSettingsMappings(UserEsEneity.class); } @Test public void aliases() throws Exception { System.out.println(es8Client.aliases()); } @Test public void indexs() throws Exception { System.out.println(es8Client.indexs()); } @Test public void selectDocIdExists() throws Exception { if (es8Client.getClient().exists(b -> b.index("main-index").id("1")).value()) { System.out.println("product exists"); } } @Test public void queryAll() throws Exception { List<UserEsEneity> querys = es8Client.queryAll(UserEsEneity.class); System.out.println(querys); } @Test public void addData() throws IOException { UserEsEneity userEsEneity = new UserEsEneity(); userEsEneity.setId(22L); userEsEneity.setName("xxxxa"); userEsEneity.setAge(22); userEsEneity.setDec("xxxxxxxx"); userEsEneity.setPrice(22.1); userEsEneity.setSku("aaa1"); String s = es8Client.addData(userEsEneity,true); System.out.println(s); } @Test public void addDatas() throws IOException { UserEsEneity userEsEneity = new UserEsEneity(); userEsEneity.setId(23L); userEsEneity.setName("hu"); userEsEneity.setAge(22); userEsEneity.setDec("游泳"); userEsEneity.setPrice(22.1); userEsEneity.setSku("aaa1"); UserEsEneity userEsEneity1 = new UserEsEneity(); userEsEneity1.setId(24L); userEsEneity1.setName("an"); userEsEneity1.setAge(22); userEsEneity1.setDec(""); userEsEneity1.setPrice(22.2); userEsEneity1.setSku("vvvvvv"); List<UserEsEneity> list = new ArrayList<>(); list.add(userEsEneity); list.add(userEsEneity1); es8Client.addDatas(list,true); } @Test public void getDocId() throws IOException { UserEsEneity docId = es8Client.getDocId("24", UserEsEneity.class); System.out.println(docId); } @Test public void complexQuery_MatchAll() throws IOException { Query query = Query.of(q -> q.matchAll(m -> m)); List<UserEsEneity> userEsEneities = es8Client.complexQuery(query, UserEsEneity.class); System.out.println(userEsEneities); } @Test public void complexQuery_MatchAll_Alals() throws IOException { Query query = Query.of(q -> q.matchAll(m -> m)); List<UserEsEneity> userEsEneities = es8Client.complexQuery(query, UserEsEneity.class); System.out.println(userEsEneities); } @Test public void complexQuery_MatchQuery() throws IOException { Query query = Query.of(q -> q.match(m -> m.field("name").query("xxxxa"))); List<UserEsEneity> userEsEneities = es8Client.complexQuery(query, UserEsEneity.class); System.out.println(userEsEneities); } @Test public void complexQuery_query_bool_must() throws IOException { Query age = Query.of(q -> q.match(m -> m.field("age").query(22))); Query price = Query.of(q -> q.match(m -> m.field("price").query(22.1))); Query bool = Query.of(q -> q.bool(b -> b.must(age).must(price))); List<UserEsEneity> userEsEneities = es8Client.complexQuery(bool, UserEsEneity.class); System.out.println(userEsEneities); } @Test public void complexQueryHighlight() throws IOException { Query dec = Query.of(q -> q.matchPhrase(m -> m.field("dec").query("匹配"))); List<Map<String, Object>> maps = es8Client.complexQueryHighlight(dec, UserEsEneity.class, "dec"); System.out.println(maps); } @Test public void complexQuery_query_complexQueryAggregations() throws IOException { Query query = Query.of(q -> q.matchAll(m -> m)); Aggregate age = es8Client.complexQueryAggregations(query, a -> a.terms(t -> t.field("age")) , UserEsEneity.class); for (LongTermsBucket longTermsBucket : age.lterms().buckets().array()) { System.out.println("key:"+longTermsBucket.key()+":共多少:"+longTermsBucket.docCount()); } Aggregate name = es8Client.complexQueryAggregations(query, a -> a.terms(t -> t.field("name")) , UserEsEneity.class); for (StringTermsBucket stringTermsBucket : name.sterms().buckets().array()) { System.out.println("key:"+stringTermsBucket.key()+":共多少:"+stringTermsBucket.docCount()); } Aggregate price = es8Client.complexQueryAggregations(query, a -> a.avg(t -> t.field("price")) , UserEsEneity.class); System.out.println(price.avg().value()); } @Test public void delDocId() throws IOException { es8Client.delDocId( "23",UserEsEneity.class); } @Test public void delQuery() throws IOException { Query price = Query.of(q -> q.match(m -> m.field("price").query("0.0"))); es8Client.delQuery( price,UserEsEneity.class); } @Test public void upDocId() throws Exception { UserEsEneity userEsEneity1 = new UserEsEneity(); // userEsEneity1.setId(24L); // userEsEneity1.setName("an"); userEsEneity1.setAge(21); userEsEneity1.setDec("嘻嘻嘻嘻嘻嘻擦擦擦"); userEsEneity1.setPrice(28.2); userEsEneity1.setSku("mmmmm"); es8Client.upDocId("241",userEsEneity1,true); } @Test public void upQuery() throws Exception { UserEsEneity userEsEneity1 = new UserEsEneity(); // userEsEneity1.setId(24L); // userEsEneity1.setName("an"); // userEsEneity1.setAge(21); userEsEneity1.setDec("嘻嘻嘻嘻嘻嘻擦擦擦"); // userEsEneity1.setPrice(28.2); // userEsEneity1.setSku("mmmmm"); Query name = Query.of(q -> q.matchPhrase(m -> m.field("age").query("33"))); es8Client.upQuery(name,userEsEneity1,true); } }

点赞 -收藏-关注-便于以后复习和收到最新内容有其他问题在评论区讨论-或者私信我-收到会在第一时间回复如有侵权，请私信联系我感谢,配合,希望我的努力对你有帮助^_^