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的依赖：

  <!--amqp-->
  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>
  

c.MQ架构规划：

• 1.使用RabbitMQ，然后MQ结构如下图,声明exchange、queue、RoutingKey
  
• 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中，定义配置类，声明队列、交换机：

  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;@Configuration
  public class MqConfig {@Beanpublic TopicExchange topicExchange(){return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);}@Beanpublic Queue insertQueue(){return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);}@Beanpublic Queue deleteQueue(){return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);}@Beanpublic Binding insertQueueBinding(){return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);}@Beanpublic Binding deleteQueueBinding(){return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);}
  }
  

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中新增新增、删除业务

  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.准备RequestDeleteRequest 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包新增一个类：

  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;@Component
  public class HotelListener {@Autowiredprivate IHotelService hotelService;/*** 监听酒店新增或修改的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)public void listenHotelInsertOrUpdate(Long id){hotelService.insertById(id);}/*** 监听酒店删除的业务* @param id 酒店id*/@RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)public void listenHotelDelete(Long id){hotelService.deleteById(id);}
  }