Redis 消息发布订阅 与 Spring boot 使用快速入门
目录Redis 发布订阅 命令Redis 发布订阅 命令命令描述Redis Unsubscribe 命令指退订给定的频道。UNSUBSCRIBE channel [channel ...]Redis Subscribe 命令订阅给定的一个或多个频道的信息。SUBSCRIBE channel [channel ...]Redis Pubsub 命令查看订阅与发布系统状态。PUBSUB <sub
目录
Redis 消息发布订阅
Reis Pub/Sub:REDIS pubsub -- Redis中国用户组(CRUG)
Redis 发布订阅 命令:redis命令手册
1、Redis 中"pub/sub"的消息,为"即发即失",server 不会保存消息,如果 publish 的消息没有任何 client 处于 "subscribe" 状态,消息将会被丢弃;如果 client 在 subcribe 时,链接断开后重连,那在么此期间的消息也将丢失。
2、Redis server 将会"尽力"将消息发送给处于 subscribe 状态的 client,但是仍不会保证每条消息都能被正确接收。
命令 | 描述 |
---|---|
Redis Unsubscribe 命令 | 指退订给定的频道。UNSUBSCRIBE channel [channel ...] |
Redis Subscribe 命令 | 订阅给定的一个或多个频道的信息。SUBSCRIBE channel [channel ...] |
Redis Pubsub 命令 | 查看订阅与发布系统状态。PUBSUB <subcommand> [argument [argument ...]] |
Redis Punsubscribe 命令 | 退订所有给定模式的频道。PUNSUBSCRIBE [pattern [pattern ...]] |
Redis Publish 命令 | 将信息发送到指定的频道。PUBLISH channel message |
Redis Psubscribe 命令 | 订阅一个或多个符合给定模式的频道。PSUBSCRIBE pattern [pattern ...] 每个模式以 * 作为匹配符,比如 it* 匹配所有以 it 开头的频道( it.news 、 it.blog 、 it.tweets 等等) |
Spring boot 实现消息发布订阅
1、引入 Redis 依赖:https://gitee.com/wangmx1993/thymeleafapp/blob/master/pom.xml
<!--Spring Boot redis 启动器-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
2、Redis 数据库配置:https://gitee.com/wangmx1993/thymeleafapp/blob/master/src/main/resources/application.yml
spring:
data:
redis:
database: 0
host: localhost
port: 6379
password:
环境:Java Jdk 1.8 + Spring boot 2.1.3 + Redis 5.0 。
发布消息
/**
* redis 将信息发送到指定的频道
* http://localhost:8080/redis/pub?topic=memoryCache&context=flushAll
* http://localhost:8080/redis/pub?topic=basic-service&context=flushDb
*
* @param topic :消息所属的主题/频道
* @param context :消息内容
* @return
*/
@GetMapping("redis/pub")
@ResponseBody
public Map<String, Object> retryable(@RequestParam String topic, @RequestParam Object context) {
/**
* void convertAndSend(String destination, Object message): 将给定消息发布到给定通道
* 1、 destination:要发布到的通道不能为 null
*/
redisTemplate.convertAndSend(topic, context);
Map<String, Object> dataMap = new HashMap<>(4);
dataMap.put(topic, context);
return dataMap;
}
消息监听
1、 Redis 消息订阅-消息监听器,当收到阅订的消息时,会将消息交给这个类处理。
/**
* Redis 消息订阅-消息监听器,当收到阅订的消息时,会将消息交给这个类处理
* <p>
* 1、可以直接实现 MessageListener 接口,也可以继承它的实现类 MessageListenerAdapter.
* 2、自动多线程处理,打印日志即可看出,即使手动延迟,也不会影响后面消息的接收。
*
* @author wangMaoXiong
* @version 1.0
* @date 2022/5/21 16:00
*/
@Component
public class RedisSubListener implements MessageListener {
// 直接从容器中获取
@Resource
private RedisTemplate redisTemplate;
/**
* 监听到的消息必须进行与发送时相同的方式进行反序列
* 1、订阅端与发布端 Redis 序列化的方式必须相同,否则会乱码。
*
* @param message :消息实体
* @param pattern :匹配模式
*/
@Override
public void onMessage(Message message, byte[] pattern) {
// 消息订阅的匹配规则,如 new PatternTopic("basic-*") 中的 basic-*
String msgPattern = new String(pattern);
// 消息所属的通道,可以根据不同的通道做不同的业务逻辑
String channel = (String) redisTemplate.getStringSerializer().deserialize(message.getChannel());
// 接收的消息内容,可以根据自己需要强转为自己需要的对象,但最好先使用 instanceof 判断一下
Object body = redisTemplate.getValueSerializer().deserialize(message.getBody());
// 收到 Redis 订阅消息: channel=basic-service body=flushDb pattern=basic-*
// 收到 Redis 订阅消息: channel=memoryCache body=flushAll pattern=memoryCache
log.info("收到 Redis 订阅消息: channel={} body={} pattern={} ", channel, body, msgPattern);
// 手动延迟,模拟数据处理
ThreadUtil.safeSleep(800 + new Random().nextInt(3000));
log.info("------------数据处理完成.......");
}
}
主题订阅
1、自定义 RedisTemplate 序列化方式(发布者和订阅者必须相同)。
2、配置主题订阅 - Redis 消息监听器绑定监听指定通道。
/**
* 自定义 RedisTemplate 序列化方式
* 配置主题订阅 - Redis 消息监听器绑定监听指定通道
*
* @author wangMaoXiong
* @version 1.0
* @date 2022/5/21 16:13
*/
@Configuration
public class RedisConfig {
// 自定义的消息订阅监听器,当收到阅订的消息时,会将消息交给这个类处理
@Resource
private RedisSubListener redisSubListener;
// 自定义 RedisTemplate 序列化方式
@Bean
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();//创建 RedisTemplate,key 和 value 都采用了 Object 类型
redisTemplate.setConnectionFactory(redisConnectionFactory);//绑定 RedisConnectionFactory
//创建 Jackson2JsonRedisSerializer 序列方式,对象类型使用 Object 类型,
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(objectMapper);//设置一下 jackJson 的 ObjectMapper 对象参数
// 设置 RedisTemplate 序列化规则。因为 key 通常是普通的字符串,所以使用 StringRedisSerializer 即可。
// 而 value 是对象时,才需要使用序列化与反序列化
redisTemplate.setKeySerializer(new StringRedisSerializer());// key 序列化规则
redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);// value 序列化规则
redisTemplate.setHashKeySerializer(new StringRedisSerializer());// hash key 序列化规则
redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// hash value 序列化规则
redisTemplate.afterPropertiesSet();//属性设置后操作
return redisTemplate;//返回设置好的 RedisTemplate
}
/**
* 配置主题订阅
* RedisMessageListenerContainer - Redis 消息监听器绑定监听指定通道
* 1、可以添加多个监听器,监听多个通道,只需要将消息监听器与订阅的通道/主题绑定即可。
* 2、订阅的通道可以配置在全局配置文件中,也可以配置在数据库中,
* <p>
* addMessageListener(MessageListener listener, Collection<? extends Topic> topics):将消息监听器与多个订阅的通道/主题绑定
* addMessageListener(MessageListener listener, Topic topic):将消息监听器与订阅的通道/主题绑定
*
* @param connectionFactory
* @return
*/
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
// 设置连接工厂,RedisConnectionFactory 可以直接从容器中取,也可以从 RedisTemplate 中取
container.setConnectionFactory(connectionFactory);
// 订阅名称叫 memoryCache 的通道, 类似 Redis 中的 subscribe 命令
container.addMessageListener(redisSubListener, new ChannelTopic("memoryCache"));
// 订阅名称以 'basic-' 开头的全部通道, 类似 Redis 的 pSubscribe 命令
container.addMessageListener(redisSubListener, new PatternTopic("basic-*"));
return container;
}
}
Spring boot 监听 Key 过期事件
1、Redis 数据库可以通过命令设置 Key 的有效时间,当一个 Key 过期后会自动从数据库中删除,释放空间。得益于于这个特性,可以很轻松地实现诸多类似于 “Session” 管理、数据缓存等功能。它们都有一个共同点就是,数据不会永久保存!
2、在有些场景中,可能希望在某些 Key 过期的时候获取到通知,进行一些业务处理。或者是干脆用于 “定时通知/任务” 功能,例如:下单 30 分钟后未支付,则取消订单。那么可以在用户下单的时候使用订单号作为 key 设置到 Redis 数据库中,并且设置过期时间为 30 分钟。当超时后,可以在 “key 过期通知” 中获取到 key 也就是订单号,判断用户是否已经支付从而是否取消订单。
3、Redis 的 Key 过期通知功能本质上是通过 发布/订阅 功能实现的,所以它「不能保证通知消息的交付」,当 Key 过期时如果服务器停机、重启后则该通知消息会永久丢失。
消息监听
1、Spring Data Redis 专门提供了一个密钥过期事件消息侦听器:KeyExpirationEventMessageListener,自定义监听器类继承它,然后覆写 doHandleMessage(Message message) 方法即可。
2、doHandleMessage 方法用于处理 Redis Key 过期通知事件,其中 Message 参数表示通知消息,只有 2 属性,分别表示消息正文(在这里就是过期的 Key 名称)以及来自于哪个 channel。
3、在 Redis Key 过期事件中,「只能获取到已过期的 Key 的名称,不能获取到值。」
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
/**
* Redis 缓存 Key 过期监听器
* Spring Data Redis 专门提供了一个密钥过期事件消息侦听器:KeyExpirationEventMessageListener,
* 自定义监听器类继承它,然后覆写 doHandleMessage(Message message) 方法即可。
*
* @author wangMaoXiong
* @version 1.0
* @date 2023/11/17 17:37
*/
@Component
public class KeyExpireListener extends KeyExpirationEventMessageListener {
private static final Logger logger = LoggerFactory.getLogger(KeyExpireListener.class);
/**
* 通过构造函数注入 RedisMessageListenerContainer 给 KeyExpirationEventMessageListener
*
* @param listenerContainer : Redis消息侦听器容器
*/
public KeyExpireListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
/**
* doHandleMessage 方法用于处理 Redis Key 过期通知事件,
* 在 Redis Key 过期事件中,「只能获取到已过期的 Key 的名称,不能获取到值。」
*
* @param message:通知消息,只有 2 属性,分别表示消息正文(在这里就是过期的 Key 名称)以及来自于哪个 channel。
*/
@Override
public void doHandleMessage(Message message) {
// 过期的 key
String key = new String(message.getBody());
// 消息通道
String channel = new String(message.getChannel());
logger.info("过期key={} 消息通道(channel)={}", key, channel);
}
}
src/main/java/com/wmx/thymeleafapp/listeners/KeyExpireListener.java · 汪少棠/thymeleafapp - Gitee.com。
主题订阅
1、与上面稍微有点不同,因为 key 过期事件属于 Redis 内部消息,内部频道/通道,所以只需要往容器中注入 RedisMessageListenerContainer 就行,不需要 addMessageListener 手动设置监听器 监听指定的通道/频道(topic 表达式)。
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.RedisMessageListenerContainer;
@Configuration
public class RedisConfig {
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory factory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
// container.setTaskExecutor(null); // 设置用于执行监听器方法的 Executor
// container.setErrorHandler(null); // 设置监听器方法执行过程中出现异常的处理器
// container.addMessageListener(null, null); // 手动设置监听器 & 监听的 topic 表达式
return container;
}
}
src/main/java/com/wmx/thymeleafapp/config/RedisConfig.java · 汪少棠/thymeleafapp - Gitee.com。
开放原子开发者工作坊旨在鼓励更多人参与开源活动,与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动,如meetup、训练营等,主打技术交流,干货满满,真诚地邀请各位开发者共同参与!
更多推荐
所有评论(0)