Redis实现发布/订阅功能（实战篇）

用户11877422

发布于 2025-11-03 17:32:54

1240

前言

博主在学习 Redis 实现发布订阅功能的时候，踩了太多的坑。

不是讲解不详细，看的一知半解；就是代码有问题，实际压根跑不起来！

于是博主萌生了自己写一个最新版且全程无错的博客供各位参考。希望各位不要把我才过的坑再踩一遍。（实战篇的所有代码均由本人测试，全程无Bug。）

废话不多说，让我们进入实战篇的学习！

在开始实战篇的之前，我们先一起回顾下原理篇的内容。

Redis 发布/订阅的优点

高性能：Redis 作为内存存储，具备极高的读写性能，能够快速处理发布和订阅消息。
简单易用：Redis 的发布/订阅接口简单，易于集成和使用。
实时性强：发布的消息会立即传递给所有订阅者，具备高实时性。

Redis 发布/订阅的缺点

消息丢失：由于 Redis 是内存存储，如果 Redis 实例宕机，未处理的消息可能会丢失。
无法持久化：Redis 的发布/订阅模式不支持消息持久化，无法存储和检索历史消息。
订阅者不可控：发布者无法控制订阅者的数量和状态，无法保证所有订阅者都能接收到消息。
无确认机制：发布者无法确认消息是否被订阅者接收和处理。

正如上述中Redis的缺点，Redis的发布订阅功能并不可靠，如果我们需要保证消息的可靠性、包括确认、重试等要求，我们还是要选择使用MQ实现发布订阅。

Redis的发布/订阅应用场景：

对于消息处理可靠性要求不强
消息无需持久化
消费能力无需通过增加消费方进行增强
架构简单中小型系统不希望应用过多中间件

Redis发布订阅命令

命令	描述
Redis Unsubscribe 命令	指退订给定的频道
Redis Subscribe 命令	订阅给定的一个或多个频道的信息
Redis Pubsub 命令	查看订阅与发布系统状态
Redis Punsubscribe 命令	退订所有给定模式的频道
Redis Publish 命令	将信息发送到指定的频道
Redis Psubscribe 命令	订阅一个或多个符合给定模式的频道

正式开始

一、添加依赖

首先，确保你已经安装并配置好 Redis 服务器，并创建了 Spring Boot 项目，在pom.xml中引入依赖。

<!-- 所需依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

二、配置文件中配置Redis

spring:
  # 项目名称
  application:
    name: test-redis-boot

  # Redis 配置
  data:
    redis:
      host: 填写自己的主机IP
      port: 8000
      password: 有则填，没有去掉这个属性
      database: 1
      # 连接超时时间
      timeout: 10s
      lettuce:
        pool:
          # 连接池中的最小空闲连接
          min-idle: 5
          # 连接池中的最大空闲连接
          max-idle: 8
          # 连接池的最大数据库连接数
          max-active: 20
          # #连接池最大阻塞等待时间（使用负值表示没有限制）
          max-wait: -1ms

三、创建Redis配置类

创建一个配置类，用于配置 Redis连接工厂和消息监听器监听通道信息。

注：此配置类无需死记硬背。只需大致了解每个方法的作用即可。

/**
 * @Description Redis 配置类，用于配置 Redis 连接工厂和消息监听器监听通道信息
 * @Author gongming.Zhang
 * @Date 2024/9/11 18:27
 * @Version 1.0
 */
@Configuration
@Slf4j
public class RedisConfig {

    /**
     * 自定义 RedisTemplate 序列化方式
     *
     * @param redisConnectionFactory Redis 连接的线程安全工厂
     * @return 模板类
     */
    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        // 绑定 RedisConnectionFactory
        redisTemplate.setConnectionFactory(redisConnectionFactory);

        // 创建 Jackson2JsonRedisSerializer 序列方式，对象类型使用 Object 类型。
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(new LaissezFaireSubTypeValidator(), ObjectMapper.DefaultTyping.NON_FINAL);
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(objectMapper, Object.class);

