Elasticsearch使用——结合MybatisPlus使用ES & es和MySQL数据一致性 & 结合RabbitMQ实现解耦

Bertha 。 2024-03-02 10:58 25阅读 0赞

![在这里插入图片描述][73c346f5cd924016949305ae5dbcfa15.png]

## 前言 ##

本篇博客是一篇elasticsearch的使用案例，包括结合MybatisPlus使用ES，如何保证MySQL和es的数据一致性，另外使用了RabbitMQ进行解耦，自定义了发消息的方法。

其他相关的Elasticsearch的文章列表如下：

*  [Elasticsearch的Docker版本的安装和参数设置 & 端口开放和浏览器访问][Elasticsearch_Docker_ _]
 *  [Elasticsearch的可视化Kibana工具安装 & IK分词器的安装和使用][Elasticsearch_Kibana_ _ IK]
 *  [Elasticsearch的springboot整合 & Kibana进行全查询和模糊查询][Elasticsearch_springboot_ _ Kibana]

#### 目录 ####

*  前言
 *  引出
 *  结合MybatisPlus使用ES
 *   *  1.引入依赖
     *  2.进行配置
     *  3.实体类上加入注解
     *  4.创建操作的 Repository
     *  5.初始化es中的数据
     *  6.进行全查询以及分页
     *   *  带条件分页查询
 *  es和mysql的数据一致性
 *   *  延迟双删
     *  加锁的方式
 *  用rabbitmq进行解耦
 *   *  配置yml文件
     *  rabbitmq的配置类
     *  callback回调方法
     *  自定义发消息工具类
     *  进行消息的发送
     *  接收到消息，更新es
 *  总结

## 引出 ##

--------------------

1.elasticsearch的使用案例，包括结合MybatisPlus使用ES；  
2.如何保证MySQL和es的数据一致性；  
3.使用了RabbitMQ进行解耦，自定义了发消息的方法。

## 结合MybatisPlus使用ES ##

### 1.引入依赖 ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <scope>runtime</scope>
            </dependency>
    
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid-spring-boot-starter</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>com.baomidou</groupId>
                <artifactId>mybatis-plus-boot-starter</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
            </dependency>

### 2.进行配置 ###

package com.tianju.es.config;
    
    
    import org.elasticsearch.client.RestHighLevelClient;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.elasticsearch.client.ClientConfiguration;
    import org.springframework.data.elasticsearch.client.RestClients;
    import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
    
    /**
     * 你也可以不继承 AbstractElasticsearchConfiguration 类，而将 ESConfig 写成一般的配置类的型式。
     * 不过继承 AbstractElasticsearchConfiguration 好处在于，它已经帮我们配置好了elasticsearchTemplate 直接使用。
     */
    @Configuration
    public class ESConfig extends AbstractElasticsearchConfiguration {
        
        @Override
        public RestHighLevelClient elasticsearchClient() {
        
            ClientConfiguration clientConfiguration =
                    ClientConfiguration.builder()
                            .connectedTo("192.168.111.130:9200")
                            .build();
            return RestClients.create(clientConfiguration).rest();
        }
    }

### 3.实体类上加入注解 ###

![在这里插入图片描述][f6303b0cb37f43a3bde5d437a0549c93.png]

package com.tianju.es.entity;
    
    import com.baomidou.mybatisplus.annotation.IdType;
    import com.baomidou.mybatisplus.annotation.TableField;
    import com.baomidou.mybatisplus.annotation.TableId;
    import com.baomidou.mybatisplus.annotation.TableName;
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import org.springframework.data.elasticsearch.annotations.Document;
    import org.springframework.data.elasticsearch.annotations.Field;
    import org.springframework.data.elasticsearch.annotations.FieldType;
    
    import java.math.BigDecimal;
    
    /**
     * 产品，包括库存，价格信息
     */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @TableName("finance_sku")
    @Document(indexName = "finance_sku")
    public class FinanceSkuES {
        
        @TableId(value = "ID",type = IdType.AUTO)
        private Long id;
        @TableField("finance_sku_describe")
        @Field(index = true,analyzer = "ik_smart",
                searchAnalyzer = "ik_smart",type = FieldType.Text)
        private String detail; // 详情
        @TableField("finance_sku_price")
        private BigDecimal price;
        @TableField("finance_sku_stock")
        private Long stock;
        @TableField("finance_state")
        private Integer status;
    }

