ES 和 MySQL 数据同步

携手创作，共同成长！这是我参与「掘金日新计划 · 8 月更文挑战」的第5天，点击查看活动详情

如果有需要可以参考一下这里的数据同步方法https://blog.51cto.com/u_15300825/3097338

es 中的数据来自于 mysql 数据库，因此 mysql 数据发生改变的时候， es 也必须跟着改变，这个就是 es 和 mysql 之间的数据同步。

**注意：**后续多个数据库之间进行共同开发，数据同步也是非常重要的一个点。

1. 思路分析

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

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

接下来我们有一个具体的场景，就是我们数据库端关于酒店的信息将进行增删改的操作，使用不同的方法将 mysql 和 es 的数据同步。

1.1 同步调用

• 基本步骤

  当用户进行修改操作，首先调用服务a（酒店信息管理服务）对 mysql 数据库进行修改，然后服务a远程调用服务b（酒店搜素服务），服务b对 es 数据库进行修改。

  

  等待时间 = 服务a响应时间 + 服务b响应时间

• 优缺点：

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

1.2 异步通知

• 基本步骤：

  当用户进行修改操作，服务a修改 mysql 数据库之后将修改信息发送到消息队列中，有对应的服务b将监听消息并修改 es 数据库。

  

  等待时间 = 服务a响应时间

• 优缺点：

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

1.3 监听 binlog

• 基本步骤：

  当用户进行修改操作，服务a修改 mysql 数据库即返回结果给用户。mysql数据库自动鉴定监听 mysql binlog 实现数据变化后的实时通知服务b，服务b修改 es 数据库。

  

  等待时间 = 服务a响应时间

• 优缺点：

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

低耦合，实现复杂度高，增加 mysql 负担。追求极致的耦合和性能可以选择这个。

2.实现数据同步

相关代码在这里可以查看https://gitee.com/hgd20010523/code-management/tree/master/SpringCloud_ES

2.1.思路

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

步骤：

• 导入课前资料提供的hotel-admin项目，启动并测试酒店数据的CRUD

• 声明exchange、queue、RoutingKey

• 在hotel-admin中的增、删、改业务中完成消息发送

• 在hotel-demo中完成消息监听，并更新elasticsearch中数据

• 启动并测试数据同步功能

2.2.导入demo

导入课前资料提供的hotel-admin项目：

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

其中包含了酒店的CRUD功能：

2.3.声明交换机、队列

MQ结构如图：

1）引入依赖

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

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

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

2.4.发送MQ消息

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

2.5.接收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.准备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);
    }
}

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 {

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