分布式 SpringCloudAlibaba、Feign与RabbitMQ实现MySQL到ES数据同步

偏执的太偏执、 2024-05-01 04:55 6阅读 0赞

#### 文章目录 ####

*  ⛄引言
 *  一、思路分析
 *   *  ⛅实现方式
     *  ⚡框架选择
 *  二、实现数据同步
 *   *  ⌚需求分析
     *  ⏰搭建环境
     *  ⚡核心源码
 *  三、测试
 *  四、源码获取
 *  ⛵小结

## ⛄引言 ##

**本文参考黑马 分布式Elastic search**

Elasticsearch是一款非常强大的开源搜索引擎，具备非常多强大功能，可以帮助我们从海量数据中快速找到需要的内容

## 一、思路分析 ##

### ⛅实现方式 ###

**同步调用**

方案一：同步调用

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

基本步骤如下：

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

**异步通知**

方案二：异步通知

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

流程如下：

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

**监听binlong**

方案三：监听binlog

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

流程如下：

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

### ⚡框架选择 ###

方式一：同步调用

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

方式二：异步通知

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

方式三：监听binlog

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

**本次实现方式我们选择 以RabbitMQ 异步方式 搭载 SpringCloud Alibaba + Feign 实现。**

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

### ⌚需求分析 ###

**需求**

实现酒店管理增删改查业务，已提供页面。 完成其**数据发生增删改查操作时 同步 ElasticSearch**

**分析**

**我们采用分布式技术的方式来实现**

框架采用 SpringCloud Alibaba、Nacos 、OpenFeign 远程调用、RabbitMQ 作为消息承载体承载数据、 Elastic Search 搜索引擎

### ⏰搭建环境 ###

**以下为模块概览**

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

**主要分为两大模块**

*  完成酒店模块增删改查业务，引入MQ依赖，完成其向MQ的发送消息 **此模块作为生产者**
 *  完成ES-MQ模块，引入MQ、ES依赖，完成`接受MQ的消息以及完成对ES的更新` **此模块作为消费者**

**注意：Nacos需要自行下载，本项目依赖于Nacos注册中心， 运行起来后不影响后面的服务注册进nacos**

**本次所用到的 RabbitMQ、 ElasticSearch 均部署在 云服务器**

MQ结构如图：

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

### ⚡核心源码 ###

**hotel-service 业务模块**

导入hotel-service 核心代码，已完成基础的增删改查工作。 **具体源码公众号搜索 `程序员Bug终结者` 回复 es 获取**

**ES模块引入依赖**

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
    </dependency>

**声明队列交换机名称**

public class MqConstants {
        
    
        /**
         * 交换机
         */
        public static final String HOTEL_EXCHANGE = "hotel.topic";
    
        /**
         * 新增或修改的routing_key
         */
        public static final String HOTEL_INSERT_KEY = "hotel.insert";
    
        /**
         * 删除的 routing_key
         */
        public static final String HOTEL_DELETE_KEY = "hotel.delete";
    }

**hotel-service模块 发送消息**

@RestController
    @RequestMapping("hotel")
    public class HotelController {
        
    
        @Autowired
        private HotelService hotelService;
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        @GetMapping("/{id}")
        public Hotel queryById(@PathVariable("id") Long id){
        
            return hotelService.getById(id);
        }
    
        @GetMapping("/list")
        public PageResult hotelList(
                @RequestParam(value = "page", defaultValue = "1") Integer page,
                @RequestParam(value = "size", defaultValue = "1") Integer size
        ){
        
            Page<Hotel> result = hotelService.page(new Page<>(page, size));
    
            return new PageResult(result.getTotal(), result.getRecords());
        }
    
        @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);
        }
    }

**ES模块接受消息**

