elasticsearch与mysql数据同步

忘是亡心i 2023-09-25 18:22 71阅读 0赞

#### 目录 ####

*  数据同步
 *   *  一.思路分析
     *   *  1.同步调用
         *  2.异步通知
         *  3.监听binlog
         *  4.选择
     *  二.实现数据同步
     *   *  1.思路
         *  2.导入demo
         *  3.声明交换机、队列
         *   *  3.1引入依赖
             *  3.2 配置文件
             *  3.3 声明队列交换机名称
             *  3.4 声明队列交换机
         *  4.发送MQ消息
         *   *  4.1 事务配置类
             *  4.2 service 代码
         *  5.接收MQ消息

## 数据同步 ##

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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HAdv7YDC-1679468296276)(assets/image-20210723214758392.png)\]][img-HAdv7YDC-1679468296276_assets_image-20210723214758392.png]

### 一.思路分析 ###

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

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

#### 1.同步调用 ####

方案一：同步调用

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2Kr0UnlI-1679466723120)(assets/image-20210723214931869.png)\]][img-2Kr0UnlI-1679466723120_assets_image-20210723214931869.png]

基本步骤如下：

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

#### 2.异步通知 ####

方案二：异步通知

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-H3Zr9Vqz-1679466723124)(assets/image-20210723215140735.png)\]][img-H3Zr9Vqz-1679466723124_assets_image-20210723215140735.png]

流程如下：

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

#### 3.监听binlog ####

方案三：监听binlog

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KuKuKj7V-1679466723126)(assets/image-20210723215518541.png)\]][img-KuKuKj7V-1679466723126_assets_image-20210723215518541.png]

流程如下：

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

#### 4.选择 ####

方式一：同步调用

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

方式二：异步通知

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

方式三：监听binlog

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

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

#### 1.思路 ####

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

步骤：

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

#### 2.导入demo ####

**资料：**[资料在这里][Link 1]

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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NgNS6gN5-1679466723128)(assets/image-20210723220237930.png)\]][img-NgNS6gN5-1679466723128_assets_image-20210723220237930.png]

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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EHjziqrE-1679466723129)(assets/image-20210723220354464.png)\]][img-EHjziqrE-1679466723129_assets_image-20210723220354464.png]

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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-EesTGlEK-1679466723129)(assets/image-20210723220511090.png)\]][img-EesTGlEK-1679466723129_assets_image-20210723220511090.png]

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

MQ结构如图：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6IzJaY3o-1679466723130)(assets/image-20210723215850307.png)\]][img-6IzJaY3o-1679466723130_assets_image-20210723215850307.png]

##### 3.1引入依赖 #####

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

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

##### 3.2 配置文件 #####

spring:
      rabbitmq:
        host: 192.168.1.100
        username: guest
        password: guest
        virtual-host: /

##### 3.3 声明队列交换机名称 #####

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

package cn.itcast.hotel.constants;
    
        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.4 声明队列交换机 #####

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

#### 4.发送MQ消息 ####

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

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nelp7h1M-1679466723130)(assets/image-20210723221843816.png)\]][img-nelp7h1M-1679466723130_assets_image-20210723221843816.png]

##### 4.1 事务配置类 #####

保证Rabbitmq在提交事务后执行。[事务控制的详情][Link 2]

package cn.itcast.hotel.config;
    
    import com.sun.istack.internal.NotNull;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    import org.springframework.transaction.support.TransactionSynchronizationAdapter;
    import org.springframework.transaction.support.TransactionSynchronizationManager;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    @Component("afterCommitExecutor")
    public class AfterCommitExecutor extends TransactionSynchronizationAdapter implements Executor {
        
        private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();
        private ThreadPoolExecutor threadPool;
    
        private Logger logger = LoggerFactory.getLogger(AfterCommitExecutor.class);
        
        @PostConstruct
        public void init() {
        
            logger.debug("初始化线程池。。。");
            int availableProcessors = Runtime.getRuntime().availableProcessors();
            if (0 >= availableProcessors) {
        
                availableProcessors = 1;
            }
            int maxPoolSize = (availableProcessors > 5) ? availableProcessors * 2 : 5;
            logger.debug("CPU Processors :%s MaxPoolSize:%s", availableProcessors, maxPoolSize);
            threadPool = new ThreadPoolExecutor(
                availableProcessors,
                maxPoolSize,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(maxPoolSize * 2),
                Executors.defaultThreadFactory(),
                new RejectedExecutionHandler() {
        
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
                        logger.debug("Task:%s rejected", r.toString());
                        if (!executor.isShutdown()) {
        
                            executor.getQueue().poll();
                            executor.execute(r);
                        }
                    }
                }
            );
        }
    
        @PreDestroy
        public void destroy() {
        
            logger.debug("销毁线程池。。。");
            if (null != threadPool && !threadPool.isShutdown()) {
        
                threadPool.shutdown();
            }
        }
    
        @Override
        public void execute(@NotNull Runnable runnable) {
        
            if (!TransactionSynchronizationManager.isSynchronizationActive()) {
        
                runnable.run();
                return;
            }
            List<Runnable> threadRunnables = RUNNABLES.get();
            if (threadRunnables == null) {
        
                threadRunnables = new ArrayList<Runnable>();
                RUNNABLES.set(threadRunnables);
                TransactionSynchronizationManager.registerSynchronization(this);
            }
            threadRunnables.add(runnable);
        }
    
        @Override
        public void afterCommit() {
        
            logger.debug("事务提交完成处理 ... ");
            List<Runnable> threadRunnables = RUNNABLES.get();
            for (int i = 0; i < threadRunnables.size(); i++) {
        
                Runnable runnable = threadRunnables.get(i);
                try {
        
                    threadPool.execute(runnable);
                } catch (RuntimeException e) {
        
                    logger.error("", e);
                }
            }
        }
    
        @Override
        public void afterCompletion(int status) {
        
            logger.debug("事务处理完毕 .... ");
            RUNNABLES.remove();
        }
    }