参数解释

@Document(indexName = "books", shards = 1, replicas = 0)
    @Data
    public class Book {
        
        @Id
        @Field(type = FieldType.Integer)
        private Integer id;
        
        @Field(type = FieldType.Keyword)
        private String title;
        
        @Field(type = FieldType.Text)
        private String press;
        
        @Field(type = FieldType.Keyword)
        private String author;
        
        @Field(type = FieldType.Keyword,index=false)
        private BigDecimal price;
        
        @Field(type = FieldType.Text)
        private String description;
    }

*  @Document ：注解会对实体中的所有属性建立索引；  
    indexName = “books” ：表示创建一个名称为 “books” 的索引；  
    shards = 1 ： 表示只使用一个分片；  
    replicas = 0 ： 表示不使用复制备份；  
    index = false： 不能索引查询
 *  @Field(type = FieldType.Keyword) ： 用以指定字段的数据类型。

### 4.创建操作的 Repository ###

![在这里插入图片描述][12f72394924c4ea3b4db70fb96f270e6.png]

从它的祖先们那里继承了大量的现成的方法，除此之外，它还可以按 spring data 的规则定义特定的方法。

![在这里插入图片描述][771ace1d3b904a1ca98ce12c9a20c60e.png]

package com.tianju.es.mapper;
    
    import com.tianju.es.entity.FinanceSkuES;
    import org.springframework.data.domain.Page;
    import org.springframework.data.domain.Pageable;
    import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
    import org.springframework.stereotype.Repository;
    
    /**
     * 操作es，类似于之前的mapper
     */
    @Repository
    public interface SkuESMapper extends ElasticsearchRepository<FinanceSkuES, Long> {
        
    
        /**
         * 根据关键字进行 分词 分页查询 sku数据
         * @param detail 查询条件
         * @param pageable 分页
         * @return
         */
        Page<FinanceSkuES> findFinanceSkuESByDetail(String detail, Pageable pageable);
    
        /**
         * 根据id进行删除
         * @param id
         */
        void removeFinanceSkuESById(Long id);
    
    }

### 5.初始化es中的数据 ###

![在这里插入图片描述][a789d651816b4ab6b406ff2c96b5c8fe.png]

运行的后台信息

![在这里插入图片描述][76f64c12e2654450832c1cddac867850.png]

查看es页面的信息，index management

![在这里插入图片描述][75c143a3e254411c8c2a87db3e7fcd0f.png]

### 6.进行全查询以及分页 ###

![在这里插入图片描述][09242468bc7e40d6b7d86791abc93ccc.png]

进行全查询

![在这里插入图片描述][82266e47b3244460abfc969f98aaf7cd.png]

{
        
      "content": [
        {
        
          "id": 1,
          "detail": "HUAWEI MateBook X Pro 2023 微绒典藏版 13代酷睿i7 32GB 2TB 14.2英寸3.1K原色全面屏 墨蓝",
          "price": 13999.0,
          "stock": 50,
          "status": 1
        },
        {
        
          "id": 2,
          "detail": "HUAWEI Mate 60 Pro+ 16GB+1TB 宣白",
          "price": 9999.0,
          "stock": 60,
          "status": 1
        },
        {
        
          "id": 3,
          "detail": "iPhone 15 Pro Max 超视网膜 XDR 显示屏",
          "price": 9299.0,
          "stock": 46,
          "status": 1
        },
        {
        
          "id": 4,
          "detail": "MacBook Air Apple M2 芯片 8 核中央处理器 8 核图形处理器 8GB 统一内存 256GB 固态硬盘",
          "price": 8999.0,
          "stock": 60,
          "status": 1
        }
      ],
      "pageable": {
        
        "sort": {
        
          "empty": true,
          "sorted": false,
          "unsorted": true
        },
        "offset": 0,
        "pageSize": 4,
        "pageNumber": 0,
        "paged": true,
        "unpaged": false
      },
      "totalElements": 4,
      "last": true,
      "totalPages": 1,
      "number": 0,
      "size": 4,
      "sort": {
        
        "empty": true,
        "sorted": false,
        "unsorted": true
      },
      "numberOfElements": 4,
      "first": true,
      "empty": false
    }

