Spring Data Redis Pub/Sub：发布订阅模式 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

好的，各位观众老爷们，大家好！我是你们的老朋友，人见人爱，花见花开，车见车爆胎的编程界小喇叭——Bug猎手！今天，咱们要聊聊一个既实用又有趣的话题：Spring Data Redis Pub/Sub，也就是Redis的发布订阅模式。

想象一下，你是一名广播电台的DJ，每天的工作就是对着麦克风叭叭叭，把各种劲爆消息、动听音乐、天气预报一股脑儿地甩出去。听众们呢，就像一个个乖巧的小喇叭，默默地收听着你发送的内容。这就是发布订阅模式最形象的比喻！

一、啥是发布订阅？为啥要用它？

发布订阅（Publish-Subscribe，简称Pub/Sub）是一种消息传递模式，它将消息的发送者（发布者）和消息的接收者（订阅者）解耦。发布者不需要知道谁是订阅者，订阅者也不需要知道谁是发布者。他们之间的桥梁，就是消息通道，在我们的案例中，这个消息通道就是Redis。

为啥要用它呢？因为它香啊！😎

解耦！解耦！还是解耦！ 就像前女友一样，让你摆脱了依赖的苦海，拥抱自由！发布者和订阅者不再直接依赖，修改一方不会影响另一方，系统更加灵活。
异步处理！ 想象一下，你点了外卖，难道要傻乎乎地站在餐馆门口等着小哥做完再送过来吗？当然不用！你只需要下单，然后就可以愉快地刷剧，等着外卖小哥敲门。Pub/Sub也是一样，发布者发送消息后就可以去做其他事情，订阅者异步处理消息，提高了系统的响应速度和吞吐量。
广播通知！ 一条消息，千家万户受益！比如，你发布了一条“今晚打老虎”的消息，所有关注打老虎频道的订阅者都会收到通知，是不是很刺激？

二、Redis Pub/Sub：技术细节大揭秘

Redis作为高性能的内存数据库，天然支持Pub/Sub模式。它提供了三个主要的命令：

PUBLISH channel message: 发布消息到指定的频道。
SUBSCRIBE channel [channel …]: 订阅一个或多个频道。
PSUBSCRIBE pattern [pattern …]: 订阅一个或多个符合给定模式的频道。

简单来说，PUBLISH是DJ对着麦克风说话，SUBSCRIBE和PSUBSCRIBE是听众打开收音机，调到对应的频道或者频率。

让我们用一个表格来总结一下：

命令	作用	备注
PUBLISH	将消息发布到指定频道	发布者使用，广播消息。
SUBSCRIBE	订阅一个或多个频道	订阅者使用，精确匹配频道名。
PSUBSCRIBE	订阅一个或多个符合给定模式的频道	订阅者使用，可以使用通配符（例如`news.*`），订阅所有以`news.`开头的频道。
PUNSUBSCRIBE	取消订阅符合给定模式的频道	订阅者使用, 取消PSUBSCRIBE订阅
UNSUBSCRIBE	取消订阅一个或多个频道	订阅者使用，取消SUBSCRIBE订阅

三、Spring Data Redis：让Pub/Sub更上一层楼

Spring Data Redis是对Redis客户端的封装，它提供了更高级别的抽象，让我们可以更方便地使用Redis的各种功能，包括Pub/Sub。

1. 配置Spring Data Redis

首先，你需要在你的Spring项目中引入Spring Data Redis的依赖。如果你使用Maven，可以在pom.xml文件中添加以下内容：

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

然后，你需要配置Redis连接信息。可以在application.properties或application.yml文件中添加以下配置：

spring.redis.host=localhost
spring.redis.port=6379
# spring.redis.password=your_password  # 如果Redis有密码，请配置

2. 编写发布者（Publisher）

发布者负责将消息发布到Redis频道。我们需要创建一个RedisTemplate实例，并使用它的convertAndSend()方法来发布消息。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class RedisMessagePublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    private String topic = "news.sports"; // 频道名称

    public void publish(String message) {
        redisTemplate.convertAndSend(topic, message);
        System.out.println("Published message: " + message + " to topic: " + topic);
    }
}

3. 编写订阅者（Subscriber）

订阅者负责监听Redis频道，并处理接收到的消息。我们需要创建一个消息监听器（MessageListener），并将其注册到Redis消息监听容器（RedisMessageListenerContainer）。

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

@Component
public class RedisMessageSubscriber implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        String topic = new String(pattern);
        System.out.println("Received message: " + body + " from topic: " + topic);
    }
}

4. 配置Redis消息监听容器（RedisMessageListenerContainer）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;

@Configuration
public class RedisConfig {

    @Autowired
    private RedisMessageSubscriber redisMessageSubscriber;

    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 订阅一个频道
        container.addMessageListener(redisMessageSubscriber, new ChannelTopic("news.sports"));

