搜索
首页数据库RedisRedis如何实现延迟队列？方法介绍
Redis如何实现延迟队列？方法介绍
青灯夜游
青灯夜游
Jul 08, 2020 pm 04:01 PM
redis延迟队列

Redis如何实现延迟队列？方法介绍

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

1. 背景

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

  • 当订单一直处于未支付状态时，如何及时的关闭订单
  • 如何定期检查处于退款状态的订单是否已经退款成功
  • 在订单长时间没有收到下游系统的状态通知的时候，如何实现阶梯式的同步订单状态的策略
  • 在系统通知上游系统支付成功终态时，上游系统返回通知失败，如何进行异步通知实行分频率发送:15s 3m 10m 30m 30m 1h 2h 6h 15h

1.1 解决方案

  • 最简单的方式，定时扫表。例如对于订单支付失效要求比较高的，每2S扫表一次检查过期的订单进行主动关单操作。优点是简单，缺点是每分钟全局扫表，浪费资源，如果遇到表数据订单量即将过期的订单量很大，会造成关单延迟。

  • 使用RabbitMq或者其他MQ改造实现延迟队列，优点是，开源，现成的稳定的实现方案，缺点是：MQ是一个消息中间件，如果团队技术栈本来就有MQ，那还好，如果不是，那为了延迟队列而去部署一套MQ成本有点大

  • 使用Redis的zset、list的特性，我们可以利用redis来实现一个延迟队列RedisDelayQueue

2. 设计目标

  • 实时性：允许存在一定时间的秒级误差
  • 高可用性：支持单机、支持集群
  • 支持消息删除：业务会随时删除指定消息
  • 消息可靠性：保证至少被消费一次
  • 消息持久化：基于Redis自身的持久化特性，如果Redis数据丢失，意味着延迟消息的丢失，不过可以做主备和集群保证。这个可以考虑后续优化将消息持久化到MangoDB中

3. 设计方案

设计主要包含以下几点：

  • 将整个Redis当做消息池，以KV形式存储消息
  • 使用ZSET做优先队列，按照Score维持优先级
  • 使用LIST结构，以先进先出的方式消费
  • ZSET和LIST存储消息地址(对应消息池的每个KEY)
  • 自定义路由对象，存储ZSET和LIST名称，以点对点的方式将消息从ZSET路由到正确的LIST
  • 使用定时器维护路由
  • 根据TTL规则实现消息延迟

3.1 设计图

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

3.2 数据结构

  • ZING:DELAY_QUEUE:JOB_POOL 是一个Hash_Table结构，里面存储了所有延迟队列的信息。KV结构：K=prefix+projectName  field = topic+jobId  V=CONENT;V由客户端传入的数据，消费的时候回传
  • ZING:DELAY_QUEUE:BUCKET 延迟队列的有序集合ZSET，存放K=ID和需要的执行时间戳，根据时间戳排序
  • ZING:DELAY_QUEUE:QUEUE LIST结构，每个Topic一个LIST，list存放的都是当前需要被消费的JOB

图片仅供参考，基本可以描述整个流程的执行过程，图片源于文末的参考博客中
图片仅供参考，基本可以描述整个流程的执行过程，图片源于文末的参考博客中

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 基本概念

  • JOB：需要异步处理的任务，是延迟队列里的基本单元
  • Topic：一组相同类型Job的集合（队列）。供消费者来订阅

3.4.2 消息结构

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

  • jobId：Job的唯一标识。用来检索和删除指定的Job信息
  • topic：Job类型。可以理解成具体的业务名称
  • delay：Job需要延迟的时间。单位：秒。（服务端会将其转换为绝对时间）
  • body：Job的内容，供消费者做具体的业务处理，以json格式存储
  • retry：失败重试次数
  • url：通知URL

3.5 设计细节

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

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

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

  • 使用Redis的分布式锁来控制消息的搬运，从而避免消息被重复搬运导致的问题
  • 使用分布式锁来保证定时器的执行频率

4. 核心代码实现

4.1 技术说明

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

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

4.2 核心实体

4.2.1 Job新增对象

/**
 * 消息结构
 *
 * @author 睁眼看世界
 * @date 2020年1月15日
 */
