文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

Redis实现延迟队列方法介绍

2023-09-17 11:22

关注

延迟队列，顾名思义它是一种带有延迟功能的消息队列。那么，是在什么场景下我才需要这样的队列呢？

1. 背景

我们先看看以下业务场景：

1.1 解决方案

2. 设计目标

3. 设计方案

设计主要包含以下几点：

3.1 设计图

还是基于有赞的延迟队列设计，进行优化改造及代码实现。有赞设计

3.2 数据结构

图片仅供参考，基本可以描述整个流程的执行过程

3.3 任务的生命周期

  1. 新增一个JOB，会在ZING:DELAY_QUEUE:JOB_POOL中插入一条数据，记录了业务方消费方。ZING:DELAY_QUEUE:BUCKET也会插入一条记录，记录执行的时间戳
  2. 搬运线程会去ZING:DELAY_QUEUE:BUCKET中查找哪些执行时间戳的RunTimeMillis比现在的时间小，将这些记录全部删除；同时会解析出每个任务的Topic是什么，然后将这些任务PUSH到TOPIC对应的列表ZING:DELAY_QUEUE:QUEUE中
  3. 每个TOPIC的LIST都会有一个监听线程去批量获取LIST中的待消费数据，获取到的数据全部扔给这个TOPIC的消费线程池
  4. 消费线程池执行会去ZING:DELAY_QUEUE:JOB_POOL查找数据结构，返回给回调结构，执行回调方法。

3.4 设计要点

3.4.1 基本概念

3.4.2 消息结构

每个JOB必须包含以下几个属性

3.5 设计细节

3.5.1 如何快速消费ZING:DELAY_QUEUE:QUEUE

最简单的实现方式就是使用定时器进行秒级扫描，为了保证消息执行的时效性，可以设置每1S请求Redis一次，判断队列中是否有待消费的JOB。但是这样会存在一个问题，如果queue中一直没有可消费的JOB，那频繁的扫描就失去了意义，也浪费了资源，幸好LIST中有一个BLPOP阻塞原语，如果list中有数据就会立马返回,如果没有数据就会一直阻塞在那里，直到有数据返回,可以设置阻塞的超时时间,超时会返回NULL;具体的实现方式及策略会在代码中进行具体的实现介绍

3.5.2 避免定时导致的消息重复搬运及消费

4. 核心代码实现

4.1 技术说明

技术栈：SpringBoot，Redisson，Redis，分布式锁，定时器

注意：本项目没有实现设计方案中的多Queue消费，只开启了一个QUEUE，这个待以后优化

4.2 核心实体

4.2.1 Job新增对象

@Datapublic class Job implements Serializable {  private static final long serialVersionUID = 1L;   @NotBlank private String jobId;    @NotBlank private String topic;   private Long delay;   @NotBlank private String body;   private int retry = 0;   @NotBlank private String url;}4.2.2 Job删除对象@Datapublic class JobDie implements Serializable {  private static final long serialVersionUID = 1L;   @NotBlank private String jobId;    @NotBlank private String topic;}

4.3 搬运线程

@Slf4j@Componentpublic class CarryJobScheduled {  @Autowired private RedissonClient redissonClient;   @Scheduled(cron = "*/1 * * * * *") public void carryJobToQueue() { System.out.println("carryJobToQueue --->"); RLock lock = redissonClient.getLock(RedisQueueKey.CARRY_THREAD_LOCK); try { boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS); if (!lockFlag) { throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL); } RScoredSortedSet bucketSet = redissonClient.getScoredSortedSet(RD_ZSET_BUCKET_PRE); long now = System.currentTimeMillis(); Collection jobCollection = bucketSet.valueRange(0, false, now, true); List jobList = jobCollection.stream().map(String::valueOf).collect(Collectors.toList()); RList readyQueue = redissonClient.getList(RD_LIST_TOPIC_PRE); readyQueue.addAll(jobList); bucketSet.removeAllAsync(jobList); } catch (InterruptedException e) { log.error("carryJobToQueue error", e); } finally { if (lock != null) { lock.unlock(); } } }}  
4.4 消费线程
 