        // 可以订阅多个频道
        // container.addMessageListener(redisMessageSubscriber, Arrays.asList(new ChannelTopic("news.sports"), new ChannelTopic("news.tech")));

        // 使用模式匹配订阅频道，例如订阅所有以"news."开头的频道
        // container.addMessageListener(redisMessageSubscriber, new PatternTopic("news.*"));

        return container;
    }
}

代码解释：

RedisMessageListenerContainer：Redis消息监听容器，负责监听Redis频道上的消息。
RedisConnectionFactory：Redis连接工厂，用于创建Redis连接。
RedisMessageSubscriber：消息监听器，用于处理接收到的消息。
ChannelTopic：频道主题，用于指定要订阅的频道。可以订阅一个或者多个channel。
PatternTopic: 模式匹配topic，比如 "news.*" 可以匹配 news.sports, news.tech 等频道。

5. 测试

现在，你可以启动你的Spring应用程序，并使用RedisMessagePublisher发布消息。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class AppRunner implements CommandLineRunner {

    @Autowired
    private RedisMessagePublisher redisMessagePublisher;

    @Override
    public void run(String... args) throws Exception {
        redisMessagePublisher.publish("Cristiano Ronaldo scores a hat-trick!");
        redisMessagePublisher.publish("LeBron James wins MVP!");
    }
}

你会看到RedisMessageSubscriber接收到了发布的消息，并在控制台打印出来。

四、进阶技巧：模式匹配、序列化、异常处理

1. 模式匹配（Pattern Matching）

如果你想订阅多个频道，可以使用PSUBSCRIBE命令或者Spring Data Redis提供的PatternTopic。例如，你可以订阅所有以news.开头的频道：

container.addMessageListener(redisMessageSubscriber, new PatternTopic("news.*"));

这样，所有发布到news.sports、news.tech、news.finance等频道的消息，都会被RedisMessageSubscriber接收到。

2. 序列化（Serialization）

默认情况下，RedisTemplate使用JdkSerializationRedisSerializer进行序列化，这可能会导致一些问题，比如性能问题和类版本兼容性问题。因此，建议使用更高效的序列化方式，比如Jackson2JsonRedisSerializer或者StringRedisSerializer。

例如，使用Jackson2JsonRedisSerializer：

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);

        // 使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(com.fasterxml.jackson.annotation.PropertyAccessor.ALL, com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        // 设置键（key）的序列化采用StringRedisSerializer。
        template.setKeySerializer(new StringRedisSerializer());
        template.setHashKeySerializer(new StringRedisSerializer());
        // 设置值（value）的序列化采用Jackson2JsonRedisSerializer。
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);

        template.afterPropertiesSet();
        return template;
    }
}

3. 异常处理（Exception Handling）

在实际应用中，我们需要考虑各种异常情况，比如Redis连接失败、消息处理失败等。可以使用try-catch块来捕获异常，并进行相应的处理。

@Component
public class RedisMessageSubscriber implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            String body = new String(message.getBody());
            String topic = new String(pattern);
            System.out.println("Received message: " + body + " from topic: " + topic);
            // 处理消息的逻辑
        } catch (Exception e) {
            // 记录日志
            System.err.println("Error processing message: " + e.getMessage());
            // 可以选择重新发布消息，或者将消息放入死信队列
        }
    }
}

五、真实案例：用Pub/Sub构建实时聊天室

想象一下，你想做一个实时聊天室，让用户可以随时随地聊天。使用Redis Pub/Sub，可以很容易地实现这个功能。

客户端（浏览器）：客户端通过WebSocket连接到服务器，并订阅一个聊天频道（例如chat.room1）。
服务器：当客户端发送消息时，服务器将消息发布到对应的聊天频道。
Redis：Redis作为消息中间件，负责将消息传递给所有订阅了该频道的客户端。

这样，所有加入chat.room1的客户端，都可以实时地收到其他客户端发送的消息。

六、总结：Pub/Sub，你的得力助手！

Spring Data Redis Pub/Sub是一种强大的消息传递模式，它可以帮助你构建解耦、异步、可扩展的应用程序。无论是广播通知、实时聊天、还是事件驱动架构，Pub/Sub都能发挥重要作用。

就像一位默默奉献的老黄牛，Pub/Sub在幕后辛勤工作，为你提供可靠的消息传递服务。所以，下次当你需要构建一个高性能、高可用的系统时，不妨考虑一下Pub/Sub，它一定会给你带来惊喜！

希望今天的讲解对你有所帮助！记住，编程的乐趣在于不断学习、不断探索。让我们一起努力，成为更优秀的程序员！💪

最后，送大家一个彩蛋：

程序员的浪漫是什么？

是把BUG变成FEATURE！ 😜

下次再见！拜拜！ 👋

发表回复 取消回复

发表回复取消回复