springBoot 整合redisson实现消息订阅详细教程

• 官方文档 Redisson 官方文档 - 6.7. 话题（订阅分发）

 maven 引入依赖

    org.redisson
    redisson-spring-boot-starter
    3.17.5

 yml配置文件

spring:
  redis:
    # 地址
    host: localhost
    # 端口，默认为6379
    port: 6379
    # 数据库索引
    database: 0
    # 密码(如没有密码请注释掉)
    # password:
    # 连接超时时间
    timeout: 10s
    # 是否开启ssl
    ssl: false
 
redisson:
  # redis key前缀
  keyPrefix:
  # 线程池数量
  threads: 4
  # Netty线程池数量
  nettyThreads: 8
  # 单节点配置
  singleServerConfig:
    # 客户端名称
    clientName: clientName
    # 最小空闲连接数
    connectionMinimumIdleSize: 8
    # 连接池大小
    connectionPoolSize: 32
    # 连接空闲超时，单位：毫秒
    idleConnectionTimeout: 10000
    # 命令等待超时，单位：毫秒
    timeout: 3000
    # 发布和订阅连接池大小
    subscriptionConnectionPoolSize: 50

RedissonProperties #映射yml配置文件

package com.example.demo.redisson;
 
import lombok.Data;
import lombok.NoArgsConstructor;
import org.redisson.config.ReadMode;
import org.redisson.config.SubscriptionMode;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:38
 * redisson 配置文件
 */
@Data
@Component
@ConfigurationProperties(prefix = "redisson")
public class RedissonProperties {
 
    /**
     * redis缓存key前缀
     */
    private String keyPrefix;
 
    /**
     * 线程池数量,默认值 = 当前处理核数量 * 2
     */
    private int threads;
 
    /**
     * Netty线程池数量,默认值 = 当前处理核数量 * 2
     */
    private int nettyThreads;
 
    /**
     * 单机服务配置
     */
    private SingleServerConfig singleServerConfig;
 
    /**
     * 集群服务配置
     */
    private ClusterServersConfig clusterServersConfig;
 
    @Data
    @NoArgsConstructor
    public static class SingleServerConfig {
 
        /**
         * 客户端名称
         */
        private String clientName;
 
        /**
         * 最小空闲连接数
         */
        private int connectionMinimumIdleSize;
 
        /**
         * 连接池大小
         */
        private int connectionPoolSize;
 
        /**
         * 连接空闲超时，单位：毫秒
         */
        private int idleConnectionTimeout;
 
        /**
         * 命令等待超时，单位：毫秒
         */
        private int timeout;
 
        /**
         * 发布和订阅连接池大小
         */
        private int subscriptionConnectionPoolSize;
 
    }
 
    @Data
    @NoArgsConstructor
    public static class ClusterServersConfig {
 
        /**
         * 客户端名称
         */
        private String clientName;
 
        /**
         * master最小空闲连接数
         */
        private int masterConnectionMinimumIdleSize;
 
        /**
         * master连接池大小
         */
        private int masterConnectionPoolSize;
 
        /**
         * slave最小空闲连接数
         */
        private int slaveConnectionMinimumIdleSize;
 
        /**
         * slave连接池大小
         */
        private int slaveConnectionPoolSize;
 
        /**
         * 连接空闲超时，单位：毫秒
         */
        private int idleConnectionTimeout;
 
        /**
         * 命令等待超时，单位：毫秒
         */
        private int timeout;
 
        /**
         * 发布和订阅连接池大小
         */
        private int subscriptionConnectionPoolSize;
 
        /**
         * 读取模式
         */
        private ReadMode readMode;
 
        /**
         * 订阅模式
         */
        private SubscriptionMode subscriptionMode;
 
    }
 
}

RedisConfig #redis配置

package com.example.demo.redisson;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:37
 * redis配置
 */
 
import cn.hutool.core.util.ObjectUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.spring.starter.RedissonAutoConfigurationCustomizer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Slf4j
@Configuration
@EnableCaching
@EnableConfigurationProperties(RedissonProperties.class)
public class RedisConfig extends CachingConfigurerSupport {
 
    @Autowired
    private RedissonProperties redissonProperties;
 
    @Autowired
    private ObjectMapper objectMapper;
 
