Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)
Posted LL.LEBRON
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)相关的知识,希望对你有一定的参考价值。
文章目录
视频指路👉B站黑马微服务超级推荐!
MQ实现elasticsearch与mysql的数据同步
elasticsearch中的酒店数据来自于mysql数据库,因此mysql数据发生改变时,elasticsearch也必须跟着改变,这个就是elasticsearch与mysql之间的数据同步。
1.思路分析
常见的数据同步方案有三种:
- 同步调用
- 异步通知
- 监听binlog
1.1 同步调用
方案一:同步调用
基本步骤如下:
hotel-demo
对外提供接口,用来修改elasticsearch
中的数据- 酒店管理服务在完成数据库操作后,直接调用
hotel-demo
提供的接口
1.2 异步通知
方案二:异步通知
流程如下:
hotel-admin
对mysql
数据库数据完成增、删、改后,发送MQ
消息hotel-demo
监听MQ
,接收到消息后完成elasticsearch
数据修改
1.3 监听binlog
方案三:监听binlog
流程如下:
- 给
mysql
开启binlog
功能 mysql
完成增、删、改操作都会记录在binlog
中hotel-demo
基于canal
监听binlog
变化,实时更新elasticsearch
中的内容
1.4 选择
方式一:同步调用
- 优点:实现简单,粗暴
- 缺点:业务耦合度高
方式二:异步通知
- 优点:低耦合,实现难度一般
- 缺点:依赖mq的可靠性
方式三:监听binlog
- 优点:完全解除服务间耦合
- 缺点:开启binlog增加数据库负担、实现复杂度高
2.MQ实现数据同步
2.1 思路
利用课前资料提供的hotel-admin
项目作为酒店管理的微服务。当酒店数据发生增、删、改时,要求对elasticsearch
中数据也要完成相同操作。
步骤:
-
导入课前资料提供的
hotel-admin
项目,启动并测试酒店数据的CRUD -
声明
exchange
、queue
、RoutingKey
-
在
hotel-admin
中的增、删、改业务中完成消息发送 -
在
hotel-demo
中完成消息监听,并更新elasticsearch
中数据 -
启动并测试数据同步功能
2.2 声明交换机、队列
MQ结构如图:
(1)引入依赖
在hotel-admin
、hotel-demo
中引入rabbitmq
的依赖:
<!--amqp-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
(2)声明队列交换机名称
在hotel-admin
和hotel-demo
中的cn.itcast.hotel.constatnts
包下新建一个类MqConstants
:
package cn.itcast.hotel.constatnts;
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";
(3)声明队列交换机
在hotel-demo
中,定义配置类MqConfig
,声明队列、交换机、绑定队列和交换机:
@Configuration
public class MqConfig
/**
* 定义交换机
* @return
*/
@Bean
public TopicExchange topicExchange()
return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
/**
* 定义新增或修改的队列
* @return
*/
@Bean
public Queue insertQueue()
return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
/**
* 定义删除的队列
* @return
*/
@Bean
public Queue deleteQueue()
return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
/**
* 绑定修改队列和交换机
* @return
*/
@Bean
public Binding insertQueueBinding()
return BindingBuilder.bind(insertQueue())
.to(topicExchange())
.with(MqConstants.HOTEL_INSERT_KEY);
/**
* 绑定删除队列和交换机
* @return
*/
@Bean
public Binding deleteQueueBinding()
return BindingBuilder.bind(deleteQueue())
.to(topicExchange())
.with(MqConstants.HOTEL_DELETE_KEY);
2.4 发送MQ消息
当在hotel-admin
中发生增、删、改业务中分别发送MQ
消息:
要传入酒店的id给hotel-demo
用于修改
@PostMapping
public void saveHotel(@RequestBody Hotel hotel)
// 新增酒店
hotelService.save(hotel);
// 发送MQ消息
rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.INSERT_KEY, hotel.getId());
@PutMapping()
public void updateById(@RequestBody Hotel hotel)
if (hotel.getId() == null)
throw new InvalidParameterException("id不能为空");
hotelService.updateById(hotel);
// 发送MQ消息
rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.INSERT_KEY, hotel.getId());
@DeleteMapping("/id")
public void deleteById(@PathVariable("id") Long id)
hotelService.removeById(id);
// 发送MQ消息
rabbitTemplate.convertAndSend(HotelMqConstants.EXCHANGE_NAME, HotelMqConstants.DELETE_KEY, id);
2.5 接受MQ消息
hotel-demo
接收到MQ
消息要做的事情包括:
- 新增消息:根据传递的hotel的id查询hotel信息,然后新增一条数据到索引库
- 删除消息:根据传递的hotel的id删除索引库中的一条数据
(1)首先在hotel-demo
的cn.itcast.hotel.service
包下的IHotelService
中新增新增、删除业务
void deleteById(Long id);
void insertById(Long id);
(2)给hotel-demo
中的cn.itcast.hotel.service.impl
包下的HotelService
中实现业务:
@Override
public void deleteById(Long id)
try
// 1.准备Request
DeleteRequest request = new DeleteRequest("hotel", id.toString());
// 2.发送请求
client.delete(request, RequestOptions.DEFAULT);
catch (IOException e)
throw new RuntimeException(e);
@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(JSON.toJSONString(hotelDoc), XContentType.JSON);
// 3.发送请求
client.index(request, RequestOptions.DEFAULT);
catch (IOException e)
throw new RuntimeException(e);
(3)编写监听器
在hotel-demo
中的cn.itcast.hotel.mq
包新增一个类:
@Component
public class Hotelistener
@Autowired
private IHotelService iHotelServicel;
/**
* 监听酒店新增或者修改的业务
* @param id 酒店id
*/
@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
public void listenHotelInsertOrupdate(Long id)
iHotelServicel.saveById(id);
/**
* 监听酒店删除的业务
* @param id 酒店id
*/
@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
public void listenHotelDelete(Long id)
iHotelServicel.deleteById(id);
最后喜欢的小伙伴,记得三连哦!😏🍭😘
以上是关于Elasticsearch学习笔记-p8(实现elasticsearch与mysql的数据同步)的主要内容,如果未能解决你的问题,请参考以下文章
阿里工作7年,肝到P8就剩这份学习笔记了,已助10个朋友拿到offer
阿里工作8年,肝到P8就剩这份学习笔记了,已助朋友拿到10个Offer
阿里工作6年,肝到P8就剩这份学习笔记了,已助朋友拿到7个Offer