@Data
public class Job implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * Job的唯一标识。用来检索和删除指定的Job信息
     */
    @NotBlank
    private String jobId;


    /**
     * Job类型。可以理解成具体的业务名称
     */
    @NotBlank
    private String topic;

    /**
     * Job需要延迟的时间。单位：秒。（服务端会将其转换为绝对时间）
     */
    private Long delay;

    /**
     * Job的内容，供消费者做具体的业务处理，以json格式存储
     */
    @NotBlank
    private String body;

    /**
     * 失败重试次数
     */
    private int retry = 0;

    /**
     * 通知URL
     */
    @NotBlank
    private String url;
}

4.2.2 Job删除对象

/**
 * 消息结构
 *
 * @author 睁眼看世界
 * @date 2020年1月15日
 */
@Data
public class JobDie implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * Job的唯一标识。用来检索和删除指定的Job信息
     */
    @NotBlank
    private String jobId;


    /**
     * Job类型。可以理解成具体的业务名称
     */
    @NotBlank
    private String topic;
}

4.3 搬运线程

/**
 * 搬运线程
 *
 * @author 睁眼看世界
 * @date 2020年1月17日
 */
@Slf4j
@Component
public class CarryJobScheduled {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 启动定时开启搬运JOB信息
     */
    @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<Object> bucketSet = redissonClient.getScoredSortedSet(RD_ZSET_BUCKET_PRE);
            long now = System.currentTimeMillis();
            Collection<Object> jobCollection = bucketSet.valueRange(0, false, now, true);
            List<String> jobList = jobCollection.stream().map(String::valueOf).collect(Collectors.toList());
            RList<String> 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
@Component
public class ReadyQueueContext {

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ConsumerService consumerService;

    /**
     * TOPIC消费线程
     */
    @PostConstruct
    public void startTopicConsumer() {
        TaskManager.doTask(this::runTopicThreads, "开启TOPIC消费线程");
    }

    /**
     * 开启TOPIC消费线程
     * 将所有可能出现的异常全部catch住，确保While(true)能够不中断
     */
    @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<String> queue = redissonClient.getBlockingQueue(RD_LIST_TOPIC_PRE);
                String topicId = queue.poll(60, TimeUnit.SECONDS);
                if (StringUtils.isEmpty(topicId)) {
                    continue;
                }

                // 2. 获取job元信息内容
                RMap<String, Job> jobPoolMap = redissonClient.getMap(JOB_POOL_KEY);
                Job job = jobPoolMap.get(topicId);

                // 3. 消费
                FutureTask<Boolean> 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<Object> 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

/**
 * 提供给外部服务的操作接口
 *
 * @author why
 * @date 2020年1月15日
 */
@Slf4j
@Service
public class RedisDelayQueueServiceImpl implements RedisDelayQueueService {

    @Autowired
    private RedissonClient redissonClient;


    /**
     * 添加job元信息
     *
     * @param job 元信息
     */
    @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<String, Job> 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<Object> 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();
            }
        }
    }


    /**
     * 删除job信息
     *
     * @param job 元信息
     */
    @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<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
            jobPool.remove(topicId);

            RScoredSortedSet<Object> 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. 源码

更多详细源码请在下面地址中获取

  • RedisDelayQueue实现 zing-delay-queue（https://gitee.com/whyCodeData/zing-project/tree/master/zing-delay-queue）
  • RedissonStarter redisson-spring-boot-starter（https://gitee.com/whyCodeData/zing-project/tree/master/zing-starter/redisson-spring-boot-starter）
  • 项目应用 zing-pay（https://gitee.com/whyCodeData/zing-pay）

7. 参考

  • https://tech.youzan.com/queuing_delay/
  • https://blog.csdn.net/u010634066/article/details/98864764

更多redis知识，请关注：redis入门教程栏目。

以上是Redis如何实现延迟队列？方法介绍的详细内容。更多信息请关注PHP中文网其他相关文章！

声明
本文转载于：csdn。如有侵权，请联系admin@php.cn删除
相关文章
es和redis区别es和redis区别Jul 06, 2019 pm 01:45 PM

Redis是现在最热门的key-value数据库，Redis的最大特点是key-value存储所带来的简单和高性能；相较于MongoDB和Redis，晚一年发布的ES可能知名度要低一些，ES的特点是搜索，ES是围绕搜索设计的。