    @Bean
    public RedissonAutoConfigurationCustomizer redissonCustomizer() {
        return config -> {
            config.setThreads(redissonProperties.getThreads())
                    .setNettyThreads(redissonProperties.getNettyThreads())
                    .setCodec(new JsonJacksonCodec(objectMapper));
            RedissonProperties.SingleServerConfig singleServerConfig = redissonProperties.getSingleServerConfig();
            if (ObjectUtil.isNotNull(singleServerConfig)) {
                // 使用单机模式
                config.useSingleServer()
                        //设置redis key前缀
                        .setNameMapper(new KeyPrefixHandler(redissonProperties.getKeyPrefix()))
                        .setTimeout(singleServerConfig.getTimeout())
                        .setClientName(singleServerConfig.getClientName())
                        .setIdleConnectionTimeout(singleServerConfig.getIdleConnectionTimeout())
                        .setSubscriptionConnectionPoolSize(singleServerConfig.getSubscriptionConnectionPoolSize())
                        .setConnectionMinimumIdleSize(singleServerConfig.getConnectionMinimumIdleSize())
                        .setConnectionPoolSize(singleServerConfig.getConnectionPoolSize());
            }
            // 集群配置方式 参考下方注释
            RedissonProperties.ClusterServersConfig clusterServersConfig = redissonProperties.getClusterServersConfig();
            if (ObjectUtil.isNotNull(clusterServersConfig)) {
                config.useClusterServers()
                        //设置redis key前缀
                        .setNameMapper(new KeyPrefixHandler(redissonProperties.getKeyPrefix()))
                        .setTimeout(clusterServersConfig.getTimeout())
                        .setClientName(clusterServersConfig.getClientName())
                        .setIdleConnectionTimeout(clusterServersConfig.getIdleConnectionTimeout())
                        .setSubscriptionConnectionPoolSize(clusterServersConfig.getSubscriptionConnectionPoolSize())
                        .setMasterConnectionMinimumIdleSize(clusterServersConfig.getMasterConnectionMinimumIdleSize())
                        .setMasterConnectionPoolSize(clusterServersConfig.getMasterConnectionPoolSize())
                        .setSlaveConnectionMinimumIdleSize(clusterServersConfig.getSlaveConnectionMinimumIdleSize())
                        .setSlaveConnectionPoolSize(clusterServersConfig.getSlaveConnectionPoolSize())
                        .setReadMode(clusterServersConfig.getReadMode())
                        .setSubscriptionMode(clusterServersConfig.getSubscriptionMode());
            }
            log.info("初始化 redis 配置");
        };
    }
 
    /**
     * 自定义缓存管理器 整合spring-cache
     */
    @Bean
    public CacheManager cacheManager() {
        return new PlusSpringCacheManager();
    }
}

KeyPrefixHandler #redis缓存key前缀处理

package com.example.demo.redisson;
 
import com.example.demo.util.StringUtils;
import org.redisson.api.NameMapper;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:40
 * redis缓存key前缀处理
 */
public class KeyPrefixHandler implements NameMapper {
    private final String keyPrefix;
 
    public KeyPrefixHandler(String keyPrefix) {
        //前缀为空 则返回空前缀
        this.keyPrefix = StringUtils.isBlank(keyPrefix) ? "" : keyPrefix + ":";
    }
 
    /**
     * 增加前缀
     */
    @Override
    public String map(String name) {
        if (StringUtils.isBlank(name)) {
            return null;
        }
        if (StringUtils.isNotBlank(keyPrefix) && !name.startsWith(keyPrefix)) {
            return keyPrefix + name;
        }
        return name;
    }
 
    /**
     * 去除前缀
     */
    @Override
    public String unmap(String name) {
        if (StringUtils.isBlank(name)) {
            return null;
        }
        if (StringUtils.isNotBlank(keyPrefix) && name.startsWith(keyPrefix)) {
            return name.substring(keyPrefix.length());
        }
        return name;
    }
}

PlusSpringCacheManager # 修改 RedissonSpringCacheManager 源码、 重写 cacheName 处理方法 支持多参数

package com.example.demo.redisson;
 
import org.redisson.api.RMap;
import org.redisson.api.RMapCache;
import org.redisson.spring.cache.CacheConfig;
import org.redisson.spring.cache.RedissonCache;
import org.springframework.boot.convert.DurationStyle;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.cache.transaction.TransactionAwareCacheDecorator;
import org.springframework.util.StringUtils;
 
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:43
 *  修改 RedissonSpringCacheManager 源码
 *  重写 cacheName 处理方法 支持多参数
 */
public class PlusSpringCacheManager implements CacheManager {
 
    private boolean dynamic = true;
 
    private boolean allowNullValues = true;
 
    private boolean transactionAware = true;
 
