Elasticsearch与MySQL数据同步

Posted 2022-06-08 小钟要学习！！！

Elasticsearch数据同步

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

1、数据同步思路分析

方案一：同步调用

基本步骤如下：

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

2、实现Elasticsearch与数据库数据同步

导入项目完成对酒店数据的【增删改查】实现数据同步操作

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

步骤：

• 创建hotel-admin项目，启动并测试酒店数据的CRUD

• 声明exchange、queue、RoutingKey

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

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

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

2.1、hotel_demo声明交换机、队列

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

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

配置springboot核心配置文件yml

spring:
  rabbitmq:
    host: 192.168.26.131
    port: 5672
    username: itcast
    password: 123321
    virtual-host: /

在hotel-demo项目中定义一个类记录交换机和队列的名称

package cn.itcast.hotel.constants;

/**
 * 项目名称：hotel-demo
 * 描述：定义全局的交换机和队列以及RoutingKey
 *
 * @author zhong
 * @date 2022-06-05 14:00
 */
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中，定义配置类，声明队列、交换机之间的绑定关系：

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;

/**
 * 项目名称：hotel-demo
 * 描述：创建交换机和队列，绑定关系
 *
 * @author zhong
 * @date 2022-06-05 14:06
 */
@Configuration
public class MqConfig 
    /**
     * 1、定义持久化交换机
     * @return
     */
    @Bean
    public TopicExchange topicExchange() 
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE,true,false);
    

    /**
     * 2、创建新增的队列,持久化
     */
    @Bean
    public Queue insertQueue()
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    

    /**
     * 3、创建新增的队列,持久化
     */
    @Bean
    public Queue deleteQueue()
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    

    /**
     * 4、定义交换机和队列的绑定关系以及通信的值
     */
    @Bean
    public Binding insertQueueBinding()
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    

    /**
     * 5、定义删除的交换机和队列关系以及通讯的值
     * @return
     */
    @Bean
    public Binding deleteQueueBinding()
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    

2.2、hotel_admin声明交换机、队列发送更新信息

1. 引入依赖
2. 添加yml配置文件，连接队列
3. 复制MqConstants类到constants包中
4. 因为交换机和队列已经在消费者hotel_demo中创建了，接下来就在提供者中发送消息

在控制器中注入发送消息的RabbitTemplate和对应的消息发送内容

因为队列是占内存的，所以我们直接传递一个id过去就可以了

/**
 * 注入发送消息类
 */
@Autowired
private RabbitTemplate rabbitTemplate;

// 添加、修改数据中加入
rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId());

// 删除类中加入
rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_DELETE_KEY,id);

2.3、在hotel_demo消费者中监听队列消息

当我们监听到队列的信息后就需要对文档库等做一个操作，所以这里注入了IHotelService业务层的接口，下面就要创建这些接口的实现类

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;

/**
 * 项目名称：hotel-demo
 * 描述：监听发送的队列消息
 *
 * @author zhong
 * @date 2022-06-05 14:29
 */
@Component
public class HotelListener 

    @Autowired
    IHotelService hotelService;

    /**
     * 监听酒店添加或修改的队列
     * @param id
     */
    @RabbitListener(priority = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenerHotelInsertOrUpdate(Long id)
        // 调用修改
        hotelService.insertById(id);
    

    /**
     * 监听酒店删除队列消息
     * @param id
     */
    @RabbitListener(priority = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenerHotelInsertOrDelete(Long id)
        // 调用删除
        hotelService.deleteById(id);
    

2.4、创建监听调用的实现类

但提供者有消息发送过来的时候就会执行下面的两段代码

/**
 * 删除数据监听
 * @param id
 */
@Override
public void deleteById(Long id) 
    try 
        // 1、获取request
        DeleteRequest request = new DeleteRequest("hotel").id(id.toString());
        // 2、发送请求
        client.delete(request,RequestOptions.DEFAULT);
     catch (IOException e) 
        throw new RuntimeException(e);
    


/**
 * 修改数据(添加)监听
 * @param id
 */
@Override
public void insertById(Long id) 
    try 
        // 0、根据id查询数据库数据【实现了MP接口可以直接调用】
        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);
    

2.6、测试

1. 重启服务消费者hotel_demo

   查看管理页面队列和交换机绑定关系

   

   交换机绑定

   

2. 重启服务提供者hotel_admin

3. 修改酒店消息然后返回刷新页面的消息，查看酒店信息是否修改了