#### 带条件分页查询 ####

![在这里插入图片描述][6d28932e940c446b831a96cefa52e464.png]

注意分页查询的page从0开始，尝试发现需要输入分词器分词后最小单元，比如hu不是最小单元，而HUAWEI是

![在这里插入图片描述][ae11953c8f284a3b81d976b66ed72f58.png]

分词器进行分词的结果

![在这里插入图片描述][cff7b2431a764078b21d2d705b2ba004.png]

## es和mysql的数据一致性 ##

### 延迟双删 ###

![在这里插入图片描述][47128b2f1e0e47479ef8866ab7f4e5a4.png]

@Override
        public FinanceSkuES updateByIddDoubleDelete(FinanceSkuES financeSkuES) {
        
            // 把es看做是缓存，如何保证es 和 mysql的 数据一致性？
            // 延迟双删的模式
            // 1.先删除缓存 es
            skuESMapper.deleteAll();
            // 2.更新数据库 mysql
            updateById(financeSkuES);
            // 3.延时操作
            try {
        
                Thread.sleep(3000);
            } catch (InterruptedException e) {
        
                throw new RuntimeException(e);
            }
            // 4.再次删除缓存 es
            skuESMapper.deleteAll();
    
            // 5.最后更新缓存 es
            skuESMapper.saveAll(list());
            Optional<FinanceSkuES> byId = skuESMapper.findById(financeSkuES.getId());
            log.debug("byId: "+byId);
            return byId.get();
        }

上面代码有不妥的地方，我这里是修改，结果一开始直接从es中全部删除，应该是根据id把修改的数据删除，然后把修改好的数据set进es里面

### 加锁的方式 ###

感觉好像没什么用的样子，就是用了一下加锁

![在这里插入图片描述][46cb8a4f65224019835116e0a070d4a8.png]

## 用rabbitmq进行解耦 ##

![在这里插入图片描述][39288b17cb7c41208a32cbb3239372ad.png]

### 配置yml文件 ###

![在这里插入图片描述][baf2bc142e014709a9dc363649f81803.png]

spring:
      main:
        allow-circular-references: true
      datasource:
        driver-class-name: com.mysql.cj.jdbc.Driver
        ### 本地的数据库
        url: jdbc:mysql://127.0.0.1:3306/consumer_finance_product?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&allowMultiQueries=true
        username: root
        password: 123
    
      # redis的相关配置
      redis:
        host: 119.3.162.127
        port: 6379
        database: 0
        password: Pet3927
    
      # rabbitmq相关
      rabbitmq:
        host: 192.168.111.130
        port: 5672
        username: admin
        password: 123
        virtual-host: /test
    
        # 生产者保证消息可靠性
        publisher-returns: true
        publisher-confirm-type: correlated
    
        # 设置手动确认
        listener:
          simple:
            acknowledge-mode: manual

### rabbitmq的配置类 ###

![在这里插入图片描述][9de20132ecc3460f858ed046b3b9d1d5.png]

> 将Java对象转换成json字符串传输

![在这里插入图片描述][28f3ae793cf649a88dc0aacb46b1a250.png]

