【redis】发布订阅

Redis的发布订阅（Pub/Sub）是一种基于消息多播的通信机制，它允许消息的**发布者（Publisher）向特定频道发送消息，而订阅者（Subscriber）**通过订阅频道或模式来接收消息。

其核心特点如下：

轻量级：无需额外组件，直接通过Redis服务实现
实时性：消息即时推送，无轮询延迟
广播模式：一个消息可被多个订阅者同时接收
无状态性：不存储历史消息，订阅者只能接收订阅后的消息

发布订阅命令的使用

有关发布订阅的命令可以通过help @pubsub命令来查看。有关命令的使用可以通过help 命令来查看，例如help publish。

基础命令速查表

命令	作用	示例
`SUBSCRIBE`	订阅一个或多个频道	`SUBSCRIBE news sports`
`PSUBSCRIBE`	使用模式匹配订阅频道	`PSUBSCRIBE sensor.*`
`PUBLISH`	向指定频道发送消息	`PUBLISH news "Hello"`
`UNSUBSCRIBE`	退订指定频道	`UNSUBSCRIBE news`
`PUNSUBSCRIBE`	退订模式订阅	`PUNSUBSCRIBE sensor.*`
`PUBSUB CHANNELS`	查看活跃频道列表	`PUBSUB CHANNELS "sensor.*"`

操作示例

# 订阅者A（终端1）
127.0.0.1:6379> subscribe notifications
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "notifications"
3) (integer) 1# 订阅者B（终端2） 
127.0.0.1:6379> psubscribe system.*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "system.*"
3) (integer) 1# 发布消息（终端3）
127.0.0.1:6379> publish notifications "Service will be upgraded soon"
(integer) 1127.0.0.1:6379> publish system.alert "CPU usage exceeds 90%"
(integer) 1# 订阅者A收到：
1) "message"
2) "notifications"
3) "Service will be upgraded soon"# 订阅者B收到： 
1) "pmessage"
2) "system.*"
3) "system.alert"
4) "CPU usage exceeds 90%"

发布订阅的使用场景与优缺点

适用场景

实时通知系统：用户在线状态更新，即时聊天消息推送
事件驱动架构：缓存失效广播，分布式配置更新
轻量级监控：服务器状态报警，业务指标异常通知

优点

极低延迟（平均<1ms）
支持百万级TPS消息吞吐
模式匹配订阅实现灵活路由
零外部依赖（仅需Redis服务）

缺点

消息不可靠性：不保证送达，离线订阅者会丢失消息

无持久化机制：重启后所有订阅关系丢失

客户端阻塞:订阅操作会占用连接线程（需异步处理）

替代方案建议：需要可靠消息时，使用Redis Streams（支持消息持久化、消费者组）或RabbitMQ/Kafka

在Java中使用RedisTemplate实现

配置RedisTemplate

package com.morris.redis.demo.pubsub;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.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;/*** 对redis的键值进行序列化*/
@Configuration
public class RedisConfig {@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(factory);// 使用 String 序列化 keytemplate.setKeySerializer(new StringRedisSerializer());// 使用 JSON 序列化 value（需要额外依赖 jackson）template.setValueSerializer(new GenericJackson2JsonRedisSerializer());// 对于 Hash 结构同理template.setHashKeySerializer(new StringRedisSerializer());template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());return template;}
}

实现消息发布者

package com.morris.redis.demo.pubsub;import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;import javax.annotation.Resource;/*** 消息发布者*/
@Service
public class MessagePublisher {@Resourceprivate RedisTemplate<String, Object> redisTemplate;public void sendNotification(String channel, String message) {redisTemplate.convertAndSend(channel, message);}
}

实现消息订阅者

package com.morris.redis.demo.pubsub;import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;/*** 消息订阅者*/
@Component
public class MessageSubscriber implements MessageListener {@Resourceprivate RedisTemplate redisTemplate;@Overridepublic void onMessage(Message message, byte[] pattern) {String channel = new String(message.getChannel());String body = (String) redisTemplate.getValueSerializer().deserialize(message.getBody());System.out.printf("收到频道[%s]的消息: %s\n", channel, body);}
}

配置订阅监听

package com.morris.redis.demo.pubsub;import lombok.extern.slf4j.Slf4j;
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.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;/*** 配置redis消息订阅监听器*/
@Configuration
@Slf4j
public class RedisPubSubConfig {@Beanpublic RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory, MessageSubscriber messageSubscriber) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(factory);// 订阅具体频道container.addMessageListener(messageSubscriber, new ChannelTopic("notifications"));// 订阅模式匹配container.addMessageListener(messageSubscriber, new PatternTopic("system.*"));// 异常处理container.setErrorHandler((e) -> {log.error("[listen message] error ", e);});return container;}
}

使用示例

package com.morris.redis.demo.pubsub;import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;/*** 使用接口发布消息*/
@RestController
@RequestMapping("/pubsub")
public class PubSubDemoController {@Resourceprivate MessagePublisher publisher;// 发布告警@GetMapping("/alert")public String sendAlert(@RequestParam String message) {publisher.sendNotification("system.alert", message);return "警报已发送";}// 发布通知@GetMapping("/notify")public String sendNotify(@RequestParam String message) {publisher.sendNotification("notifications", message);return "通知已发送";}
}