Elasticsearch基本操作

1.1 索引库操作

引入依赖 :

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

因为SpringBoot默认的ES版本是7.17.10，所以我们需要覆盖默认的ES版本：

<properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><elasticsearch.version>7.12.1</elasticsearch.version>
</properties>

初始化RestClient :

@Bean
public RestHighLevelClient client(){return  new RestHighLevelClient(RestClient.builder(HttpHost.create("http://192.168.164.128:9200")));
}

在启动类加上如上的代码初始化client.
结合数据表结构创建索引库结构 :

PUT /items
{"mappings": {"properties": {"id": {"type": "keyword"},"name":{"type": "text","analyzer": "ik_max_word","copy_to": "all"},"price":{"type": "integer"},"stock":{"type": "integer"},"image":{"type": "keyword","index": false},"category":{"type": "keyword","copy_to": "all"},"brand":{"type": "keyword","copy_to": "all"},"sold":{"type": "integer"},"commentCount":{"type": "integer"},"isAD":{"type": "boolean"},"updateTime":{"type": "date"},"all":{"type": "text","analyzer": "ik_max_word"}}	}
}

1.1.1 创建索引库 :

@Test
void testCreateIndex() throws IOException {// 1.创建Request对象CreateIndexRequest request = new CreateIndexRequest("items");// 2.准备请求参数request.source(MAPPING_TEMPLATE, XContentType.JSON);// 3.发送请求client.indices().create(request, RequestOptions.DEFAULT);
}// 这个可以放到constants里面 
static final String MAPPING_TEMPLATE = "{\n" +"  \"mappings\": {\n" +"    \"properties\": {\n" +"      \"id\": {\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"name\":{\n" +"        \"type\": \"text\",\n" +"        \"analyzer\": \"ik_max_word\"\n" +"      },\n" +"      \"price\":{\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"stock\":{\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"image\":{\n" +"        \"type\": \"keyword\",\n" +"        \"index\": false\n" +"      },\n" +"      \"category\":{\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"brand\":{\n" +"        \"type\": \"keyword\"\n" +"      },\n" +"      \"sold\":{\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"commentCount\":{\n" +"        \"type\": \"integer\"\n" +"      },\n" +"      \"isAD\":{\n" +"        \"type\": \"boolean\"\n" +"      },\n" +"      \"updateTime\":{\n" +"        \"type\": \"date\"\n" +"      }\n" +"    }\n" +"  }\n" +"}";

1.1.2 删除索引库 :

@Test
void testDeleteIndex() throws IOException {// 1.创建Request对象DeleteIndexRequest request = new DeleteIndexRequest("items");// 2.发送请求client.indices().delete(request, RequestOptions.DEFAULT);
}

1.1.3 判断索引库是否存在

@Test
void testExistsIndex() throws IOException {// 1.创建Request对象GetIndexRequest request = new GetIndexRequest("items");// 2.发送请求boolean exists = client.indices().exists(request, RequestOptions.DEFAULT);// 3.输出System.err.println(exists ? "索引库已经存在！" : "索引库不存在！");
}

1.2 文档操作

1.2.1 新增文档

索引库结构与数据库结构还存在一些差异，因此我们要定义一个索引库结构对应的实体

package com.hmall.item.domain.dto;import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;import java.time.LocalDateTime;@Data
@ApiModel(description = "索引库实体")
public class ItemDTO{@ApiModelProperty("商品id")private String id;@ApiModelProperty("商品名称")private String name;@ApiModelProperty("价格（分）")private Integer price;@ApiModelProperty("库存数量")private Integer stock;@ApiModelProperty("商品图片")private String image;@ApiModelProperty("类目名称")private String category;@ApiModelProperty("品牌名称")private String brand;@ApiModelProperty("销量")private Integer sold;@ApiModelProperty("评论数")private Integer commentCount;@ApiModelProperty("是否是推广广告，true/false")private Boolean isAD;@ApiModelProperty("更新时间")private LocalDateTime updateTime;
}

操作代码 :

@Test
void testAddDocument() throws IOException {// 1.根据id查询商品数据Item item = itemService.getById(100002644680L);// 2.转换为文档类型ItemDTO itemDTO = BeanUtil.copyProperties(item, ItemDTO.class);// 3.将ItemDTO转jsonString doc = JSONUtil.toJsonStr(itemDTO);// 1.准备Request对象IndexRequest request = new IndexRequest("items").id(itemDTO.getId());// 2.准备Json文档request.source(doc, XContentType.JSON);// 3.发送请求client.index(request, RequestOptions.DEFAULT);
}

1.2.2 查询文档

@Test
void testGetDocumentById() throws IOException {// 1.准备Request对象GetRequest request = new GetRequest("items").id("100002644680");// 2.发送请求GetResponse response = client.get(request, RequestOptions.DEFAULT);// 3.获取响应结果中的sourceString json = response.getSourceAsString();ItemDTO itemDTO = JSONUtil.toBean(json, ItemDTO.class);System.out.println("itemDTO = " + itemDTO);
}

1.2.3 删除文档

