有没有只做软装方案收设计费的网站,南京广告公司排名前十,wordpress首页无法看到后台登录,做网站开什么端口最近用到一个延迟消息的功能#xff0c;第一时间想到使用MQ或者MQ的插件#xff0c;因为数据量不大#xff0c;所以尝试使用Redis来实现了#xff0c;毕竟Redis也天生支持类似MQ的队列消费#xff0c;所以#xff0c;在这里总结了一下Redis实现延迟消息队列的方式。
一、…最近用到一个延迟消息的功能第一时间想到使用MQ或者MQ的插件因为数据量不大所以尝试使用Redis来实现了毕竟Redis也天生支持类似MQ的队列消费所以在这里总结了一下Redis实现延迟消息队列的方式。
一、监听key过期时间
处理流程当redis的一个key过期时redis会生成一个事件通知订阅了该事件的客户端KeyExpirationEventMessageListener然后在客户端的回调方法中处理逻辑。 1新建SpringBoot项目maven依赖及yml如下 maven依赖
dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency/dependenciesyml文件
server:port: 8000spring:redis:database: 0host: xxxxport: 6379password: xxxxxxlettuce:pool:#最大连接数max-active: 8#最大阻塞等待时间max-wait: -1#最大空闲max-idle: 8#最小空闲min-idle: 0#连接超时时间timeout: 50002修改redis.conf文件开启事件通知配置 默认的配置notify-keyspace-events “” 修改为notify-keyspace-events Ex该配置表示监听key的过期事件
3设置Redis监听配置注入Bean RedisMessageListenerContaine
Configuration
public class RedisTimeoutConfiguration {Autowiredprivate RedisConnectionFactory redisConnectionFactory;Beanpublic RedisMessageListenerContainer redisMessageListenerContainer() {RedisMessageListenerContainer redisMessageListenerContainer new RedisMessageListenerContainer();redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);return redisMessageListenerContainer;}Beanpublic KeyExpiredListener keyExpiredListener() {return new KeyExpiredListener(this.redisMessageListenerContainer());}
}4创建监听器类重写key过期回调方法onMessage
Slf4j
public class KeyExpiredListener extends KeyExpirationEventMessageListener {Autowiredpublic RedisTemplateString, String redisTemplate;public KeyExpiredListener(RedisMessageListenerContainer listenerContainer) {super(listenerContainer);}Overridepublic void onMessage(Message message, byte[] bytes) {String channel new String(message.getChannel(), StandardCharsets.UTF_8);//过期的keyString key new String(message.getBody(), StandardCharsets.UTF_8);log.info(redis key 过期bytes{},channel{},key{}, new String(bytes), channel, key);}
}5编写测试接口写入一个带过期时间的key
RestController
RequestMapping(/demo)
public class BasicController {Autowiredpublic RedisTemplateString, String redisTemplate;GetMapping(value /test)public void redisTest() {redisTemplate.opsForValue().set(test, 5s后过期, 5, TimeUnit.SECONDS);}
}执行后onMessage监听方法打印结果: redis key 过期bytes__keyevent*__:expired,channel__keyevent0__:expired,keytest该方案缺点可靠性问题Redis 是一个内存数据库尽管它提供了数据持久化选项如 RDB 和 AOF但在某些情况下如意外崩溃或重启可能会丢失一些未处理的过期事件。
二、zset score
基本思路是将消息按需发送的时间作为分数存储在有序集合zset中然后定期检查并处理到期的消息。代码例子如下 1创建 DelayedMessageService 类
Slf4j
Service
public class DelayedMessageService {private static final String DELAYED_MESSAGES_ZSET delayed:messages;Autowiredprivate RedisTemplateString, String redisTemplate;public void addMessage(String message, long delayMillis) {long score System.currentTimeMillis() delayMillis;redisTemplate.opsForZSet().add(DELAYED_MESSAGES_ZSET, message, score);}Scheduled(fixedRate 1000)public void processMessages() {long now System.currentTimeMillis();SetZSetOperations.TypedTupleString messages redisTemplate.opsForZSet().rangeByScoreWithScores(DELAYED_MESSAGES_ZSET, 0, now);if (messages ! null !messages.isEmpty()) {for (ZSetOperations.TypedTupleString message : messages) {String msg message.getValue();long score message.getScore().longValue();if (score now) {// Process the messageSystem.out.println(Processing message: msg);// Remove the message from the zsetredisTemplate.opsForZSet().remove(DELAYED_MESSAGES_ZSET, msg);}}}else{log.info(定时任务执行~);}}}2编写Controller接口测试初始化zset内容
RestController
RequestMapping(/demo)
public class BasicController {Autowiredprivate DelayedMessageService delayedMessageService;GetMapping(value /test2)public void redisZsetTest() {// Add some messages with delaysdelayedMessageService.addMessage(Message 1, 5000); // 5 seconds delaydelayedMessageService.addMessage(Message 2, 10000); // 10 seconds delaydelayedMessageService.addMessage(Message 3, 15000); // 15 seconds delay}
}说明
redisZsetTest接口通过调用DelayedMessageService的addMessage方法将消息及其到期时间添加到 Redis 的 zset 中开启一个定时任务定期检查和处理到期的消息。使用 Scheduled 注解定期执行每秒检查一次注意这里使用Scheduled不要忘了启动类上添加EnableScheduling注解否则定时任务不会生效。fixedRate 属性表示以固定的频率毫秒为单位执行方法。即方法执行完成后会立即等待指定的毫秒数然后再次执行。通过 redisTemplate.opsForZSet().rangeByScoreWithScores 方法按时间范围获取到期的消息消息处理完成后从zset 中移除处理过的消息
三、Redisson框架
利用 Redisson 提供的数据结构RDelayedQueue和RBlockingDeque可以自动处理过期的任务并将它们移动到阻塞队列中这样我们就可以从阻塞队列中获取任务并进行消费处理。例子如下 1添加依赖
!-- Redisson 依赖项 --
dependencygroupIdorg.redisson/groupIdartifactIdredisson-spring-boot-starter/artifactIdversion2.15.1/version
/dependency2创建DelayedMessageService
Slf4j
Service
public class DelayedMessageService {Autowiredprivate RedissonClient redissonClient;private RBlockingDequeString blockingDeque;private RDelayedQueueString delayedQueue;PostConstructpublic void init() {this.blockingDeque redissonClient.getBlockingDeque(delayedQueue);this.delayedQueue redissonClient.getDelayedQueue(blockingDeque);Executors.newSingleThreadExecutor().submit(this::processMessages);}public void addMessage(String message, long delayMillis) {delayedQueue.offer(message, delayMillis, TimeUnit.MILLISECONDS);}public void processMessages() {try {while (true) {String message blockingDeque.take();// Process the messagelog.info(消息被处理: message);// ..业务逻辑处理}} catch (InterruptedException e) {Thread.currentThread().interrupt();log.error(中断异常,e);}}}3测试接口
GetMapping(value /test3)public void redisQueueTest() {// Add some messages with delaysdelayedMessageService.addMessage(Message 1, 5000); // 5 seconds delaydelayedMessageService.addMessage(Message 2, 10000); // 10 seconds delaydelayedMessageService.addMessage(Message 3, 15000); // 15 seconds delay}说明
RDelayedQueue 是 Redisson 提供的延迟队列它将消息存储在指定的队列中直到消息到期才会被转移到该队列。它的主要作用包括 延迟消息管理我们可以使用 RDelayedQueue 的 offer 方法将消息添加到延迟队列并指定延迟时间消息在延迟时间到期前一直保留在 RDelayedQueue 中。消息转移一旦消息到期RDelayedQueue 会自动将消息转移到指定的RBlockingDeque 中。 RBlockingQueue是 Redisson 提供的阻塞队列它支持阻塞操作。主要作用包括 阻塞操作支持阻塞的 take 操作如果队列中没有元素会一直阻塞直到有元素可供消费。
总结 个人推荐使用Redisson 的RDelayedQueue 方式感觉更加可靠和简单一些当然zsetscore也可以是个不错选择毕竟更加灵活延迟消息还有其他不同的方案比如rocketmq、rabbitmq插件等假如项目中用了redis又不想引入更多的中间件可以尝试使用redis来实现为了测试这里例子都比较简单在实际使用过程中还要考虑补偿机制、幂等性等问题。
参考 1.https://blog.csdn.net/qq_34826261/article/details/120598731
2.https://github.com/redisson/redisson/wiki/7.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%9B%86%E5%90%88#715-%E5%BB%B6%E8%BF%9F%E9%98%9F%E5%88%97delayed-queue