    Map configMap = new ConcurrentHashMap<>();
    ConcurrentMap instanceMap = new ConcurrentHashMap<>();
 
    /**
     * Creates CacheManager supplied by Redisson instance
     */
    public PlusSpringCacheManager() {
    }
 
 
    /**
     * Defines possibility of storing {@code null} values.
     * 
     * Default is true
     *
     * @param allowNullValues stores if true
     */
    public void setAllowNullValues(boolean allowNullValues) {
        this.allowNullValues = allowNullValues;
    }
 
    /**
     * Defines if cache aware of Spring-managed transactions.
     * If {@code true} put/evict operations are executed only for successful transaction in after-commit phase.
     * 
     * Default is false
     *
     * @param transactionAware cache is transaction aware if true
     */
    public void setTransactionAware(boolean transactionAware) {
        this.transactionAware = transactionAware;
    }
 
    /**
     * Defines 'fixed' cache names.
     * A new cache instance will not be created in dynamic for non-defined names.
     * 
     * `null` parameter setups dynamic mode
     *
     * @param names of caches
     */
    public void setCacheNames(Collection names) {
        if (names != null) {
            for (String name : names) {
                getCache(name);
            }
            dynamic = false;
        } else {
            dynamic = true;
        }
    }
 
    /**
     * Set cache config mapped by cache name
     *
     * @param config object
     */
    public void setConfig(Map config) {
        this.configMap = (Map) config;
    }
 
    protected CacheConfig createDefaultConfig() {
        return new CacheConfig();
    }
 
    @Override
    public Cache getCache(String name) {
        Cache cache = instanceMap.get(name);
        if (cache != null) {
            return cache;
        }
        if (!dynamic) {
            return cache;
        }
 
        CacheConfig config = configMap.get(name);
        if (config == null) {
            config = createDefaultConfig();
            configMap.put(name, config);
        }
 
        // 重写 cacheName 支持多参数
        String[] array = StringUtils.delimitedListToStringArray(name, "#");
        name = array[0];
        if (array.length > 1) {
            config.setTTL(DurationStyle.detectAndParse(array[1]).toMillis());
        }
        if (array.length > 2) {
            config.setMaxIdleTime(DurationStyle.detectAndParse(array[2]).toMillis());
        }
        if (array.length > 3) {
            config.setMaxSize(Integer.parseInt(array[3]));
        }
 
        if (config.getMaxIdleTime() == 0 && config.getTTL() == 0 && config.getMaxSize() == 0) {
            return createMap(name, config);
        }
 
        return createMapCache(name, config);
    }
 
    private Cache createMap(String name, CacheConfig config) {
        RMap map = RedisUtils.getClient().getMap(name);
 
        Cache cache = new RedissonCache(map, allowNullValues);
        if (transactionAware) {
            cache = new TransactionAwareCacheDecorator(cache);
        }
        Cache oldCache = instanceMap.putIfAbsent(name, cache);
        if (oldCache != null) {
            cache = oldCache;
        }
        return cache;
    }
 
    private Cache createMapCache(String name, CacheConfig config) {
        RMapCache map = RedisUtils.getClient().getMapCache(name);
 
        Cache cache = new RedissonCache(map, config, allowNullValues);
        if (transactionAware) {
            cache = new TransactionAwareCacheDecorator(cache);
        }
        Cache oldCache = instanceMap.putIfAbsent(name, cache);
        if (oldCache != null) {
            cache = oldCache;
        } else {
            map.setMaxSize(config.getMaxSize());
        }
        return cache;
    }
 
    @Override
    public Collection getCacheNames() {
        return Collections.unmodifiableSet(configMap.keySet());
    }
 
 
}

SpringUtils #spring工具类

package com.example.demo.util;
 
import cn.hutool.extra.spring.SpringUtil;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.stereotype.Component;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:45
 * spring 工具类
 */
@Component
public final class SpringUtils extends SpringUtil {
 
    /**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name) {
        return getBeanFactory().containsBean(name);
    }
 
    /**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。
     * 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().isSingleton(name);
    }
 
    /**
     * @param name
     * @return Class 注册对象的类型
     */
    public static Class getType(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().getType(name);
    }
 
    /**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException {
        return getBeanFactory().getAliases(name);
    }
 
    /**
     * 获取aop代理对象
     *
     * @param invoker
     * @return
     */
    @SuppressWarnings("unchecked")
    public static  T getAopProxy(T invoker) {
        return (T) AopContext.currentProxy();
    }
 
}

StringUtils #字符串工具类