##### 4.2 service 代码 #####

@Override
        @Transactional
        public boolean save(Hotel hotel) {
        
            logger.info("----- into  insert service -----");
            hotel.setId(Long.getLong(UUID.randomUUID().toString()));
            int i = hotelMapper.insert(hotel);
            afterCommitExecutor.execute(new Runnable() {
        
                @Override
                public void run() {
        
                    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId().toString());
                    logger.info("----- rabbitmq send message -----");
                }
            });
    
    
            if (1==i ||"1".equals(1) ){
        
                return true;
    
            }else {
        
                return false;
            }
        }
    
      
        @Override
        @Transactional
        public boolean updateById(Hotel hotel) {
        
            logger.info("----- into service -----");
    
            //修改DB的数据
            int i = hotelMapper.updateById(hotel);
    
            // 使用AfterCommitExecutor
            afterCommitExecutor.execute(new Runnable() {
        
                @Override
                public void run() {
        
                    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId().toString());
                    logger.info("----- rabbitmq send message -----");
                }
            });
    
            logger.info("--------- out service ----------");
    
            if (1==i ||"1".equals(1) ){
        
                return true;
    
            }else {
        
                return false;
            }
        }
    
        @Override
        public void removeById(Long id) {
        
            logger.info("----- into service -----");
    
            //修改DB的数据
            int i = hotelMapper.deleteById(id);
    
            // 使用AfterCommitExecutor
            afterCommitExecutor.execute(new Runnable() {
        
                @Override
                public void run() {
        
                    rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_DELETE_KEY, id);
                    logger.info("----- rabbitmq send message -----");
                }
            });
    
            logger.info("--------- out service ----------");
    
    
        }

#### 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.发送请求
            restHighLevelClient.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.发送请求
            restHighLevelClient.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);
        }
    }

[img-HAdv7YDC-1679468296276_assets_image-20210723214758392.png]: https://img-blog.csdnimg.cn/257c636e348b45c1a60602dc04225d82.png
[img-2Kr0UnlI-1679466723120_assets_image-20210723214931869.png]: https://img-blog.csdnimg.cn/da04ec3377334f83a6d0d0e2dea922ca.png
[img-H3Zr9Vqz-1679466723124_assets_image-20210723215140735.png]: https://img-blog.csdnimg.cn/a7d7a4f69cf64a7cb51cd17851e3a309.png
[img-KuKuKj7V-1679466723126_assets_image-20210723215518541.png]: https://img-blog.csdnimg.cn/e5c6163535584eda80a9de8967b5d6d0.png
[Link 1]: https://pan.baidu.com/s/1iIZ_yAlCZNlFdh3hV8MUYA?pwd=1234
[img-NgNS6gN5-1679466723128_assets_image-20210723220237930.png]: https://img-blog.csdnimg.cn/ce796d85b84b4658a60bf607d0d0be36.png
[img-EHjziqrE-1679466723129_assets_image-20210723220354464.png]: https://img-blog.csdnimg.cn/57cb780e159d47dcbc36f252155a6ebf.png
[img-EesTGlEK-1679466723129_assets_image-20210723220511090.png]: https://img-blog.csdnimg.cn/eae45f33220a4a6283e56b68485c0e65.png
[img-6IzJaY3o-1679466723130_assets_image-20210723215850307.png]: https://img-blog.csdnimg.cn/63f6edb65c544b7e94db0fd520cd8260.png
[img-nelp7h1M-1679466723130_assets_image-20210723221843816.png]: https://img-blog.csdnimg.cn/a1cfbb83fd704333a950748a808f3bdb.png
[Link 2]: https://blog.csdn.net/weixin_53604412/article/details/129729052?spm=1001.2014.3001.5501