一起来聊聊Redis有什么优势和特点一起来聊聊Redis有什么优势和特点May 16, 2022 pm 06:04 PM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了关于redis的一些优势和特点，Redis 是一个开源的使用ANSI C语言编写、遵守 BSD 协议、支持网络、可基于内存、分布式存储数据库，下面一起来看一下，希望对大家有帮助。

实例详解Redis Cluster集群收缩主从节点实例详解Redis Cluster集群收缩主从节点Apr 21, 2022 pm 06:23 PM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了Redis Cluster集群收缩主从节点的相关问题，包括了Cluster集群收缩概念、将6390主节点从集群中收缩、验证数据迁移过程是否导致数据异常等，希望对大家有帮助。

Redis实现排行榜及相同积分按时间排序功能的实现Redis实现排行榜及相同积分按时间排序功能的实现Aug 22, 2022 pm 05:51 PM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了Redis实现排行榜及相同积分按时间排序,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，希望对大家有帮助。

详细解析Redis中命令的原子性详细解析Redis中命令的原子性Jun 01, 2022 am 11:58 AM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了关于原子操作中命令原子性的相关问题，包括了处理并发的方案、编程模型、多IO线程以及单命令的相关内容，下面一起看一下，希望对大家有帮助。

一文搞懂redis的bitmap一文搞懂redis的bitmapApr 27, 2022 pm 07:48 PM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了bitmap问题，Redis 为我们提供了位图这一数据结构，位图数据结构其实并不是一个全新的玩意，我们可以简单的认为就是个数组，只是里面的内容只能为0或1而已，希望对大家有帮助。

实例详解Redis实现排行榜及相同积分按时间排序功能的实现实例详解Redis实现排行榜及相同积分按时间排序功能的实现Aug 26, 2022 pm 02:09 PM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了Redis实现排行榜及相同积分按时间排序,本文通过实例代码给大家介绍的非常详细，下面一起来看一下，希望对大家有帮助。

一起聊聊Redis实现秒杀的问题一起聊聊Redis实现秒杀的问题May 27, 2022 am 11:40 AM

本篇文章给大家带来了关于redis的相关知识，其中主要介绍了关于实现秒杀的相关内容，包括了秒杀逻辑、存在的链接超时、超卖和库存遗留的问题，下面一起来看一下，希望对大家有帮助。

See all articles

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序，用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

AI Hentai Generator

AI Hentai Generator

免费生成ai无尽的。

显示更多

热门文章

R.E.P.O.能量晶体解释及其做什么（黄色晶体）
2 周前By尊渡假赌尊渡假赌尊渡假赌
仓库：如何复兴队友
4 周前By尊渡假赌尊渡假赌尊渡假赌
Hello Kitty Island冒险：如何获得巨型种子
4 周前By尊渡假赌尊渡假赌尊渡假赌
击败分裂小说需要多长时间？
3 周前ByDDD
R.E.P.O.保存文件位置：在哪里以及如何保护它？
3 周前ByDDD
显示更多

热工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

DVWA

DVWA

Damn Vulnerable Web App (DVWA) 是一个PHP/MySQL的Web应用程序，非常容易受到攻击。它的主要目标是成为安全专业人员在合法环境中测试自己的技能和工具的辅助工具，帮助Web开发人员更好地理解保护Web应用程序的过程，并帮助教师/学生在课堂环境中教授/学习Web应用程序安全。DVWA的目标是通过简单直接的界面练习一些最常见的Web漏洞，难度各不相同。请注意，该软件中

SecLists

SecLists

SecLists是最终安全测试人员的伙伴。它是一个包含各种类型列表的集合，这些列表在安全评估过程中经常使用，都在一个地方。SecLists通过方便地提供安全测试人员可能需要的所有列表，帮助提高安全测试的效率和生产力。列表类型包括用户名、密码、URL、模糊测试有效载荷、敏感数据模式、Web shell等等。测试人员只需将此存储库拉到新的测试机上，他就可以访问到所需的每种类型的列表。

Atom编辑器mac版下载

Atom编辑器mac版下载

最流行的的开源编辑器

ZendStudio 13.5.1 Mac

ZendStudio 13.5.1 Mac

功能强大的PHP集成开发环境

显示更多

热门话题

gmail邮箱登陆入口在哪里
7323
9
Java教程
1625
14
CakePHP 教程
1350
46
Laravel 教程
1262
25
PHP教程
1209
29
显示更多