文章目录
- 前言
- 一、ES与Mysql数据同步
- 1.1.同步调用
- 1.2.异步通知
- 1.3.监听binlog
- 二、实现数据同步
- 2.1 导入依赖和yaml
- 2.2 声明交换机、队列
- 2.3 发送MQ消息
- 2.4 接收MQ消息
- 2.5 测试
前言
Mysql数据同步
当数据发生增、删、改时，要求对Elasticsearch中数据也要完成相同操作。
一、ES与Mysql数据同步
Elasticsearch中的酒店数据来自于mysql数据库，因此Mysql数据发生改变时，Elasticsearch也必须跟着改变，这个就是Elasticsearch与Mysql之间的数据同步。
常见的数据同步方案有三种：
- 同步调用
- 异步通知
- 监听binlog
方式一：同步调用
- 优点：实现简单，粗暴
- 缺点：业务耦合度高
方式二：异步通知【常用】
- 优点：低耦合，实现难度一般
- 缺点：依赖mq的可靠性
方式三：监听binlog
- 优点：完全解除服务间耦合
- 缺点：开启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中的内容
二、实现数据同步
步骤：
- 单机部署并启动MQ（单机部署在MQ部分有讲）
- 接收者中声明exchange、queue、RoutingKey
- 在hotel-admin发送者中的增、删、改业务中完成消息发送
- 在hotel-demo接收者中完成消息监听，并更新Elasticsearch中数据
- 启动并测试数据同步功能
2.1 导入依赖和yaml
对发送者和消费者都添加依赖和yaml信息
1）引入依赖
<!--amqp--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>2）yaml
spring:rabbitmq:#MQ配置host:192.168.194.131# 主机名port:5672# 端口virtual-host:/# 虚拟主机username:itcast# 用户名password:123321# 密码2.2 声明交换机、队列
MQ结构如图：
1）声明队列交换机名称
在hotel-admin发送者和hotel-demo消费者中的cn.itcast.hotel.constatnts包下新建一个类MqConstants：
packagecn.itcast.hotel.constatnts;publicclassMqConstants{/** * 交换机 */publicfinalstaticStringHOTEL_EXCHANGE="hotel.topic";/** * 监听新增和修改的队列 */publicfinalstaticStringHOTEL_INSERT_QUEUE="hotel.insert.queue";/** * 监听删除的队列 */publicfinalstaticStringHOTEL_DELETE_QUEUE="hotel.delete.queue";/** * 新增或修改的RoutingKey */publicfinalstaticStringHOTEL_INSERT_KEY="hotel.insert";/** * 删除的RoutingKey */publicfinalstaticStringHOTEL_DELETE_KEY="hotel.delete";}2）声明队列交换机
在hotel-demo消费者中，定义配置类，声明队列、交换机：
packagecn.itcast.hotel.config;importcn.itcast.hotel.constants.MqConstants;importorg.springframework.amqp.core.Binding;importorg.springframework.amqp.core.BindingBuilder;importorg.springframework.amqp.core.Queue;importorg.springframework.amqp.core.TopicExchange;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassMqConfig{@BeanpublicTopicExchangetopicExchange(){returnnewTopicExchange(MqConstants.HOTEL_EXCHANGE,true,false);}@BeanpublicQueueinsertQueue(){returnnewQueue(MqConstants.HOTEL_INSERT_QUEUE,true);}@BeanpublicQueuedeleteQueue(){returnnewQueue(MqConstants.HOTEL_DELETE_QUEUE,true);}@BeanpublicBindinginsertQueueBinding(){returnBindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);}@BeanpublicBindingdeleteQueueBinding(){returnBindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);}}2.3 发送MQ消息
在hotel-admin发送者中的增、删、改业务中分别发送MQ消息：
2.4 接收MQ消息
hotel-demo接收到MQ消息要做的事情包括：
- 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库
- 删除消息：根据传递的hotel的id删除索引库中的一条数据
1）写SDL业务
首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、删除业务
voiddeleteById(Longid);voidinsertById(Longid);给hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中实现业务：
@OverridepublicvoiddeleteById(Longid){try{// 1.准备RequestDeleteRequestrequest=newDeleteRequest("hotel",id.toString());// 2.发送请求client.delete(request,RequestOptions.DEFAULT);}catch(IOExceptione){thrownewRuntimeException(e);}}@OverridepublicvoidinsertById(Longid){try{// 0.根据id查询酒店数据Hotelhotel=getById(id);// 转换为文档类型HotelDochotelDoc=newHotelDoc(hotel);// 1.准备Request对象IndexRequestrequest=newIndexRequest("hotel").id(hotel.getId().toString());// 2.准备Json文档request.source(JSON.toJSONString(hotelDoc),XContentType.JSON);// 3.发送请求client.index(request,RequestOptions.DEFAULT);}catch(IOExceptione){thrownewRuntimeException(e);}}2）编写监听器
在hotel-demo中的cn.itcast.hotel.mq包新增一个类：
packagecn.itcast.hotel.mq;importcn.itcast.hotel.constants.MqConstants;importcn.itcast.hotel.service.IHotelService;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;@ComponentpublicclassHotelListener{@AutowiredprivateIHotelServicehotelService;/** * 监听酒店新增或修改的业务 * @param id 酒店id */@RabbitListener(queues=MqConstants.HOTEL_INSERT_QUEUE)publicvoidlistenHotelInsertOrUpdate(Longid){hotelService.insertById(id);}/** * 监听酒店删除的业务 * @param id 酒店id */@RabbitListener(queues=MqConstants.HOTEL_DELETE_QUEUE)publicvoidlistenHotelDelete(Longid){hotelService.deleteById(id);}}2.5 测试
用postman调用增加/删除/修改mysql数据库的接口，然后去页面搜索看看删除的数据还是否能查到，或者修改/增加的数据能不能查出来。
本文的引用仅限自我学习如有侵权，请联系作者删除。
参考知识
ElasticSearch (ES从入门到精通一篇就够了)
ELK介绍