package com.tianju.es.rabbit;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitConfig {
        
    
        public static final String ES_EXCHANGE = "es_exchange";
        public static final String ES_QUEUE = "es_queue";
        public static final String ES_KEY = "es_key";
    
        @Bean
        public DirectExchange directExchange(){
        
            return new DirectExchange(ES_EXCHANGE);
        }
    
        @Bean
        public Queue esQueue(){
        
            return new Queue(ES_QUEUE);
        }
    
        @Bean
        public Binding esQueueToDirectExchange(){
        
            return BindingBuilder.bind(esQueue())
                    .to(directExchange())
                    .with(ES_KEY);
        }
    
        /**
         * 将对象转换为json字符串
         * @return
         */
        @Bean
        public MessageConverter messageConverter(){
        
            return  new Jackson2JsonMessageConverter();
        }
    
        @Bean
        public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            rabbitTemplate.setMessageConverter(messageConverter());// 修改转换器
            return rabbitTemplate;
        }
    
    }

![在这里插入图片描述][eaa8c157a890448ba43fac8128f3ae5c.png]

### callback回调方法 ###

package com.tianju.es.rabbit;
    
    import com.alibaba.fastjson2.JSON;
    import com.alibaba.fastjson2.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    
    /**
     * 生产者消息可靠性
     */
    // RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback
    @Configuration
    @Slf4j
    public class CallbackConfig
            implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        // 初始化
        @PostConstruct
        public void init(){
        
            rabbitTemplate.setConfirmCallback(this);
            rabbitTemplate.setReturnCallback(this);
            rabbitTemplate.setMandatory(true);
        }
    
        /**
         * 不管成功或者失败都会执行
         * @param correlationData correlation对象需要在 发送消息时候 给
         * @param ack true表示成功，false表示发送失败
         * @param cause 如果失败的话，会写失败原因；如果成功，返回为null
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
            log.debug("ack是否成功："+ack);
            log.debug("cause信息："+cause);
    
            if (correlationData!=null){
        
                JSONObject jsonObject = JSON.parseObject(correlationData.getReturnedMessage().getBody());
                String exchange = correlationData.getReturnedMessage().getMessageProperties().getReceivedExchange();
                String routingKey = correlationData.getReturnedMessage().getMessageProperties().getReceivedRoutingKey();
                log.debug("消息体："+jsonObject);
                log.debug("交换机："+exchange);
                log.debug("路由key："+routingKey);
            }
    
    
            if (ack){
        
                return;
            }
    
            // 失败了
    
            // 1、重试重试上限次数（默认值5）每重试一次时间间隔会增加
            // 2、把消息、交换机名称、路由键等相关的消息保存到数据库，有一个程序定时扫描相关的消息，然后重新发送消息。
            // 重发上限次数（默认值5）超过阈值会转人工处理
    
            // 2、把消息体、交换机名称、路由键等相关的消息保存到数据库，有一个程序定时扫描相关的消息，然后重新发送消息。
            // 重发上限次数（默认值5）超过阈值会转人工处理
            // 2.1需要把相关的信息存放到数据中，表字段：消息体、交换机名称、路由键、状态、次数
            // 2.2定时任务（单体：spring定时任务  分布式：XxL-job),发送消息
    
    
    
    
        }
    
    
        /**
         * 只有失败了才会执行
         * @param message
         * @param replyCode
         * @param replyText
         * @param exchange
         * @param routingKey
         */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
            // 2、把消息、交换机名称、路由键等相关的消息保存到数据库，有一个程序定时扫描相关的消息，然后重新发送消息。
    
        }
    }

### 自定义发消息工具类 ###

![在这里插入图片描述][cd4053cd32bb4dfdb38d33c2af69d772.png]

