SpringCloud：ElasticSearch之数据同步

╰+攻爆jí腚メ 2024-03-22 10:39 46阅读 0赞

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

![在这里插入图片描述][4e073213bff44201987e53078b109896.png_pic_center]

### 1.思路分析 ###

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

*  同步调用
 *  异步通知
 *  监听`binlog`

#### 1.1.同步调用 ####

方案一：同步调用

![在这里插入图片描述][7897417ab9bd4147897442d43c74a04f.png_pic_center]

基本步骤如下：

*  `hotel-demo`对外提供接口，用来修改`elasticsearch`中的数据
 *  酒店管理服务在完成数据库操作后，直接调用`hotel-demo`提供的接口，

#### 1.2.异步通知 ####

方案二：异步通知

![在这里插入图片描述][3b6272c6230b4c8481785161d993dc78.png_pic_center]

流程如下：

*  `hotel-admin`对`mysql`数据库数据完成增、删、改后，发送`MQ`消息
 *  `hotel-demo`监听`MQ`，接收到消息后完成`elasticsearch`数据修改

#### 1.3.监听`binlog` ####

方案三：监听`binlog`

![在这里插入图片描述][fe8cdfd52ac3404f8fe0986ebf91c637.png_pic_center]

流程如下：

*  给`mysql`开启`binlog`功能
 *  `mysql`完成增、删、改操作都会记录在`binlog`中
 *  `hotel-demo`基于`canal`监听`binlog`变化，实时更新`elasticsearch`中的内容

#### 1.4.选择 ####

方式一：同步调用

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

方式二：异步通知

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

方式三：监听`binlog`

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

### 2.实现数据同步 ###

#### 2.1.思路 ####

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

步骤：

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

#### 2.2.创建`hotel-admin` ####

创建`hotel-admin`项目：

![在这里插入图片描述][cb3cf74ba862455eadf62883e62bf911.png_pic_center]

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

![在这里插入图片描述][53fe330e1530499e9ef134428a6f480f.png_pic_center]

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

![在这里插入图片描述][f90247e843c84ba89f8d9cf09b7832c3.png_pic_center]

#### 2.3.声明交换机、队列 ####

`MQ`结构如图：

![在这里插入图片描述][cb5d5ff9e6c4461cbbaa1c63f3c2d3fa.png_pic_center]

##### 1）引入依赖 #####

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

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

##### 2）声明队列交换机名称 #####

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

package com.dcxuexi.hotel.constants;
    
    /***
     * @Title MqConstants
     * @Description TOTD
     * @Auter DongChuang
     * @Date 2023/4/23 22:23
     * @Version 1.0.0
     */
    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 com.dcxuexi.hotel.config;
    
    import com.dcxuexi.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;
    
    /***
     * @Title MqConfig
     * @Description TOTD
     * @Auter DongChuang
     * @Date 2023/4/23 22:25
     * @Version 1.0.0
     */
    @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`消息：

![在这里插入图片描述][20690e44b3a040f0bf8e95f13d43e1a1.png_pic_center]

#### 2.5.接收`MQ`消息 ####

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

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

1）首先在`hotel-demo`的`com.dcxuexi.hotel.service`包下的`IHotelService`中新增新增、删除业务

void deleteById(Long id);
    
    void insertById(Long id);

2）给`hotel-demo`中的`com.dcxuexi.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`中的`com.dcxuexi.hotel.mq`包新增一个类：

package com.dcxuexi.hotel.mq;
    
    import com.dcxuexi.hotel.constants.MqConstants;
    import com.dcxuexi.hotel.service.IHotelService;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    /***
     * @Title HotelListener
     * @Description TOTD
     * @Auter DongChuang
     * @Date 2023/4/28 21:28
     * @Version 1.0.0
     */
    @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);
        }
    
    }

[4e073213bff44201987e53078b109896.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/e519dba66cde42d4ac150960043a58e3.png
[7897417ab9bd4147897442d43c74a04f.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3d0c5fe9156b4ca08954b48596044786.png
[3b6272c6230b4c8481785161d993dc78.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3334066c124d40f7bac06fbad301ff8d.png
[fe8cdfd52ac3404f8fe0986ebf91c637.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/ccec148a4160469eb8b12a39d11e350e.png
[cb3cf74ba862455eadf62883e62bf911.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/8948f1d471c94a1e80d43ae1132b80de.png
[53fe330e1530499e9ef134428a6f480f.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/88f75cb2cfa34f93a444931365fb8e70.png
[f90247e843c84ba89f8d9cf09b7832c3.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/2f93d454bd9b46d0bdf5ec4768f3143c.png
[cb5d5ff9e6c4461cbbaa1c63f3c2d3fa.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3654ee0adb7b43f58e3bc29a108f1b02.png
[20690e44b3a040f0bf8e95f13d43e1a1.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/716324ca11ff464fa8fc4d16773d2456.png