@Slf4j@Componentpublic class ReadyQueueContext {  @Autowired private RedissonClient redissonClient;  @Autowired private ConsumerService consumerService;   @PostConstruct public void startTopicConsumer() { TaskManager.doTask(this::runTopicThreads, "开启TOPIC消费线程"); }   @SuppressWarnings("InfiniteLoopStatement") private void runTopicThreads() { while (true) { RLock lock = null; try { lock = redissonClient.getLock(CONSUMER_TOPIC_LOCK); } catch (Exception e) { log.error("runTopicThreads getLock error", e); } try { if (lock == null) { continue; } // 分布式锁时间比Blpop阻塞时间多1S，避免出现释放锁的时候，锁已经超时释放，unlock报错 boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS); if (!lockFlag) { continue; }  // 1. 获取ReadyQueue中待消费的数据 RBlockingQueue queue = redissonClient.getBlockingQueue(RD_LIST_TOPIC_PRE); String topicId = queue.poll(60, TimeUnit.SECONDS); if (StringUtils.isEmpty(topicId)) { continue; }  // 2. 获取job元信息内容 RMap jobPoolMap = redissonClient.getMap(JOB_POOL_KEY); Job job = jobPoolMap.get(topicId);  // 3. 消费 FutureTask taskResult = TaskManager.doFutureTask(() -> consumerService.consumerMessage(job.getUrl(), job.getBody()), job.getTopic() + "-->消费JobId-->" + job.getJobId()); if (taskResult.get()) { // 3.1 消费成功，删除JobPool和DelayBucket的job信息 jobPoolMap.remove(topicId); } else { int retrySum = job.getRetry() + 1; // 3.2 消费失败，则根据策略重新加入Bucket  // 如果重试次数大于5，则将jobPool中的数据删除，持久化到DB if (retrySum > RetryStrategyEnum.RETRY_FIVE.getRetry()) { jobPoolMap.remove(topicId); continue; } job.setRetry(retrySum); long nextTime = job.getDelay() + RetryStrategyEnum.getDelayTime(job.getRetry()) * 1000; log.info("next retryTime is [{}]", DateUtil.long2Str(nextTime)); RScoredSortedSet delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE); delayBucket.add(nextTime, topicId); // 3.3 更新元信息失败次数 jobPoolMap.put(topicId, job); } } catch (Exception e) { log.error("runTopicThreads error", e); } finally { if (lock != null) { try { lock.unlock(); } catch (Exception e) { log.error("runTopicThreads unlock error", e); } } } } }}  
4.5 添加及删除JOB
 
@Slf4j@Servicepublic class RedisDelayQueueServiceImpl implements RedisDelayQueueService {  @Autowired private RedissonClient redissonClient;    @Override public void addJob(Job job) {  RLock lock = redissonClient.getLock(ADD_JOB_LOCK + job.getJobId()); try { boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS); if (!lockFlag) { throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL); } String topicId = RedisQueueKey.getTopicId(job.getTopic(), job.getJobId());  // 1. 将job添加到 JobPool中 RMap jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY); if (jobPool.get(topicId) != null) { throw new BusinessException(ErrorMessageEnum.JOB_ALREADY_EXIST); }  jobPool.put(topicId, job);  // 2. 将job添加到 DelayBucket中 RScoredSortedSet delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE); delayBucket.add(job.getDelay(), topicId); } catch (InterruptedException e) { log.error("addJob error", e); } finally { if (lock != null) { lock.unlock(); } } }    @Override public void deleteJob(JobDie jobDie) {  RLock lock = redissonClient.getLock(DELETE_JOB_LOCK + jobDie.getJobId()); try { boolean lockFlag = lock.tryLock(LOCK_WAIT_TIME, LOCK_RELEASE_TIME, TimeUnit.SECONDS); if (!lockFlag) { throw new BusinessException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL); } String topicId = RedisQueueKey.getTopicId(jobDie.getTopic(), jobDie.getJobId());  RMap jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY); jobPool.remove(topicId);  RScoredSortedSet delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE); delayBucket.remove(topicId); } catch (InterruptedException e) { log.error("addJob error", e); } finally { if (lock != null) { lock.unlock(); } } }}  
5. 待优化的内容
 
  1. 目前只有一个Queue队列存放消息，当需要消费的消息大量堆积后，会影响消息通知的时效。改进的办法是，开启多个Queue，进行消息路由，再开启多个消费线程进行消费，提供吞吐量
  2. 消息没有进行持久化，存在风险，后续会将消息持久化到MangoDB中
 
6. 源码
 
更多详细源码请在下面地址中获取
 
百度网盘： 百度网盘 请输入提取码
来源地址：https://blog.csdn.net/weixin_64051447/article/details/130491324
阅读原文内容投诉
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。 
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
软考中级精品资料免费领 
  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 资料下载
  • 历年真题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）
     难度      813人已做 
     查看 
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析
     难度      354人已做 
     查看 
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析
     难度      318人已做 
     查看 
  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）
     难度      435人已做 
     查看 
  • 2024年上半年系统架构设计师考试综合知识真题
     难度      224人已做 
     查看 
相关文章
发现更多好内容
猜你喜欢
AI推送时光机
Redis实现延迟队列方法介绍
后端开发2023-09-17
Redis实现延迟队列的方法是什么
后端开发2023-07-05
如何实现Redis延迟队列
后端开发2024-04-02
redis延迟队列如何实现
redis延迟队列如何实现
后端开发2024-06-12
Redis如何实现延迟队列
后端开发2023-04-28
Redis延迟队列和分布式延迟队列的简答实现
后端开发2024-04-02
Go+Redis实现延迟队列实操
后端开发2024-04-02
Redis实现延迟队列的方案总结
后端开发Redis延迟队列2024-11-29
怎么在Redis中实现延迟队列和分布式延迟队列
后端开发2023-06-15
Redis优雅地实现延迟队列的方法分享
后端开发2023-02-26
使用Redis怎么实现延迟队列
后端开发2023-06-15
如何实现 Java 延迟队列？(java延迟队列怎么实现)
如何实现 Java 延迟队列？(java延迟队列怎么实现)
后端开发Java2024-12-20
.Net实现延迟队列
后端开发2024-04-02
基于Redis延迟队列的实现代码
后端开发2024-04-02
SpringBoot实现redis延迟队列的示例代码
SpringBoot实现redis延迟队列的示例代码
后端开发2024-04-02
Redis实现延迟队列的全流程详解
后端开发2023-03-14
如何在Redis中实现延迟任务队列
如何在Redis中实现延迟任务队列
后端开发2024-04-09
Golang实现基于Redis的可靠延迟队列
后端开发2022-06-22
Java实现异步延迟队列的方法详解
后端开发2023-03-22
RabbitMQ 如何实现延迟队列？
后端开发RabbitMQ延迟队列2024-11-30
 位置：首页-资讯-后端开发
 咦！没有更多了？去看看其它编程学习网 内容吧 
首页课程
资料下载问答资讯