package com.tianju.common.util;
    
    
    import com.alibaba.fastjson2.JSON;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.data.redis.core.StringRedisTemplate;
    
    
    @Slf4j
    public class RabbitUtil {
        
    
        /**
         * 延迟队列，发送消息，到达时间后进入死信队列中
         * @param rabbitTemplate 调用的rabbitTemplate
         * @param redisTemplate 用来在redis里面存token
         * @param msg 发送的消息
         * @param token 发送的token，用于保证幂等性
         * @param ttl 如果是延迟消费，则消息的过期时间，到达改时间后进入死信交换机，到死信队列中
         * @param exchange 交换机名字
         * @param routingKey 路由键名字
         * @param <T> 发送消息的实体类
         */
        public static  <T> void sendMsg(RabbitTemplate rabbitTemplate,
                                        StringRedisTemplate redisTemplate,
                                        T msg,String token,Integer ttl,
                                        String exchange,String routingKey) {
        
    
            log.debug("给交换机[{}]通过路由键[{}]发送消息 {}，token为{}",exchange,routingKey,msg,token);
    
            MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
        
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
        
                    redisTemplate.opsForValue().set(token, token,5*60000);
                    message.getMessageProperties().setMessageId(token);
                    if (ttl!=null){
        
                        message.getMessageProperties().setExpiration(ttl.toString());
                    }
                    return message;
                }
            };
    
            CorrelationData correlationData = new CorrelationData();
            // 消息体
            Message message = new Message(JSON.toJSONBytes(msg));
            // 交换机名称
            message.getMessageProperties().setReceivedExchange(exchange);
            // 路由键
            message.getMessageProperties().setReceivedRoutingKey(routingKey);
            correlationData.setReturnedMessage(message);
    
            // 发送MQ消息
            rabbitTemplate.convertAndSend(exchange, // 发给交换机
                    routingKey, // 根据这个routingKey就会给到TTL队列，到时间成死信，发给死信交换机，到死信队列
                    msg,
                    messagePostProcessor,
                    correlationData
            );
        }
    }

### 进行消息的发送 ###

![在这里插入图片描述][e60fa965f1cc4179bf5468917ff37144.png]

> 接口

package com.tianju.es.service;
    
    import com.baomidou.mybatisplus.extension.service.IService;
    import com.tianju.es.entity.FinanceSkuES;
    
    public interface SkuService extends IService<FinanceSkuES> {
        
    
        /**
         * 延迟双删的方式，保证es 缓存 和 mysql数据库的数据一致性
         * @param financeSkuES 修改的数据
         * @return
         */
        FinanceSkuES updateByIddDoubleDelete(FinanceSkuES financeSkuES);
    
        /**
         * 加锁的方式，不过感觉没啥用的样子
         * @param financeSkuES
         * @return
         */
        FinanceSkuES updateByIdRedisLock(FinanceSkuES financeSkuES);
    
        /**
         * 通过rabbitmq进行解耦
         * @param financeSkuES
         * @return
         */
        String updateByIdRabbitMQ(FinanceSkuES financeSkuES);
    }

> 实现类