package com.example.demo.util;
 
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Validator;
import cn.hutool.core.util.StrUtil;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.springframework.util.AntPathMatcher;
 
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:41
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class StringUtils extends org.apache.commons.lang3.StringUtils {
 
    /**
     * 获取参数不为空值
     *
     * @param str defaultValue 要判断的value
     * @return value 返回值
     */
    public static String blankToDefault(String str, String defaultValue) {
        return StrUtil.blankToDefault(str, defaultValue);
    }
 
    /**
     * * 判断一个字符串是否为空串
     *
     * @param str String
     * @return true：为空 false：非空
     */
    public static boolean isEmpty(String str) {
        return StrUtil.isEmpty(str);
    }
 
    /**
     * * 判断一个字符串是否为非空串
     *
     * @param str String
     * @return true：非空串 false：空串
     */
    public static boolean isNotEmpty(String str) {
        return !isEmpty(str);
    }
 
    /**
     * 去空格
     */
    public static String trim(String str) {
        return StrUtil.trim(str);
    }
 
    /**
     * 截取字符串
     *
     * @param str   字符串
     * @param start 开始
     * @return 结果
     */
    public static String substring(final String str, int start) {
        return substring(str, start, str.length());
    }
 
    /**
     * 截取字符串
     *
     * @param str   字符串
     * @param start 开始
     * @param end   结束
     * @return 结果
     */
    public static String substring(final String str, int start, int end) {
        return StrUtil.sub(str, start, end);
    }
 
    /**
     * 格式化文本, {} 表示占位符
     * 此方法只是简单将占位符 {} 按照顺序替换为参数
     * 如果想输出 {} 使用 \\转义 { 即可，如果想输出 {} 之前的 \ 使用双转义符 \\\\ 即可
     * 例：
     * 通常使用：format("this is {} for {}", "a", "b") -> this is a for b
     * 转义{}： format("this is \\{} for {}", "a", "b") -> this is {} for a
     * 转义\： format("this is \\\\{} for {}", "a", "b") -> this is \a for b
     *
     * @param template 文本模板，被替换的部分用 {} 表示
     * @param params   参数值
     * @return 格式化后的文本
     */
    public static String format(String template, Object... params) {
        return StrUtil.format(template, params);
    }
 
    /**
     * 是否为http(s)://开头
     *
     * @param link 链接
     * @return 结果
     */
    public static boolean ishttp(String link) {
        return Validator.isUrl(link);
    }
 
    /**
     * 字符串转set
     *
     * @param str 字符串
     * @param sep 分隔符
     * @return set集合
     */
    public static Set str2Set(String str, String sep) {
        return new HashSet<>(str2List(str, sep, true, false));
    }
 
    /**
     * 字符串转list
     *
     * @param str         字符串
     * @param sep         分隔符
     * @param filterBlank 过滤纯空白
     * @param trim        去掉首尾空白
     * @return list集合
     */
    public static List str2List(String str, String sep, boolean filterBlank, boolean trim) {
        List list = new ArrayList<>();
        if (isEmpty(str)) {
            return list;
        }
 
        // 过滤空白字符串
        if (filterBlank && isBlank(str)) {
            return list;
        }
        String[] split = str.split(sep);
        for (String string : split) {
            if (filterBlank && isBlank(string)) {
                continue;
            }
            if (trim) {
                string = trim(string);
            }
            list.add(string);
        }
 
        return list;
    }
 
    /**
     * 查找指定字符串是否包含指定字符串列表中的任意一个字符串同时串忽略大小写
     *
     * @param cs                  指定字符串
     * @param searchCharSequences 需要检查的字符串数组
     * @return 是否包含任意一个字符串
     */
    public static boolean containsAnyIgnoreCase(CharSequence cs, CharSequence... searchCharSequences) {
        return StrUtil.containsAnyIgnoreCase(cs, searchCharSequences);
    }
 
    /**
     * 驼峰转下划线命名
     */
    public static String toUnderScoreCase(String str) {
        return StrUtil.toUnderlineCase(str);
    }
 
    /**
     * 是否包含字符串
     *
     * @param str  验证字符串
     * @param strs 字符串组
     * @return 包含返回true
     */
    public static boolean inStringIgnoreCase(String str, String... strs) {
        return StrUtil.equalsAnyIgnoreCase(str, strs);
    }
 
    /**
     * 将下划线大写方式命名的字符串转换为驼峰式。如果转换前的下划线大写方式命名的字符串为空，则返回空字符串。 例如：HELLO_WORLD->HelloWorld
     *
     * @param name 转换前的下划线大写方式命名的字符串
     * @return 转换后的驼峰式命名的字符串
     */
    public static String convertToCamelCase(String name) {
        return StrUtil.upperFirst(StrUtil.toCamelCase(name));
    }
 
    /**
     * 驼峰式命名法 例如：user_name->userName
     */
    public static String toCamelCase(String s) {
        return StrUtil.toCamelCase(s);
    }
 
    /**
     * 查找指定字符串是否匹配指定字符串列表中的任意一个字符串
     *
     * @param str  指定字符串
     * @param strs 需要检查的字符串数组
     * @return 是否匹配
     */
    public static boolean matches(String str, List strs) {
        if (isEmpty(str) || CollUtil.isEmpty(strs)) {
            return false;
        }
        for (String pattern : strs) {
            if (isMatch(pattern, str)) {
                return true;
            }
        }
        return false;
    }
 
    /**
     * 判断url是否与规则配置:
     * ? 表示单个字符;
     * * 表示一层路径内的任意字符串，不可跨层级;
     * ** 表示任意层路径;
     *
     * @param pattern 匹配规则
     * @param url     需要匹配的url
     * @return
     */
    public static boolean isMatch(String pattern, String url) {
        AntPathMatcher matcher = new AntPathMatcher();
        return matcher.match(pattern, url);
    }
 
    /**
     * 数字左边补齐0，使之达到指定长度。注意，如果数字转换为字符串后，长度大于size，则只保留 最后size个字符。
     *
     * @param num 数字对象
     * @param size 字符串指定长度
     * @return 返回数字的字符串格式，该字符串为指定长度。
     */
    public static final String padl(final Number num, final int size) {
        return padl(num.toString(), size, '0');
    }
 
    /**
     * 字符串左补齐。如果原始字符串s长度大于size，则只保留最后size个字符。
     *
     * @param s 原始字符串
     * @param size 字符串指定长度
     * @param c 用于补齐的字符
     * @return 返回指定长度的字符串，由原字符串左补齐或截取得到。
     */
    public static final String padl(final String s, final int size, final char c) {
        final StringBuilder sb = new StringBuilder(size);
        if (s != null) {
            final int len = s.length();
            if (s.length() <= size) {
                for (int i = size - len; i > 0; i--) {
                    sb.append(c);
                }
                sb.append(s);
            } else {
                return s.substring(len - size, len);
            }
        } else {
            for (int i = size; i > 0; i--) {
                sb.append(c);
            }
        }
        return sb.toString();
    }
 
}