@Component
    @Slf4j
    public class MqConsumerListener {
        
    
    
        @Resource
        private HotelService hotelService;
    
        /**
         * 监听酒店新增或修改的业务
         * @param id
         */
        @RabbitListener(bindings = @QueueBinding(value = @Queue(value = MqConstants.HOTEL_INSERT_KEY, durable = "true"),
                exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT), key = MqConstants.HOTEL_INSERT_KEY))
        public void listenHotelInsertOrUpdate(String id) throws IOException {
        
            hotelService.insertById(id);
        }
    
        /**
         * 监听酒店删除的业务
         * @param id
         */
        @RabbitListener(bindings = @QueueBinding(value = @Queue(value = MqConstants.HOTEL_DELETE_KEY, durable = "true"),
                exchange = @Exchange(name = MqConstants.HOTEL_EXCHANGE, type = ExchangeTypes.DIRECT), key = MqConstants.HOTEL_DELETE_KEY))
        public void listenHotelDelete(String id) throws IOException {
        
            hotelService.deleteById(id);
        }
    }

**核心方法实现**

@Service
    public class HotelService {
        
    
        @Resource
        private RestHighLevelClient client;
    
        @Resource
        private HotelClient hotelClient;
    
        public void insertById(String id) {
        
            try {
        
                //1. 根据id查询酒店数据
                Hotel hotel = hotelClient.findById(id);
                HotelDoc hotelDoc = new HotelDoc(hotel);
                //2. 准备Request
                IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
    
                //3. 准备DSL
                request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
                //4. 发送请求
                client.index(request, RequestOptions.DEFAULT);
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }
    
        public void deleteById(String id) {
        
            try {
        
                //1. 准备 Request
                DeleteRequest request = new DeleteRequest("hotel", id);
                // 2.发送请求
                client.delete(request, RequestOptions.DEFAULT);
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }

## 三、测试 ##

**运行nacos**

startup.cmd -m standalone

**将hotel-service模块注册到nacos**

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

**访问页面，对酒店数据进行增删改查操作**

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

将第一条信息价格修改为399

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

**查看es中数据的变化**

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

**成功完成数据同步**

## 四、源码获取 ##

请联系 公众号 **程序员Bug终结者** 回复 es同步 获取源码及数据库文件

## ⛵小结 ##

以上就是【**Bug 终结者**】对 **分布式 SpringCloudAlibaba、Feign与RabbitMQ实现MySQL到ES数据同步** 的简单介绍，**ES搜索引擎无疑是最优秀的分布式搜索引擎，使用它，可大大提高项目的灵活、高效性！** **通过本文已了解 MySQL数据同步ES基本过程以及核心实现** **`技术改变世界！！！`**

> 如果这篇【文章】有帮助到你，希望可以给【**Bug 终结者**】点个赞👍，创作不易，如果有对【**后端技术**】、【**前端领域**】感兴趣的小可爱，也欢迎关注❤️❤️❤️ 【**Bug 终结者**】❤️❤️❤️，我将会给你带来巨大的【收获与惊喜】💝💝💝！

[cb3de9166b17431587028bde1c8fda80.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/8564bd0e03ba460a862435a74ba3beb0.png
[ae99d9db94fa442298f8acc635c16795.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/be493691f8754c7e92fb2a0263cdd20f.png
[8252db3b2fbf4233be4f78594252dd7f.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/a46108c333d54b8f89018799ed069bb6.png
[1895323c3e3046d8ae0d375cb25764d1.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/5d603f6b753f47dc84443fc1c29ec601.png
[3cac04763c484ee5831a06546314754e.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/1585d9e7aa8f488fa3f9a98debd35f47.png
[d7592ab254724814bae42aa6c4a0394b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/e2337976ef384eed89cf1211b0f8bb6c.png
[6690af057adf4cceb0995315b2a4e6c5.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/01d39755b7eb4795a99a28fc92049bd3.png
[8cde6dd73c9044b682ff310c3576a498.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/d0d6816fd8c44af2bf880451af328f82.png
[77f63ce53e904a0e98a9347781ea0386.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/01/bb2511ffea1d4d738144534da9705209.png