package com.tianju.es.service.impl;
    
    import cn.hutool.core.util.IdUtil;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.tianju.common.util.RabbitUtil;
    import com.tianju.es.entity.FinanceSkuES;
    import com.tianju.es.mapper.SkuESMapper;
    import com.tianju.es.mapper.SkuMybatisPlusMapper;
    import com.tianju.es.rabbit.RabbitConfig;
    import com.tianju.es.service.SkuService;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Service;
    
    import java.util.Collection;
    import java.util.Optional;
    import java.util.UUID;
    
    @Service
    public class SkuServiceImpl extends ServiceImpl<SkuMybatisPlusMapper,FinanceSkuES>
            implements SkuService {
        
    
        @Autowired
        private SkuESMapper skuESMapper;
    
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Override
        public FinanceSkuES updateByIddDoubleDelete(FinanceSkuES financeSkuES) {
        
            // 把es看做是缓存，如何保证es 和 mysql的 数据一致性？
            // 延迟双删的模式
            // 1.先删除缓存 es
            skuESMapper.deleteAll();
            // 2.更新数据库 mysql
            updateById(financeSkuES);
            // 3.延时操作
            try {
        
                Thread.sleep(3000);
            } catch (InterruptedException e) {
        
                throw new RuntimeException(e);
            }
            // 4.再次删除缓存 es
            skuESMapper.deleteAll();
    
            // 5.最后更新缓存 es
            skuESMapper.saveAll(list());
            Optional<FinanceSkuES> byId = skuESMapper.findById(financeSkuES.getId());
            log.debug("byId: "+byId);
            return byId.get();
        }
    
        @Override
        public FinanceSkuES updateByIdRedisLock(FinanceSkuES financeSkuES) {
        
            // 第二种方式加锁
            String uuid = UUID.randomUUID().toString();
            // 相当于setnx指令
            Boolean skuLock = stringRedisTemplate.opsForValue().setIfAbsent("skuLock", uuid);
            try {
        
                if (skuLock){
         // 抢到了锁
                    skuESMapper.deleteAll();
                    updateById(financeSkuES);
                }
            }finally {
        
                if (uuid.equals(stringRedisTemplate.opsForValue().get("skuLock"))){
        
                    stringRedisTemplate.delete("skuLock");
                }
            }
            skuESMapper.saveAll(list());
            Optional<FinanceSkuES> byId = skuESMapper.findById(financeSkuES.getId());
            log.debug("byId: "+byId);
            return byId.get();
        }
    
        @Override
        public String updateByIdRabbitMQ(FinanceSkuES financeSkuES) {
        
            // 采用rabbitmq进行解耦
    
            updateById(financeSkuES);
            FinanceSkuES skuES = getById(financeSkuES.getId());
            String uuid = IdUtil.fastUUID();
    
            RabbitUtil.sendMsg(
                    rabbitTemplate,stringRedisTemplate,skuES,uuid,null,
                    RabbitConfig.ES_EXCHANGE,RabbitConfig.ES_KEY
            );
    
            return "已经发送消息："+skuES;
        }
    }

![在这里插入图片描述][ac022a83b3cd499da4387a6952c73759.png]

### 接收到消息，更新es ###

接收到消息进行es的更新，把原来的删除，把最新的set进去

![在这里插入图片描述][22594ccc90a14482a3ba6310f7ee9ce5.png]

package com.tianju.es.rabbit;
    
    import com.rabbitmq.client.Channel;
    import com.tianju.es.entity.FinanceSkuES;
    import com.tianju.es.mapper.SkuESMapper;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    
    @Slf4j
    @Component
    public class ESListener {
        
    
        @Autowired
        private StringRedisTemplate redisTemplate;
    
        @Autowired
        private SkuESMapper skuESMapper;
    
        @RabbitListener(queues = RabbitConfig.ES_QUEUE)
        public void esUpdate(FinanceSkuES financeSkuES, Message message, Channel channel) {
        
            String messageId = message.getMessageProperties().getMessageId();
            log.debug("进行业务----> 监听到队列{}的消息，messageId为{}",financeSkuES,messageId);
    
            try {
        
                // 幂等性
                if (redisTemplate.delete(messageId)){
        
                    // 根据id删除原有的 es 数据
                    // 然后把新的数据set进来
                    log.debug("处理es的业务，删除原有的，替换最新的");
                    skuESMapper.removeFinanceSkuESById(financeSkuES.getId());
                    skuESMapper.save(financeSkuES);
                }
    
                // 手动签收消息
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    
            }catch (Exception e){
        
                // 幂等性
                redisTemplate.opsForValue().set(messageId,messageId,5*60000);
    
                // 1、重试重试上限次数（默认值5） 每重试一次时间间隔会增加
                // 2、把消息、交换机名称、路由键等相关的消息保存到数据库，有一个程序定时扫描相关的消息，然后重新发送消息。
                // 重发上限次数（默认值5）超过阈值会转人工处理
    
                // 已知的消息，交换机，路由器，消息 message.getBody()  消息发送给的是监听的队列
    
                try {
        
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException ex) {
        
                    throw new RuntimeException(ex);
                }
            }
        }
    }