RedisUtils #redis工具类

/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:44
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
@SuppressWarnings(value = {"unchecked", "rawtypes"})
public class RedisUtils {
 
    private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class);
 
    /**
     * 发布通道消息
     *
     * @param channelKey 通道key
     * @param msg        发送数据
     * @param consumer   自定义处理
     */
    public static  void publish(String channelKey, T msg, Consumer consumer) {
        RTopic topic = CLIENT.getTopic(channelKey);
        topic.publish(msg);
        consumer.accept(msg);
    }
 
 
    /**
     * 订阅通道接收消息
     *
     * @param channelKey 通道key
     * @param clazz      消息类型
     * @param consumer   自定义处理
     */
    public static  void subscribe(String channelKey, Class clazz, Consumer consumer) {
        RTopic topic = CLIENT.getTopic(channelKey);
        topic.addListener(clazz, (channel, msg) -> consumer.accept(msg));
    }
}

TestRedisson #测试控制器

package com.example.demo.redisson;
 
import com.example.demo.base.R;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
/**
 * @author yueF_L
 * @version 1.0
 * @date 2022-09-02 15:50
 */
@RestController
@RequestMapping("/redisson")
public class TestRedisson {
 
    /**
     * 发布消息
     *
     * @param key   通道Key
     * @param value 发送内容
     */
    @GetMapping("/pub")
    public R pub(String key, String value) {
        RedisUtils.publish(key, value, consumer -> {
            System.out.println("发布通道 => " + key + ", 发送值 => " + value);
        });
        return R.ok("操作成功");
    }
 
    /**
     * 订阅消息
     *
     * @param key 通道Key
     */
    @GetMapping("/sub")
    public R sub(String key) {
        RedisUtils.subscribe(key, String.class, msg -> {
            System.out.println("订阅通道 => " + key + ", 接收值 => " + msg);
        });
        return R.ok("操作成功");
    }
}

postman #断点测试

订阅消息：

 发布消息

 测试成功：