数据同步

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

思路分析

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

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

同步调用

基本步骤如下：

• 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 增加数据库负担、实现复杂度高

实现数据同步，基于 MQ 的实现思路

利用提供的 hotel-admin 项目作为酒店管理的微服务。当酒店数据发生增、删、改时，要求对 ElasticSearch 中数据也要完成相同操作。

步骤：

• 导入 hotel-admin 项目，启动并测试酒店数据的 CRUD
• 声明 exchange、queue、RoutingKey
• 在 hotel-admin 中的增、删、改业务中完成消息发送
• 在 hotel-demo 中完成消息监听，并更新 ElasticSearch 中数据
• 启动并测试数据同步功能

导入 demo

代码链接：GitHub

运行后，访问 http://localhost:8099

声明交换机、队列

MQ 结构如图：

引入依赖并修改配置文件

在 hotel-admin、hotel-demo 中引入 rabbitmq 的依赖：

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

spring:
  rabbitmq:
    host: rabbitmq
    port: 5672
    username: halo
    password: halo
    virtual-host: /

声明交换机、队列

在 hotel-admin 和 hotel-demo 中的 cn.itcast.hotel.constatnts 包下新建一个类 MqConstants：

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 中，定义配置类，声明队列、交换机：

@Configuration
public class MqConfig {
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    }

    @Bean
    public Queue insertQueue(){
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    }

    @Bean
    public Queue deleteQueue(){
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    }

    @Bean
    public Binding insertQueueBinding(){
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding(){
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    }
}

发送 MQ 消息

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

@Autowired
private RabbitTemplate rabbitTemplate;

@PostMapping
public void saveHotel(@RequestBody Hotel hotel){
    hotelService.save(hotel);
    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
}

@PutMapping()
public void updateById(@RequestBody Hotel hotel){
    if (hotel.getId() == null) {
        throw new InvalidParameterException("id不能为空");
    }
    hotelService.updateById(hotel);
    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());
}

@DeleteMapping("/{id}")
public void deleteById(@PathVariable("id") Long id) {
    hotelService.removeById(id);
    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);
}

接收 MQ 消息

hotel-demo 接收到 MQ 消息要做的事情包括：

• 新增消息：根据传递的 hotel 的 id 查询 hotel 信息，然后新增一条数据到索引库
• 删除消息：根据传递的 hotel 的 id 删除索引库中的一条数据

首先在 hotel-demo 的 cn.itcast.hotel.service 包下的 IHotelService 中新增新增、删除业务

void deleteById(Long id);

void insertById(Long id);

给 hotel-demo 中的 cn.itcast.hotel.service.impl 包下的 HotelService 中实现业务：

@Override
public void deleteById(Long id) {
    try {
        // 1.准备Request
        DeleteRequest 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);
    }
}

编写监听器，在 hotel-demo 中的 cn.itcast.hotel.mq 包新增一个类：

@Component
public class HotelListener {

    @Autowired
    private 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);
    }
}