Redis實現延遲隊列

時間  2020-03-17
標籤 redis 實現 延遲 隊列 欄目 Redis 简体版
原文   原文鏈接

延遲隊列，顧名思義它是一種帶有延遲功能的消息隊列。那麼，是在什麼場景下我才須要這樣的隊列呢？git

1. 背景

咱們先看看如下業務場景：redis

  • 當訂單一直處於未支付狀態時，如何及時的關閉訂單
  • 如何按期檢查處於退款狀態的訂單是否已經退款成功
  • 在訂單長時間沒有收到下游系統的狀態通知的時候，如何實現階梯式的同步訂單狀態的策略
  • 在系統通知上游系統支付成功終態時，上游系統返回通知失敗，如何進行異步通知實行分頻率發送: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. 設計方案

設計主要包含如下幾點：spring

  • 將整個Redis當作消息池，以KV形式存儲消息
  • 使用ZSET作優先隊列，按照Score維持優先級
  • 使用LIST結構，以先進先出的方式消費
  • ZSET和LIST存儲消息地址(對應消息池的每一個KEY)
  • 自定義路由對象，存儲ZSET和LIST名稱，以點對點的方式將消息從ZSET路由到正確的LIST
  • 使用定時器維護路由
  • 根據TTL規則實現消息延遲

3.1 設計圖

仍是基於有讚的延遲隊列設計，進行優化改造及代碼實現。有贊設計
imagejson

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

88TsOI.png
圖片僅供參考，基本能夠描述整個流程的執行過程，圖片源於文末的參考博客中數據結構

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，分佈式鎖，定時器ide

注意：本項目沒有實現設計方案中的多Queue消費，只開啓了一個QUEUE，這個待之後優化spring-boot

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. 源碼

更多詳細源碼請在下面地址中獲取oop

7. 參考

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程