@Test
void testDeleteDocument() throws IOException {// 1.准备Request，两个参数，第一个是索引库名，第二个是文档idDeleteRequest request = new DeleteRequest("item", "100002644680");// 2.发送请求client.delete(request, RequestOptions.DEFAULT);
}

1.2.4 修改文档

@Test
void testUpdateDocument() throws IOException {// 1.准备RequestUpdateRequest request = new UpdateRequest("items", "100002644680");// 2.准备请求参数request.doc("price", 58800,"commentCount", 1);// 3.发送请求client.update(request, RequestOptions.DEFAULT);
}

1.2.5 批量导入文档

@Test
void testBulkRequest() throws IOException {// 批量查询酒店数据List<Hotel> hotels = hotelService.list();// 1.创建RequestBulkRequest request = new BulkRequest();// 2.准备参数，添加多个新增的Requestfor (Hotel hotel : hotels) {// 2.1.转换为文档类型HotelDocHotelDoc hotelDoc = new HotelDoc(hotel);// 2.2.创建新增文档的Request对象request.add(new IndexRequest("hotel").id(hotelDoc.getId().toString()).source(JSONUtil.toJsonStr(hotelDoc), XContentType.JSON));}// 3.发送请求client.bulk(request, RequestOptions.DEFAULT);
}

1.3 RestClient查询

1.3.1 普通查询

@Test
void testMatch() throws IOException {// 1. 创建Request对象SearchRequest request = new SearchRequest("hotel");// 2. 组织请求参数request.source().query(QueryBuilders.matchQuery("all", "如家"));// 3. 发送请求SearchResponse response = client.search(request, RequestOptions.DEFAULT);handleResponse(response);
}

1.3.2 复合条件查询

@Test
void testBool() throws IOException {// 1.准备RequestSearchRequest request = new SearchRequest("hotel");// 2.准备DSL// 2.1.准备BooleanQueryBoolQueryBuilder boolQuery = QueryBuilders.boolQuery();// 2.2.添加termboolQuery.must(QueryBuilders.termQuery("city", "杭州"));// 2.3.添加rangeboolQuery.filter(QueryBuilders.rangeQuery("price").lte(250));request.source().query(boolQuery);// 3.发送请求SearchResponse response = client.search(request, RequestOptions.DEFAULT);// 4.解析响应handleResponse(response);
}

1.3.3 分页排序查询

@Test
void testPageAndSort() throws IOException {int pageNo = 1, pageSize = 5;// 1.准备RequestSearchRequest request = new SearchRequest("hotel");// 2.1.搜索条件参数//request.source().query(QueryBuilders.matchAllQuery());request.source().query(QueryBuilders.matchQuery("all", "如家"));// 2.2.排序参数request.source().sort("price", SortOrder.ASC);// 2.3.分页参数request.source().from((pageNo - 1) * pageSize).size(pageSize);// 3.发送请求SearchResponse response = client.search(request, RequestOptions.DEFAULT);// 4.解析响应handleResponse(response);
}

