13. 搜索引擎-ES-Mysql数据同步-程序员充电站

文章目录

前言
一、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);}}