1、准备工作

  <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

2、Redis 配置

package com.cetc.protocolconversion.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.concurrent.Executors;

/**
 * redis 配置
 *
 * @author yjm
 * @date 2025年11月12日 15:56
 */
@Configuration
public class RedisConfig {

    /**
     * redis 订阅容器-灵活配置消息处理
     *
     * @param connectionFactory redis连接操作
     * @return 容器
     */
    @Bean
    public RedisMessageListenerContainer messageListenerContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }


    /**
     * 指定redis key,value,hash的序列化器
     *
     * @param factory redisTemplate工厂类
     * @return redisTemplate
     */
    @Bean
    RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        template.setConnectionFactory(factory);
        template.setKeySerializer(stringRedisSerializer);
        template.setValueSerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setHashValueSerializer(stringRedisSerializer);
        return template;
    }
}

3、Redis 动态订阅工具类

package com.cetc.protocolconversion.utils;

import com.cetc.protocolconversion.config.DynamicRedisMessageListener;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * redis 订阅发布管理
 *
 * @author yjm
 * @date 2025年12月29日 20:02
 */
@Component
public class RedisPubSubUtil {
    /*线程安全的订阅发布容器*/
    private final Map<String, DynamicRedisMessageListener> pubSubMap = new ConcurrentHashMap<>();

    @Resource
    RedisMessageListenerContainer listenerContainer;



    /**
     * 指定订阅处理器
     *
     * @param channel 主题
     * @param handler 指定消息接收处理
     */
    public void subscribe(String channel, DynamicRedisMessageListener.MessageHandler handler) {
        DynamicRedisMessageListener messageListener = new DynamicRedisMessageListener();
        messageListener.setMessageHandler(handler);
       
        listenerContainer.addMessageListener(messageListener, new ChannelTopic(channel));
        pubSubMap.put(channel, messageListener);
    }

    /**
     * 取消订阅发布
     *
     * @param channel 主题
     */
    public void unsubscribe(String channel) {
        DynamicRedisMessageListener listener = pubSubMap.get(channel);
        if (null == listener) {
            return;
        }
        Topic topic = new ChannelTopic(channel);
        listenerContainer.removeMessageListener(listener, topic);
        pubSubMap.remove(channel);
    }

    /**
     * 列出所有channel
     *
     * @return channel set
     */
    public Set<String> listChannels() {
        return pubSubMap.keySet();
    }
}

4、Redis 消息监听器

package com.cetc.protocolconversion.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;


/**
 * 动态redis消息监听
 *
 * @author yjm
 * @date 2025年12月29日 17:53
 */
@Slf4j
public class DynamicRedisMessageListener implements MessageListener {
    private volatile MessageHandler messageHandler;

    public interface MessageHandler {
        void handle(String channel, String message);
    }


    @Override
    public void onMessage(Message message, byte[] bytes) {
        String channel = new String(message.getChannel());
        String msg = new String(message.getBody());
        if (messageHandler != null) {
            messageHandler.handle(channel, msg);
        } else {
            log.info("默认处理：频道[{}]:{}", channel, msg);
        }
    }

    //设置动态处理器
    public void setMessageHandler(MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
    }
}

5、测Redis监听器类测试实现类

package com.cetc.protocolconversion.config.redismsglistencer;

import com.cetc.protocolconversion.config.DynamicRedisMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author yjm
 * @date 2026年01月07日 16:12
 */
@Component
@Slf4j
public class TestHandlerImpl implements DynamicRedisMessageListener.MessageHandler {
    public static final String TEST_CHANNEL = "TEST";

    @Override
    public void handle(String channel, String message) {
        log.info(channel + "-:-" + message);
    }
}

6、测试指定监听器订阅

@Resource
    RedisPubSubUtil redisPubSubUtil;
     @Resource
    TestHandlerImpl testHandler;
    
redisPubSubUtil.subscribe("TEST_CHANNEL", testHandler)

7 、测试发布

@Resource
    RedisPubSubUtil redisPubSubUtil;
    
redisUtil.publish("TEST_CHANNEL", "这就是个测试");

8、结果

按照TestHandlerImpl 类中所写，那就是测试发布类中发布一次，那就会打印一次:

TEST_CHANNEL:这就是个测试