Redisson实现简单消息队列：优雅解决缓存清理冲突

梦里梦外; 2023-10-13 23:21 6阅读 0赞

在项目中，缓存是提高应用性能和响应速度的关键手段之一。然而，当多个模块在短时间内发布工单并且需要清理同一个接口的缓存时，容易引发缓存清理冲突，导致缓存失效的问题。为了解决这一难题，我们采用Redisson的消息队列功能，实现了一个简单而高效的消息队列，优雅地解决了缓存清理冲突问题。本文将为您详细介绍Redisson实现简单消息队列的方案，以及如何在项目中使用它来优化缓存清理。

#### 第一部分：缓存清理冲突的挑战 ####

在高并发场景下，多个模块可能同时发布工单，并且这些工单可能会导致同一个接口的缓存失效。当多个模块同时发起缓存清理请求时，可能会造成数据库压力增大及缓存不一致的问题，降低应用程序的性能和稳定性。这种情况下，我们需要一种优雅的解决方案来协调缓存清理的行为。

#### 第二部分：Redisson简介 ####

Redisson是一个功能强大的Java库，基于Redis构建，它提供了分布式数据结构和服务，以及异步处理的解决方案。其中，消息队列是Redisson提供的一项强大功能，用于在多个模块之间实现高效的消息传递和处理。

#### 第三部分：使用Redisson消息队列解决方案 ####

为了解决缓存清理冲突问题，我们选择使用Redisson的消息队列功能，具体步骤如下：

创建Redisson客户端： 首先，我们需要创建一个Redisson客户端，用于连接到Redis服务器。

创建消息队列： 接下来，我们创建一个Redisson的队列，用于存放缓存清理请求消息。

发布缓存清理请求： 在每个模块发布工单时，我们将对应的缓存清理请求添加到Redisson队列中。

消费缓存清理请求： 我们创建一个定时任务，周期性的从Redisson队列中获取缓存清理请求消息。

处理缓存清理冲突： 合并消息，并执行缓存清理操作，避免多个模块同时清理同一个接口的缓存。

#### 第三部分：代码示例 ####

redisson配置文件

# 项目相关Spring redis配置
    spring:
      redisson:
        # 地址
        clusters: 192.168.10.106:6479,192.168.10.106:6579,192.168.10.106:6679
        # 密码
        password: 123456
        # 连接超时时间
        timeout: 10s

客户端连接类： RedissonManager

package cn.xj.redis;
    
    import org.redisson.Redisson;
    import org.redisson.api.RedissonClient;
    import org.redisson.client.codec.StringCodec;
    import org.redisson.config.Config;
    import org.redisson.config.ReadMode;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    /**
     * description: redisson初始化
     *
     * @author xj
     * <p>
     * create on 2020-04-09 17:21
     **/
    @Configuration
    public class RedissonManager {
        
    
        @Value("${spring.redisson.clusters}")
        private  String cluster;
        @Value("${spring.redisson.password}")
        private String password;
    
    
    
    
        @Bean(name="cacheCluster")
        public RedissonClient getRedissonCluster(){
        
            String[] nodes = cluster.split(",");
            //redisson版本是3.5，集群的ip前面要加上“redis://”，不然会报错，3.2版本可不加
            for(int i=0;i<nodes.length;i++){
        
                nodes[i] = "redis://"+nodes[i];
            }
            RedissonClient redisson = null;
            Config config = new Config();
            //设置
            config.setCodec(new StringCodec())
                    //这是用的集群server
                    .useClusterServers()
                    //设置集群状态扫描时间
                    .setScanInterval(2000)
                    .addNodeAddress(nodes)
                    .setPassword(password)
                    .setReadMode(ReadMode.MASTER);;
            redisson = Redisson.create(config);
            return redisson;
        }
    
    }

redis工具类:RedissonCache

