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

news2024/11/23 10:04:55

在项目中，缓存是提高应用性能和响应速度的关键手段之一。然而，当多个模块在短时间内发布工单并且需要清理同一个接口的缓存时，容易引发缓存清理冲突，导致缓存失效的问题。为了解决这一难题，我们采用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;
    }
}