1.3.4 高亮分页查询

    @Test	void testHighLight() throws IOException {int pageNo = 1, pageSize = 3;// 1.准备RequestSearchRequest request = new SearchRequest("hotel");// 2.1.搜索条件参数request.source().query(QueryBuilders.matchQuery("all", "如家"));// 2.2 高亮条件
//		request.source().highlighter(
//				new HighlightBuilder()
//						.field("name")
//						.preTags("<em>")
//						.postTags("</em>")
//		);request.source().highlighter(new HighlightBuilder().field("name").field("brand").requireFieldMatch(false));// 2.3.排序参数request.source().sort("price", SortOrder.ASC);// 2.4.分页参数request.source().from((pageNo - 1) * pageSize).size(pageSize);// 3.发送请求SearchResponse response = client.search(request, RequestOptions.DEFAULT);// 4.解析响应handleResponse(response);}

1.3.5 分页过滤复合查询

/*** 搜索* @param params 请求参数* @return 分页结果*/
@Override
public PageResult search(RequestParams params) {try {// 1. 准备RequestSearchRequest request = new SearchRequest("hotel");// 2.1 queryboolBasicQuery(params, request);// 2.2 分页int pageNo = params.getPage();int pageSize = params.getSize();request.source().from((pageNo - 1) * pageSize).size(pageSize);// 3. 发送请求SearchResponse response = client.search(request, RequestOptions.DEFAULT);return handleResponse(response);} catch (IOException e) {throw new RuntimeException(e);}
}
/*** 构建基本的bool查询* @param params 请求参数* @param request 请求对象*/
private void boolBasicQuery(RequestParams params, SearchRequest request) {// 1.构建BooleanQueryBoolQueryBuilder boolQuery = QueryBuilders.boolQuery();// 关键字搜索String key = params.getKey();if (StrUtil.isEmpty(key)) {boolQuery.must(QueryBuilders.matchAllQuery());} else {boolQuery.must(QueryBuilders.matchQuery("all", key));}// 城市条件String city = params.getCity();if(StrUtil.isNotEmpty(city)){boolQuery.filter(QueryBuilders.termQuery("city", city));}// 品牌条件String brand = params.getBrand();if(StrUtil.isNotEmpty(brand)){boolQuery.filter(QueryBuilders.termQuery("brand", brand));}// 星级条件String starName = params.getStarName();if(StrUtil.isNotEmpty(starName)){boolQuery.filter(QueryBuilders.termQuery("starName", starName));}// 价格条件Integer minPrice = params.getMinPrice();Integer maxPrice = params.getMaxPrice();if(minPrice != null && maxPrice != null){boolQuery.filter(QueryBuilders.rangeQuery("price").gte(minPrice).lte(maxPrice));}request.source().query(boolQuery);// // 2.算分控制// FunctionScoreQueryBuilder functionScoreQuery =// 		QueryBuilders.functionScoreQuery(// 				// 原始查询，相关性算分的查询// 				boolQuery,// 				// function score的数组// 				new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{// 						// 其中的一个function score 元素// 						new FunctionScoreQueryBuilder.FilterFunctionBuilder(// 								// 过滤条件// 								QueryBuilders.termQuery("isAD", true),// 								// 算分函数// 								ScoreFunctionBuilders.weightFactorFunction(10)// 						)// 				});// request.source().query(functionScoreQuery);
}

1.3.6 处理响应结果

private void handleResponse(SearchResponse response) {SearchHits searchHits = response.getHits();// 1. 获取总条数long total = searchHits.getTotalHits().value;log.info("总条数：{}", total);// 2. 遍历结果数组SearchHit[] hits = searchHits.getHits();for(SearchHit hit : hits) {// 3. 获取JSON字符串String json = hit.getSourceAsString();// 4. 转换为Java对象HotelDoc hotelDoc = JSONUtil.toBean(json, HotelDoc.class);// 5. 获取高亮结果Map<String, HighlightField> highlightFields = hit.getHighlightFields();if(CollUtil.isNotEmpty(highlightFields)){// 5.1 有高亮结果 获取name的高亮结果HighlightField field1 = highlightFields.get("name");HighlightField field2 = highlightFields.get("brand");if(field1 != null && field2 != null){String name = field1.getFragments()[0].string();String brand = field2.getFragments()[0].string();hotelDoc.setName(name);hotelDoc.setBrand(brand);}}log.info("HotelDoc：{}", hotelDoc);}
}

1.4 Mysql和ES数据同步

这里我使用的是rbmq做异步通知es更新数据

1.4.1 引入依赖和配置yml

<!--amqp-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

  rabbitmq:host: 192.168.164.128port: 5672username: itheimapassword: 123321virtual-host: /

1.4.2 定义交换机队列名称( 常量 )

/*** @author Ccoo* 2024/2/12*/
public class MqConstants {/*** 交换机*/public final static String HOTEL_EXCHANGE = "hotel.topic";/*** 监听新增和修改的队列*/public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";/*** 监听删除的队列*/public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";/*** 新增或修改的RoutingKey*/public final static String HOTEL_INSERT_KEY = "hotel.insert";/*** 删除的RoutingKey*/public final static String HOTEL_DELETE_KEY = "hotel.delete";
}

1.4.3 声明和绑定交换机与队列( 使用注解不需要声明 )

/*** @author Ccoo* 2024/2/12*/
@Configuration
public class MqConfig {@Beanpublic TopicExchange topicExchange(){return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);}@Beanpublic Queue insertQueue(){return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);}@Beanpublic Queue deleteQueue(){return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);}@Beanpublic Binding insertQueueBinding(){return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY)}@Beanpublic Binding deleteQueueBinding(){return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY)}
}/*** 监听酒店新增或修改的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)public void listenHotelInsertOrUpdate(Long id){hotelService.insertById(id);}/*** 监听酒店删除的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)public void listenHotelDelete(Long id){hotelService.deleteById(id);}

/*** 监听酒店新增或修改的业务* @param id 酒店id*/
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = MqConstants.HOTEL_INSERT_QUEUE, durable = "true"),exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.TOPIC),key = MqConstants.HOTEL_INSERT_KEY
))
public void listenHotelInsertOrUpdate(Long id){hotelService.insertById(id);
}/*** 监听酒店删除的业务* @param id 酒店id*/
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = MqConstants.HOTEL_DELETE_QUEUE, durable = "true"),exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.TOPIC),key = MqConstants.HOTEL_DELETE_KEY
))
public void listenHotelDelete(Long id){hotelService.deleteById(id);
}

1.4.4 编写业务逻辑

/*** 删除数据同步到ES* @param id*/
@Override
public void deleteById(Long id) {try {// 1.准备RequestDeleteRequest request = new DeleteRequest("hotel", id.toString());// 2.发送请求client.delete(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}
}
/*** 新增或修改数据同步到ES* @param id*/
@Override
public void insertById(Long id) {try {// 0.根据id查询酒店数据Hotel hotel = getById(id);// 转换为文档类型HotelDoc hotelDoc = new HotelDoc(hotel);// 1.准备Request对象IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());// 2.准备Json文档request.source(JSONUtil.toJsonStr(hotelDoc), XContentType.JSON);// 3.发送请求client.index(request, RequestOptions.DEFAULT);} catch (IOException e) {throw new RuntimeException(e);}
}