ES与MySQL数据同步实现方式

1.什么是数据同步:

1.Elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，Elasticsearch也必须跟着改变，这个就是Elasticsearch与mysql之间的数据同步

2.数据同步实现方式：

   常见的数据同步方案有三种：
  

• 同步调用
  
• 异步通知
  
• 监听binlog
  

  2.1.同步调用：

a.办理架构图：

• 1.方案一：同步调用：
  
  
  
  
• 2.基本步调如下：
  
  
  
  • hotel-demo对外提供接口，用来修改elasticsearch中的数据
    
  • 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口，
    
  
  

b.优缺点分析：

• 优点：实现简单，粗暴
  
• 缺点：业务耦合度高
  

---

2.2.异步通知

a.实现的架构图：

• 1.方案二：异步通知
  
  
  
  
• 2.流程如下：
  
  
  
  • hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息
    
  • hotel-demo监听MQ，吸收到消息后完成Elasticsearch数据修改
    
  
  

b.优缺点分析：

• 优点：低耦合，实现难度一样寻常
  
• 缺点：依靠mq的可靠性
  

---

2.3.监听binlog：

a.实现的架构图：

• 1.方案三：监听binlog
  
  
  
  
• 2.流程如下：
  
  
  
  • 给mysql开启binlog功能
    
  • mysql完成增、删、改操作都会记录在binlog中
    
  • hotel-demo基于canal监听binlog变革，实时更新elasticsearch中的内容
    
  
  

b.优缺点分析：

• 优点：完全解除服务间耦合
  
• 缺点：开启binlog增加数据库负担、实现复杂度高
  

---

3.MQ实现MySQL与Es的数据同步：

3.1.案例说明：

• 1.导入hotel-admin项目作为酒店管理的微服务。然后当酒店数据发生增、删、改时，要求对Elasticsearch中数据也要完成雷同操作
  

---

3.2.实现步调：

a.酒店管理项目的运行：

• 1.导入hotel-admin项目并运行，启动并测试酒店数据的CRUD，运行后，访问 http://localhost:8099
  
  
  
  
• 2.其中包罗了酒店的CRUD功能：
  
  
  
  

b.项目引入ES依靠：

• 1.在hotel-admin、hotel-demo中引入rabbitmq的依靠：
  1. <!--amqp-->
  2. <dependency>
  3.     <groupId>org.springframework.boot</groupId>
  4.     <artifactId>spring-boot-starter-amqp</artifactId>
  5. </dependency>

  复制代码

c.MQ架构规划：

• 1.利用RabbitMQ，然后MQ结构如下图,声明exchange、queue、RoutingKey
  
  
  
  
• 2.声明队列互换机名称：在hotel-admin和hotel-demo中的cn.itcast.hotel.constatnts包下新建一个类MqConstants：
  1. package cn.itcast.hotel.constatnts;
  3.     public class MqConstants {
  4.     /**
  5.      * 交换机
  6.      */
  7.     public final static String HOTEL_EXCHANGE = "hotel.topic";
  8.     /**
  9.      * 监听新增和修改的队列
  10.      */
  11.     public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
  12.     /**
  13.      * 监听删除的队列
  14.      */
  15.     public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
  16.     /**
  17.      * 新增或修改的RoutingKey
  18.      */
  19.     public final static String HOTEL_INSERT_KEY = "hotel.insert";
  20.     /**
  21.      * 删除的RoutingKey
  22.      */
  23.     public final static String HOTEL_DELETE_KEY = "hotel.delete";
  24. }

  复制代码
• 3.声明队列互换机在hotel-demo中，界说配置类，声明队列、互换机：
  1. package cn.itcast.hotel.config;
  3. import cn.itcast.hotel.constants.MqConstants;
  4. import org.springframework.amqp.core.Binding;
  5. import org.springframework.amqp.core.BindingBuilder;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.amqp.core.TopicExchange;
  8. import org.springframework.context.annotation.Bean;
  9. import org.springframework.context.annotation.Configuration;
  11. @Configuration
  12. public class MqConfig {
  13.     @Bean
  14.     public TopicExchange topicExchange(){
  15.         return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
  16.     }
  18.     @Bean
  19.     public Queue insertQueue(){
  20.         return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
  21.     }
  23.     @Bean
  24.     public Queue deleteQueue(){
  25.         return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
  26.     }
  28.     @Bean
  29.     public Binding insertQueueBinding(){
  30.         return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
  31.     }
  33.     @Bean
  34.     public Binding deleteQueueBinding(){
  35.         return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
  36.     }
  37. }

  复制代码

d.生产者发送消息：

• 1.发送MQ消息：在hotel-admin中的增、删、改业务中分别发送MQ消息：
  
  
  
  

e.消耗端吸收消息：

   

• 1.吸收MQ消息：hotel-demo吸收到MQ消息要做的事情包括：
  
  
  
  • 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库
    
  • 删除消息：根据传递的hotel的id删除索引库中的一条数据
    
  
  

  

• 1.首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、删除业务
  1. void deleteById(Long id);
  3. void insertById(Long id);

  复制代码
  
  
  • 2.给hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中实现业务：
    
  1. @Override
  2. public void deleteById(Long id) {
  3.     try {
  4.         // 1.准备Request
  5.         DeleteRequest request = new DeleteRequest("hotel", id.toString());
  6.         // 2.发送请求
  7.         client.delete(request, RequestOptions.DEFAULT);
  8.     } catch (IOException e) {
  9.         throw new RuntimeException(e);
  10.     }
  11. }
  13. @Override
  14. public void insertById(Long id) {
  15.     try {
  16.         // 0.根据id查询酒店数据
  17.         Hotel hotel = getById(id);
  18.         // 转换为文档类型
  19.         HotelDoc hotelDoc = new HotelDoc(hotel);
  21.         // 1.准备Request对象
  22.         IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
  23.         // 2.准备Json文档
  24.         request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
  25.         // 3.发送请求
  26.         client.index(request, RequestOptions.DEFAULT);
  27.     } catch (IOException e) {
  28.         throw new RuntimeException(e);
  29.     }
  30. }

  复制代码
  
  
• 3.编写监听器:在hotel-demo中的cn.itcast.hotel.mq包新增一个类：
  1. package cn.itcast.hotel.mq;
  3. import cn.itcast.hotel.constants.MqConstants;
  4. import cn.itcast.hotel.service.IHotelService;
  5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.stereotype.Component;
  9. @Component
  10. public class HotelListener {
  12.     @Autowired
  13.     private IHotelService hotelService;
  15.     /**
  16.      * 监听酒店新增或修改的业务
  17.      * @param id 酒店id
  18.      */
  19.     @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
  20.     public void listenHotelInsertOrUpdate(Long id){
  21.         hotelService.insertById(id);
  22.     }
  24.     /**
  25.      * 监听酒店删除的业务
  26.      * @param id 酒店id
  27.      */
  28.     @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
  29.     public void listenHotelDelete(Long id){
  30.         hotelService.deleteById(id);
  31.     }
  32. }

  复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。