        // 设置 RedisTemplate 序列化规则,因为 key 通常是普通的字符串，所以使用StringRedisSerializer即可,而 value 是对象时，才需要使用序列化与反序列化。
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        // hash key 序列化规则
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        // 属性设置后操作
        redisTemplate.afterPropertiesSet();
        log.info("RedisTemplate 自定义序列化配置完毕...");
        return redisTemplate;
    }

    /**
     * 配置主题订阅
     * 
     * 可以添加多个监听器，监听多个通道，只需要将消息监听器与订阅的通道/主题绑定即可。
     * addMessageListener(MessageListener listener, Collection<? extends Topic> topics)：将消息监听器与多个订阅的通道/主题绑定
     * addMessageListener(MessageListener listener, Topic topic)：将消息监听器与订阅的通道/主题绑定
     *
     * @param connectionFactory Redis 连接的线程安全工厂
     * @return 容器对象
     */
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                                   MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        // 设置连接工厂，RedisConnectionFactory 可以直接从容器中取，也可以从 RedisTemplate 中取
        container.setConnectionFactory(connectionFactory);
        // 订阅名称叫 cache 的通道, 类似 Redis 中的 subscribe 命令
        container.addMessageListener(listenerAdapter, new ChannelTopic("cache"));
        // 订阅名称以 'test-' 开头的全部通道, 类似 Redis 的 pSubscribe 命令
        container.addMessageListener(listenerAdapter, new PatternTopic("test-*"));
        log.info("消息监听器和通道绑定完毕...");
        return container;
    }

    /**
     * 配置消息监听适配器
     *
     * @param redisReceiveListener
     * @return
     */
    @Bean
    public MessageListenerAdapter listenerAdapter(RedisReceiveListener redisReceiveListener) {
        return new MessageListenerAdapter(redisReceiveListener);
    }
}

四、创建消息发送服务端

创建一个消息发布类，用于向客户端发布消息。

/**
 * @Description 消息发布服务端
 * @Author gongming.Zhang
 * @Date 2024/9/12 9:42
 * @Version 1.0
 */
@Component
@Slf4j
public class MessagePublisher {
    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    /**
     * 服务端发布消息
     *
     * @param channel 通道名
     * @param message 待发送的消息
     */
    public void sendMessage(String channel, String message) {
        redisTemplate.convertAndSend(channel, message);
        log.info("消息发送成功...  channel={}, message={}", channel, message);
    }

}

五、创建监听器（客户端）

用于监听服务端发送的消息，每次服务端发送新消息时，都会自动调用onMessage()方法。

/**
 * @Description Redis 消息监听器
 * @Author gongming.Zhang
 * @Date 2024/9/11 18:53
 * @Version 1.0
 */
/*
    当收到订阅的消息时，会将消息交给这个类处理。
     * 可以直接实现 MessageListener 接口，也可以继承它的实现类 MessageListenerAdapter.
     * 自动多线程处理，打印日志即可看出，即使手动延迟，也不会影响后面消息的接收。
 */
@Component
@Slf4j
public class RedisReceiveListener implements MessageListener {
    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    /**
     * 处理回调逻辑。每次新消息到达时，都会调用此方法。通过 onMessage 方法执行业务代码
     * <p>
     * 该接口不仅可以访问实际消息，还可以访问接收消息的频道（Channel），以及订阅时用于匹配频道（Channel）的模式。
     * 此信息使被调用者不仅可以通过内容区分各种消息，还可以检查其他详细信息。
     *
     * @param message 消息对象，不能为 null
     * @param pattern 与通道匹配的模式（如果指定），可以为 null
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 1.获取消息所属的通道  -->  首先获取 字符串序列化器，再从给定的二进制数据中反序列化对象。
        String channel = redisTemplate.getStringSerializer().deserialize(message.getChannel());
        // 2.获取客户端发送的消息内容  -->  后期可以根据自己项目中 消息 的类型，来确定用什么序列化器
        Object msg = redisTemplate.getValueSerializer().deserialize(message.getBody());
        log.info("收到Redis订阅消息: channel={} msg={}", channel, msg);
    }
}

六、编写Controller测试

/**
 * @Description 测试订阅发布功能
 * @Author gongming.Zhang
 * @Date 2024/9/12 10:13
 * @Version 1.0
 */
@RestController
@RequestMapping(value = "/api/v1/publish")
public class PublisherController {

    @Autowired
    private MessagePublisher publisher;

    @GetMapping("/sendMessage")
    public String sendMessage(@RequestParam(value = "message") String message, @RequestParam(value = "channel") String channel) {
        publisher.sendMessage(channel, message);
        return "Message published: " + message;
    }
}