Elasticsearch与MySQL数据同步
Elasticsearch数据同步
elasticsearch中的酒店数据来自于mysql数据库,因此mysql数据发生改变时,elasticsearch也必须跟着改变,这个就是elasticsearch与mysql之间的数据同步。
1、数据同步思路分析
方案一:同步调用
基本步骤如下:
- hotel-demo对外提供接口,用来修改elasticsearch中的数据
- 酒店管理服务在完成数据库操作后,直接调用hotel-demo提供的接口
方案二:异步通知
流程如下:
- hotel-admin对mysql数据库数据完成增、删、改后,发送MQ消息
- hotel-demo监听MQ,接收到消息后完成elasticsearch数据修改
方案三:监听binlog
流程如下:
- 给mysql开启binlog功能
- mysql完成增、删、改操作都会记录在binlog中
- hotel-demo基于canal监听binlog变化,实时更新elasticsearch中的内容
小结
方式一:同步调用
- 优点:实现简单,粗暴
- 缺点:业务耦合度高
方式二:异步通知
- 优点:低耦合,实现难度一般
- 缺点:依赖mq的可靠性
方式三:监听binlog
- 优点:完全解除服务间耦合
- 缺点:开启binlog增加数据库负担、实现复杂度高
2、实现Elasticsearch与数据库数据同步
导入项目完成对酒店数据的【增删改查】实现数据同步操作
hotel-admin项目作为酒店管理的微服务。当酒店数据发生增、删、改时,要求对elasticsearch中数据也要完成相同操作。
步骤:
-
创建hotel-admin项目,启动并测试酒店数据的CRUD
-
声明exchange、queue、RoutingKey
-
在hotel-admin中的增、删、改业务中完成消息发送
-
在hotel-demo中完成消息监听,并更新elasticsearch中数据
-
启动并测试数据同步功能
2.1、hotel_demo
声明交换机、队列
在hotel-admin、hotel-demo中引入rabbitmq的依赖:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId></dependency>
配置springboot核心配置文件yml
spring: rabbitmq: host: 192.168.26.131 port: 5672 username: itcast password: 123321 virtual-host: /
在hotel-demo
项目中定义一个类记录交换机和队列的名称
package cn.itcast.hotel.constants;/** * 项目名称:hotel-demo * 描述:定义全局的交换机和队列以及RoutingKey * * @author zhong * @date 2022-06-05 14:00 */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";}
在hotel-demo中,定义配置类,声明队列、交换机之间的绑定关系:
package cn.itcast.hotel.config;import cn.itcast.hotel.constants.MqConstants;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.Queue;import org.springframework.amqp.core.TopicExchange;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;/** * 项目名称:hotel-demo * 描述:创建交换机和队列,绑定关系 * * @author zhong * @date 2022-06-05 14:06 */@Configurationpublic class MqConfig { /** * 1、定义持久化交换机 * @return */ @Bean public TopicExchange topicExchange() { return new TopicExchange(MqConstants.HOTEL_EXCHANGE,true,false); } /** * 2、创建新增的队列,持久化 */ @Bean public Queue insertQueue(){ return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true); } /** * 3、创建新增的队列,持久化 */ @Bean public Queue deleteQueue(){ return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true); } /** * 4、定义交换机和队列的绑定关系以及通信的值 */ @Bean public Binding insertQueueBinding(){ return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY); } /** * 5、定义删除的交换机和队列关系以及通讯的值 * @return */ @Bean public Binding deleteQueueBinding(){ return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY); }}
2.2、hotel_admin
声明交换机、队列发送更新信息
- 引入依赖
- 添加yml配置文件,连接队列
- 复制
MqConstants
类到constants
包中 - 因为交换机和队列已经在消费者
hotel_demo
中创建了,接下来就在提供者中发送消息
在控制器中注入发送消息的RabbitTemplate
和对应的消息发送内容
因为队列是占内存的,所以我们直接传递一个id过去就可以了
/** * 注入发送消息类 */@Autowiredprivate RabbitTemplate rabbitTemplate;// 添加、修改数据中加入rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());// 删除类中加入rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);
2.3、在hotel_demo
消费者中监听队列消息
当我们监听到队列的信息后就需要对文档库等做一个操作,所以这里注入了
IHotelService
业务层的接口,下面就要创建这些接口的实现类
package cn.itcast.hotel.mq;import cn.itcast.hotel.constants.MqConstants;import cn.itcast.hotel.service.IHotelService;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;/** * 项目名称:hotel-demo * 描述:监听发送的队列消息 * * @author zhong * @date 2022-06-05 14:29 */@Componentpublic class HotelListener { @Autowired IHotelService hotelService; /** * 监听酒店添加或修改的队列 * @param id */ @RabbitListener(priority = MqConstants.HOTEL_INSERT_QUEUE) public void listenerHotelInsertOrUpdate(Long id){ // 调用修改 hotelService.insertById(id); } /** * 监听酒店删除队列消息 * @param id */ @RabbitListener(priority = MqConstants.HOTEL_DELETE_QUEUE) public void listenerHotelInsertOrDelete(Long id){ // 调用删除 hotelService.deleteById(id); }}
2.4、创建监听调用的实现类
但提供者有消息发送过来的时候就会执行下面的两段代码
/** * 删除数据监听 * @param id */@Overridepublic void deleteById(Long id) { try { // 1、获取request DeleteRequest request = new DeleteRequest("hotel").id(id.toString()); // 2、发送请求 client.delete(request,RequestOptions.DEFAULT); } catch (IOException e) { throw new RuntimeException(e); }}/** * 修改数据(添加)监听 * @param id */@Overridepublic void insertById(Long id) { try { // 0、根据id查询数据库数据【实现了MP接口可以直接调用】 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); }}
2.6、测试
-
重启服务消费者
hotel_demo
查看管理页面队列和交换机绑定关系
交换机绑定
-
重启服务提供者
hotel_admin
-
修改酒店消息然后返回刷新页面的消息,查看酒店信息是否修改了