package cn.xj.redis;
    
    
    import com.alibaba.fastjson2.JSON;
    import com.alibaba.fastjson2.JSONObject;
    import com.alibaba.fastjson2.TypeReference;
    import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.*;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.stereotype.Component;
    
    import java.util.Map;
    import java.util.Set;
    
    
    /**
     * redisson工具类
     * company
     * @author
     * @data 2020-04-01
     */
    @Component
    @Slf4j
    public class RedissonCache {
        
    
    
    
        @Autowired
        @Qualifier("cacheCluster")
        private RedissonClient cacheCluster;
    
        public boolean cacheAdd(String key, Map<String,Set<String>> message){
        
            String msgJsonstr = JSONObject.toJSONString(message);
            RQueue<String> queue = cacheCluster.getQueue(key);
            return queue.add(msgJsonstr);
        }
    
        public Map<String, Set<String>> cachePoll(String key){
        
            RQueue<String> queue = cacheCluster.getQueue(key);
            String msgJsonstr = queue.poll();
            return JSON.parseObject(msgJsonstr, new TypeReference<Map<String, Set<String>>>() {
        });
        }
    
    }

功能示例类：QueueController

package cn.xj.redis;
    
    import org.apache.commons.lang3.ObjectUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestBody;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    @RestController
    public class QueueController {
        
    
        private static final String KEY= "xj_test_queue";
        @Autowired
        private RedissonCache redissonCache;
    
        /**
         * 添加消息队列
         * @param params
         */
        @PostMapping("/redis/queue/add")
        public void add(@RequestBody Map<String, Set<String>> params){
        
            redissonCache.cacheAdd(KEY,params);
        }
    
    
        /**
         * 消费队列中的消息
         * @return
         */
        @GetMapping("/redis/queue/poll")
        public List<Map<String, Set<String>>> poll(){
        
            List<Map<String, Set<String>>> result = new ArrayList<>();
            //每次最大消费条数
            int batchSize = 200;
            //获取消费的所有消息
            Map<String, Set<String>> msgMap = redissonCache.cachePoll(KEY);
            while (batchSize > 0 && !ObjectUtils.isEmpty(msgMap)) {
        
                result.add(msgMap);
                batchSize--;
                msgMap = redissonCache.cachePoll(KEY);
            }
    
            try {
        
                //消息合并、处理逻辑
    
                //此处省略代码一万行
            } catch (Exception e){
        
                
                //消息处理失败的逻辑
            }
            return result;
        }
    }

我们调用模拟添加接口可以看到。每次有新的消息都是给queque的尾部添加

![\_20230724204722.png][20230724204722.png]

每次消费的时候是queque的头部开始取数据

#### 第五部分：总结和应用场景及注意事项 ####

通过Redisson的消息队列功能，我们成功实现了一个简单而高效的缓存清理解决方案。该方案有效解决了多个模块同时发布工单导致缓存清理冲突的问题，提高了应用程序的性能和稳定性。

适用场景：

*  多个模块在短时间内发布工单，并需要清理同一个接口的缓存。
 *  消息队列可以将不同模块或不同组件之间的通信解耦，实现系统的高内聚和低耦合。同时，在系统面临突发大量请求时，消息队列可以进行削峰填谷，保护系统不受过载影响。

注意事项：

*  在使用队列时，要考虑队列的容量，避免队列过大导致内存压力过大或队列过小导致消息丢失。
 *  在使用消息队列时，需要小心处理异常情况。例如，如果消息处理失败，可以将消息重新排队或将其放入一个死信队列，以便稍后进行处理。

#### 总结 ####

Redisson的消息队列是解决缓存清理冲突问题的优雅方案，通过其强大的功能，我们可以简单地实现消息传递和处理，从而优化应用程序的性能。在日常开发中，合理应用Redisson的消息队列功能，能够帮助我们处理更多类似的并发问题，提升应用程序的可靠性和扩展性。

希望本文能够为读者提供有益的参考，让您在项目中更加灵活和高效地使用Redisson实现简单消息队列。愿您的应用程序在缓存清理中更上一层楼，助您的项目更加稳健发展！