![在这里插入图片描述][726a03ed0e3d418b94b0fa27b87b3956.png]

后台打印的日志

![在这里插入图片描述][c9917b13e4e043eba80b5c82b7bb009f.png]

--------------------

## 总结 ##

1.elasticsearch的使用案例，包括结合MybatisPlus使用ES；  
2.如何保证MySQL和es的数据一致性；  
3.使用了RabbitMQ进行解耦，自定义了发消息的方法。

[73c346f5cd924016949305ae5dbcfa15.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/61ae6dc298904387835dda2a2f87c2d6.png
[Elasticsearch_Docker_ _]: https://blog.csdn.net/Pireley/article/details/133268762
[Elasticsearch_Kibana_ _ IK]: https://blog.csdn.net/Pireley/article/details/133485501
[Elasticsearch_springboot_ _ Kibana]: https://blog.csdn.net/Pireley/article/details/133513298
[f6303b0cb37f43a3bde5d437a0549c93.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/c9d56747208b49569b6ea0db9613785e.png
[12f72394924c4ea3b4db70fb96f270e6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/e0ee3a1589ed4e109932ba9be3a37470.png
[771ace1d3b904a1ca98ce12c9a20c60e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/a02192a019d64ecf9c0c6f113acf4fc4.png
[a789d651816b4ab6b406ff2c96b5c8fe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/4a7adbbb82cc40109356ea8ef26ba959.png
[76f64c12e2654450832c1cddac867850.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/073ceae464854db2b9ca5348ef7cfd32.png
[75c143a3e254411c8c2a87db3e7fcd0f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/5096c29c441e473f80d2473b834188c4.png
[09242468bc7e40d6b7d86791abc93ccc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/68492eddab60433f94fb61b3cb1d1805.png
[82266e47b3244460abfc969f98aaf7cd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/4826f17113654e468300fd9dccfc1228.png
[6d28932e940c446b831a96cefa52e464.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/6f835202123e45be8503f900cc8f406f.png
[ae11953c8f284a3b81d976b66ed72f58.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/da1a43d7cd1f4edd897d95a2dc4d15b2.png
[cff7b2431a764078b21d2d705b2ba004.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/6362ed9303d04548a4e6996c3fd7298a.png
[47128b2f1e0e47479ef8866ab7f4e5a4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/6a21955c75494d6f873e1f7bd9ff62dc.png
[46cb8a4f65224019835116e0a070d4a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/6cb9cf6776f04d81a8283792738f5426.png
[39288b17cb7c41208a32cbb3239372ad.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/57aa506b8e954989a627158332d5b3e0.png
[baf2bc142e014709a9dc363649f81803.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/c3570cfb027140479a615c60476ff2ee.png
[9de20132ecc3460f858ed046b3b9d1d5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/b2a71070cc8e4b909f041fdcfcbf0763.png
[28f3ae793cf649a88dc0aacb46b1a250.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/a476eb7a9bff4b64b64bfb8a16208adb.png
[eaa8c157a890448ba43fac8128f3ae5c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/3b97afbb862d474cb02ce4737cc92ef1.png
[cd4053cd32bb4dfdb38d33c2af69d772.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/b8a84a3908f944d4a84c3c72f779b538.png
[e60fa965f1cc4179bf5468917ff37144.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/2e4b27db8ed7499dac291e96a85df3c9.png
[ac022a83b3cd499da4387a6952c73759.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/097e933e68d84ea396affc8f978945e4.png
[22594ccc90a14482a3ba6310f7ee9ce5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/914efb39f51443ccb69985b089adfaf6.png
[726a03ed0e3d418b94b0fa27b87b3956.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/644c60971b24472fa11ee56c8d9065f5.png
[c9917b13e4e043eba80b5c82b7bb009f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/02/9f2cf11900b44758a7c2d393f310115e.png