一.项目搭建
1.sql表
用户数据库:
tb_user:用户表,其中包含用户的详细信息
tb_address:用户地址表
商品数据库
tb_item:商品表
订单数据库
tb_order:用户订单表
tb_order_detail:订单详情表,主要是订单中包含的商品信息
tb_order_logistics:订单物流表,订单的收货人信息
2.模块搭建
feign-api:是通用的api模块,一些feign的客户端、实体类、工具类都可以放这里
item-service:商品微服务,负责商品的CRUD
order-service:订单微服务,负责订单的CRUD
search-service:搜索服务,负责es搜索和数据同步
user-service:用户微服务,负责用户的CRUD、用户地址的CRUD
项目的pom略...
3.前端
页面分为后台仓库端,和商城首页,代码略,前端部署在了nginx上,nginx的部署配置截图如下
进入cmd窗口启动ngixn,启动成功后,效果如图:
4.启动注册中心
这儿示例的单机模式,没用到集群
5.配网关
链接里是配置网关的模板:http://t.csdn.cn/Un29Z
步骤简单,如果你要借鉴的话,套自己公司模板就行。
二.写项目
由于mysql和es的特性,查询方面,由mysql同步给es,至于删除修改新增都是基于mysql。
因此业务接口包括:
分页查询商品(基于mysql)
根据id查询商品
新增商品
修改商品
根据id删除商品(直接删除,不做逻辑删除)
上架、下架商品
商品表,实体类略,提供给其他微服务的实体类略
把其他都略了,本篇主要讲海量查询的业务实现。
搜索业务
搜索相关业务包括:
设计索引库数据结构
完成数据导入
实现基本搜索功能
实现过滤项聚合功能
数据同步
1.es索引库创建
应该把mysql表中的哪些字段放到ES中?
搜索时需要的字段 ,商品的标题、分类、品牌、价格
搜索后需要展示的字段: 图片,销量数、评价数
隐藏的字段:id、商品的创建时间
基本字段包括:
用于关键字全文检索的字段,比如All,里面包含name、brand、category信息
分类.品牌.价格.销量.idname.评价.数量.图片
PUT /item
{
"mappings": {
"properties": {
"id":{
"type": "long"
},
"name":{
"type": "text",
"analyzer": "ik_max_word",
"copy_to": "all"
},
"price":{
"type": "long"
},
"image":{
"type": "keyword"
},
"category":{
"type": "keyword",
"copy_to": "all"
},
"brand":{
"type": "keyword",
"copy_to": "all"
},
"sold":{
"type": "integer"
},
"commentCount":{
"type": "integer"
},
"isAD":{
"type": "boolean"
},
"all":{
"type": "text",
"analyzer": "ik_max_word",
"search_analyzer": "ik_smart"
}
}
}
}
设计文档对应的Java类:ItemDoc
@Data
@NoArgsConstructor
public class ItemDoc {
private Long id;
private String name;
private Long price;
private String image;
private String category;
private String brand;
private Integer sold;
private Integer commentCount;
private Boolean isAD;
public ItemDoc(Item item) {
// 属性拷贝
BeanUtils.copyProperties(item, this);
}
}
2.海量数据导入es
要把数据库数据导入到elasticsearch中,包括下面几步:
将商品微服务中的分页查询商品接口定义为一个FeignClient,放到feign-api模块中
搜索服务编写一个业务,实现下面功能:
调用item-service提供的FeignClient,分页查询商品 PageDTO<Item>
将查询到的商品封装为一个ItemDoc对象,放入ItemDoc集合
将ItemDoc集合批量导入elasticsearch中
注意:数据库数据比较多,一下导入内存会炸裂哈哈哈,所以分页导入,一下三千条自设定。
1.//fegin里的分页查询接口
@FeignClient("itemservice")
public interface ItemClient {
// 查询商品数据
@GetMapping("/item/list")
public PageDTO<Item> list(
@RequestParam(value = "page", defaultValue = "1") Integer pageNum,
@RequestParam(value = "size", defaultValue = "5") Integer pageSize);
}
2.
@Data
@NoArgsConstructor
@AllArgsConstructor
public class PageDTO<T> {
/**
* 总条数
*/
private Long total;
/**
* 当前页数据
*/
private List<T> list;
}
3.
@Data
public class Item {
private Long id;//商品id
private String name;//商品名称
private Long price;//价格(分)
private Integer stock;//库存数量
private String image;//商品图片
private String category;//分类名称
private String brand;//品牌名称
private String spec;//规格
private Integer sold;//销量
private Integer commentCount;//评论数
private Integer status;//商品状态 1-正常,2-下架
private Boolean isAD;// 是否是广告
}
@SpringBootTest
public class ESTest {
@Autowired
private ItemClient itemClient;
@Autowired
private RestHighLevelClient client;
@Test
public void test() throws Exception {
Integer pageNum = 1;
// 循环
while (true) {
// 查询3000条
PageDTO<Item> page = itemClient.list(pageNum, 3000);
List<Item> list = page.getList();
// 跳出循环
if (list.size() < 1) {
break;
}
// 将数据同步到索引库
// 1.创建request
BulkRequest request = new BulkRequest();
// 2.准备DSL
for (Item item : list) {
// 2-1 转为es实体
ItemDoc itemDoc = new ItemDoc(item);
// 2-2 转为json
String json = JSON.toJSONString(itemDoc);
// 2-3 新增文档
request.add(new IndexRequest("item")
.id(itemDoc.getId().toString())
.source(json, XContentType.JSON));
}
// 3.发送请求
client.bulk(request, RequestOptions.DEFAULT);
// 页码++
pageNum++;
}
}
}
3.实现基本搜索功能
@RestController
@RequestMapping("/search")
public class SearchController {
@Autowired
private SearchService searchService;
// 分页条件查询
@PostMapping("/list")
public PageDTO<ItemDoc> search(@RequestBody RequestParams params) throws IOException {
return searchService.search(params);
}
@Service
public class SearchService {
@Autowired
private RestHighLevelClient client;
// 分页条件查询
public PageDTO<ItemDoc> search(RequestParams params) throws IOException {
// 1.创建request
SearchRequest request = new SearchRequest("item");
// 2.准备DSL
// 2-1 query
buildBasicQuery(params, request);
// 2-2 分页
Integer page = params.getPage();
Integer size = params.getSize();
request.source().from((page-1)*size).size(size);
// 2-3 高亮
request.source().highlighter(new HighlightBuilder().field("name").requireFieldMatch(false));
// 3.发送请求
SearchResponse response = client.search(request, RequestOptions.DEFAULT);
// 4.解析响应
return handleResponse(response);
}
// 抽取查询条件代码
private void buildBasicQuery(RequestParams params, SearchRequest request) {
String key = params.getKey();
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
if (StrUtil.isEmpty(key)) {
boolQuery.must(QueryBuilders.matchAllQuery());
}else{
boolQuery.must(QueryBuilders.matchQuery("all", key));
}
// 品牌
if (StrUtil.isNotEmpty(params.getBrand())) {
boolQuery.filter(QueryBuilders.termQuery("brand", params.getBrand()));
}
// 分类
if (StrUtil.isNotEmpty(params.getCategory())) {
boolQuery.filter(QueryBuilders.termQuery("category", params.getCategory()));
}
// 价格
if (params.getMinPrice() != null && params.getMaxPrice() != null) {
boolQuery.filter(QueryBuilders.rangeQuery("price").gte(params.getMinPrice()*100).lte(params.getMaxPrice()*100));
}
request.source().query(boolQuery);
}
private PageDTO<ItemDoc> handleResponse(SearchResponse response) {
// 4.解析响应
SearchHits searchHits = response.getHits();
// 4.1 获取总记录数
long total = searchHits.getTotalHits().value;
// 4.2 获取文档数组
List<ItemDoc> items = new ArrayList<>();
SearchHit[] hits = searchHits.getHits();
for (SearchHit hit : hits) {
// 获取文档source
String json = hit.getSourceAsString();
ItemDoc hotelDoc = JSON.parseObject(json, ItemDoc.class);
// 处理高亮
Map<String, HighlightField> highlightFields = hit.getHighlightFields();
if (CollectionUtil.isNotEmpty(highlightFields)) {
HighlightField highlightField = highlightFields.get("name");
if (highlightField!=null) {
String name = highlightField.getFragments()[0].string();
hotelDoc.setName(name);
}
}
items.add(hotelDoc);
}
return new PageDTO<>(total,items);
}
4.过滤聚合功能
@RestController
@RequestMapping("/search")
public class SearchController {
@Autowired
private SearchService searchService;
// 分页条件查询
@PostMapping("/list")
public PageDTO<ItemDoc> search(@RequestBody RequestParams params) throws IOException {
return searchService.search(params);
}
// 聚合查询
@PostMapping("/filters")
public Map<String, List<String>> filters(@RequestBody RequestParams params) throws IOException{
return searchService.filters(params);
}
}
// 聚合查询
public Map<String, List<String>> filters(RequestParams params) throws IOException {
// 1.创建request
SearchRequest request = new SearchRequest("item");
// 2.准备DSL参数
// 2-1 query
buildBasicQuery(params, request);
// 2-2 设置size
request.source().size(0);
// 2-3 聚合
request.source().aggregation(AggregationBuilders.terms("brandAgg").field("brand").size(20));
request.source().aggregation(AggregationBuilders.terms("categoryAgg").field("category").size(20));
// 3.发出请求
SearchResponse response = client.search(request, RequestOptions.DEFAULT);
// 4.解析响应
Map<String, List<String>> result = new HashMap<>();
Aggregations aggregations = response.getAggregations();
// 4-1 根据品牌名称
List<String> brandAgg = getAggByName(aggregations, "brandAgg");
result.put("brand", brandAgg);
// 4-2 根据分类名称
List<String> cityAgg = getAggByName(aggregations, "categoryAgg");
result.put("category", cityAgg);
return result;
}
private List<String> getAggByName(Aggregations aggregations, String aggName) {
// 根据名称获取聚合结果
Terms terms = aggregations.get(aggName);
// 获取bucktes
List<? extends Terms.Bucket> buckets = terms.getBuckets();
// 遍历
List<String> list = new ArrayList<>();
for (Terms.Bucket bucket : buckets) {
// 获取聚合名称(key)
String key = bucket.getKeyAsString();
// 添加集合
list.add(key);
}
// 返回
return list;
}
数据同步
基于RabbitMQ实现数据库、elasticsearch的数据同步,要求如下:
商品上架时:search-service新增商品到elasticsearch
商品下架时:search-service删除elasticsearch中的商品
1)启动MQ docker 容器命令启动MQ的镜像
2)item-service和search-service设置连接MQ坐标和配置文件
坐标配置复制模板见此链接:http://t.csdn.cn/YQAwY
3)创建交换机和队列实现绑定
消费者或者生产者都可以创建交换机。记得交给spring管理就行。
@Configuration
public class MQConfig {
// 创建交换机
@Bean
public DirectExchange itemDirect() {
return new DirectExchange("item.direct");
}
// 上架队列
@Bean
public Queue upQueue() {
return new Queue("up.queue");
}
// 下架队列
@Bean
public Queue downQueue() {
return new Queue("down.queue");
}
// 上架队列绑定交换机
@Bean
public Binding bindingUpQueue(DirectExchange itemDirect,Queue upQueue){
return BindingBuilder.bind(upQueue).to(itemDirect).with("item.up");
}
// 下架队列绑定交换机
@Bean
public Binding bindingDownQueue(DirectExchange itemDirect,Queue downQueue){
return BindingBuilder.bind(downQueue).to(itemDirect).with("item.down");
}
}
4)item-service发送消息
@Autowired
private RabbitTemplate rabbitTemplate;
// 上下架
@PutMapping("/status/{id}/{status}")
public void updateStataus(
@PathVariable("id") Long id,
@PathVariable("status") Integer status) {
// update tb_item set status = ? where id = ?;
// 在mp通用service中支持链式编程 (修改局部字段)
itemService.update().set("status", status).eq("id", id).update();
// 发送MQ消息
if (status==1) {
// 上架
rabbitTemplate.convertAndSend("item.direct","item.up",id);
}else{
// 下架
rabbitTemplate.convertAndSend("item.direct","item.down",id);
}
}
5)search-service监听消息
@Slf4j
@Component
public class ItemListener {
@Autowired
private ItemClient itemClient;
@Autowired
private RestHighLevelClient client;
// 监听上架
@RabbitListener(queues = "up.queue")
public void listenUpQueue(Long id) {
try {
log.info("上架:{}", id);
// 查询数据库
Item item = itemClient.findById(id);
// 转为es实体
ItemDoc itemDoc = new ItemDoc(item);
// 转为json
String json = JSON.toJSONString(itemDoc);
// 保存到索引库
// 1.创建request
IndexRequest request = new IndexRequest("item").id(id.toString());
// 2.准备DSL
request.source(json, XContentType.JSON);
// 3.发送请求
client.index(request, RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
// 监听下架
@RabbitListener(queues = "down.queue")
public void listenDownQueue(Long id) {
log.info("下架:{}", id);
try {
// 1.创建request
DeleteRequest request = new DeleteRequest("item", id.toString());
// 2.发送请求
client.delete(request,RequestOptions.DEFAULT);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
6)编写fegin中间件里ItemClient实现根据id查询商品对象
@FeignClient("itemservice")
public interface ItemClient {
// 查询商品数据
@GetMapping("/item/list")
public PageDTO<Item> list(
@RequestParam(value = "page", defaultValue = "1") Integer pageNum,
@RequestParam(value = "size", defaultValue = "5") Integer pageSize);
// 根据id查询
@GetMapping("/item/{id}")
public Item findById(@PathVariable("id") Long id);
}
完成索引库代码同步文章来源:https://www.toymoban.com/news/detail-810016.html
ok了,写完了文章来源地址https://www.toymoban.com/news/detail-810016.html
到了这里,关于商城项目-es的海量查询/聚合/数据同步的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!