How to implement a delay queue?

One 、 What is delay queue

Delay queue , seeing the name of a thing one thinks of its function , That is, when the element is in line , A delay time will be specified , Expect to process this element after a specified time .

There are many ways to implement delay queues , We can implement our own delay queue in a familiar way . Generally, the delay queue can be provided to the whole company as a basic public service , This approach requires independent maintenance of a project , Yes qps stability , Data consistency , Higher accuracy requirements , It can be realized by time wheel algorithm .  I came across a scenario using delay queues in the project , Because... Is used in the project redis, So I use it redis  A delay queue is implemented to meet the requirements , This article introduces how to use redis  Implement a delay queue .

The author's lovely son

Two 、 Scenarios for delay queues ？

There are many scenarios for delay queues ：

2.1 For example, overtime customs clearance ：

That is, the user did not pay immediately after placing an order on the e-commerce platform , The order will be closed automatically after the specified time .

2.2  For example, callback retry ：

For asynchronous interfaces , If a callback is given to the caller , When the callback fails due to network failure or other reasons , We can use the delay strategy to retry the callback interface of the caller . To avoid callback failure caused by network jitter or other reasons , The delay strategy we can adopt is  1min  5 min  10 min  30 min  1hour  Callback at equal intervals .

2.3  Meeting reminder ：

Like what we use lark meeting , Before the meeting 10 Minutes to remind participants , This function can also be realized by using delay queue .

2.4  Various delay reminders ：

For example, when the user places an order but fails to pay , The system is before the customs declaration 10 Minutes to remind users to pay . For example, one of the needs of used cars I have made is to remind buyers to bid as soon as possible .

Delay queues are used in these scenarios , In fact, the above scenario can also be achieved by using timed tasks , But compared with scheduled tasks , The time control of the delay team is more accurate , The delay queue does not need to scan the library table , Less consumption on the system .

3、 ... and 、 Implementation of delay queue ？

•DelayQueue,

This is jdk A self-contained delay queue , be located java.util.concurrent  It's a bag , It is a bounded blocking queue , It encapsulates a  PriorityQueue( Priority queue )

PriorityQueue  The complete binary heap is used internally to sort queue elements , Direction  DelayQueue  When you add elements to the queue , Will give the element a  Delay( Delay time ) As a sort condition , The smallest element in the queue will take precedence at the head of the queue . The only elements in the queue are  Delay  Time to get out of the queue . Interested students can check the source code in detail .

•Quartz  Timing task ,

The requirements for time accuracy are not high , Tasks with a small amount of data , You can use timed tasks instead of delayed queues .

•redis  Overdue callback ,

We can turn on monitoring key  Whether the event is overdue , once key  Expiration triggers a callback  event . So we can set key The expiration time of , To achieve the effect of delayed queue .

•redis  sorted set,

Mainly to make use of  zset  Of score  attribute , We turn the delay time into （ current time + Delay Time ）（ Time unit millisecond ） As scroe  attribute . Then start a consumption thread rotation redis queue , When score When the value of the property is less than the current time , Prove that the delay message expires , It can be consumed .

•Mq,

adopt rabbitMq   or  rocketMq  Delay queue can be realized , The specific implementation method is omitted .

• Time wheel ,

be based on kafka  or  netty  Time wheel algorithm to achieve delay queue , The implementation method is omitted , Interested students can check by themselves .

There are many ways to implement delay queues , We can implement our own delay queue in a familiar way . Generally, the delay queue can be provided to the whole company as a basic public service , This approach requires independent maintenance of a project , Yes qps stability , Data consistency , Higher accuracy requirements , It can be realized by time wheel algorithm .

I came across a scenario using delay queues in the project , Because... Is used in the project redis, So I use it redis  A delay queue is implemented to meet the requirements , Here's how to use redis  Implement a delay queue .

Four 、 How to use redis Implement delay queue ？

adopt  redis  There are two ways to implement delay queue , The first is redis  Expiration callback of , But this method needs to be modified redis  And restart the service , This is for what we are using  redis  Service is difficult .

So we use the second method to implement a delay queue ：

4.1 We use redisson  To achieve , First introduce redisson  package

 <dependency>
 <groupId>org.redisson</groupId>
 <artifactId>redisson-spring-boot-starter</artifactId>
 <version>3.16.3</version>
 </dependency>

4.2、 To configure redisson

redisson:
 redisModel: SINGLE
 singleConfig:
 address: redis://127.0.0.1:6379

4.3 Delay service ： Add delay task , Cancel delay task

public class RedisDelayQueueServiceImpl implements RedisDelayQueueService {

 @Autowired
 private RedissonClient redissonClient;

 /**
 *  Add a delay task , add to job Meta information
 *
 * @param job  Meta information ,job Contained in the  taskId  Primary key ,topicId  queue id,retryNum  Retry count ,delaytime  Delay Time   etc. , Not expanded here .
 */
 @Override
 public void addJob(Job job) {
 // Add distributed lock , Prevent repeated addition of tasks
 RLock lock = redissonClient.getLock(RedisQueueKey.ADD_JOB_LOCK + job.getJobId());
 try {
 boolean lockFlag = lock.tryLock(RedisQueueKey.LOCK_WAIT_TIME, RedisQueueKey.LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BizException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL.getInfo());
 }
 String topicId = RedisQueueKey.getTopicId(job.getTopic(), job.getJobId());

 // 1.  take job Add to  JobPool in ,jobpool  As a global index , All unexecuted tasks exist jobPool  in .
 RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
 if (jobPool.get(topicId) != null) {
 throw new BizException(ErrorMessageEnum.JOB_ALREADY_EXIST.getInfo());
 }

 jobPool.put(topicId, job);

 // 2.  take job Add to  DelayBucket in , Sort by delay time
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.add(job.getDelay(), topicId);
 } catch (InterruptedException e) {
 log.error(&quot;addJob error&quot;, e);
 throw new BizException(&quot;add delay job error,reason:&quot; + e.getMessage());
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }


 /**
 *  Delete job Information , Why delete job  Information ？
 *  When we are sure that the last delayed task is unnecessary , We can cancel the execution of delayed tasks in advance .
 *
 * @param jobDie  Meta information
 */
 @Override
 public void deleteJob(JobDie jobDie) {

 RLock lock = redissonClient.getLock(RedisQueueKey.DELETE_JOB_LOCK + jobDie.getJobId());
 try {
 boolean lockFlag = lock.tryLock(RedisQueueKey.LOCK_WAIT_TIME, RedisQueueKey.LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BizException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL.getInfo());
 }
 String topicId = RedisQueueKey.getTopicId(jobDie.getTopic(), jobDie.getJobId());
 // Delete task from global index .
 RMap<String, Job> jobPool = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
 jobPool.remove(topicId);
 // from zset  Delete task in
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.remove(topicId);
 } catch (InterruptedException e) {
 log.error(&quot;addJob error&quot;, e);
 throw new BizException(&quot;delete job error,reason:&quot; + e.getMessage());
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }
}

4.4 Handling thread :

The purpose of handling threads is to  zset  The expired tasks in are moved to the consumption queue , Tasks in the consumption queue will be consumed by the consumption thread . The reason for adding a consumption queue , Considering our consumption capacity and data security ,

If the consumption capacity is relatively weak , It may cause the consumer thread to block , Or data loss

. We put the expired tasks into a blocking queue , The consumption thread can consume in sequence .

This place can still be optimized , For example, you can drop the warehouse , Multiple blocking queues can be established , Each blocking queue can specify a thread pool for consumption .

@Slf4j
@Component
public class CarryJobScheduled {

 @Autowired
 private RedissonClient redissonClient;

 /**
 *  Start timing start handling JOB Information
 */
 @Scheduled(cron = "*/1 * * * * *")
 public void carryJobToQueue() {
 //System.out.println("carryJobToQueue --->");
 RLock lock = redissonClient.getLock(RedisQueueKey.CARRY_THREAD_LOCK);
 try {
 boolean lockFlag = lock.tryLock(RedisQueueKey.LOCK_WAIT_TIME, RedisQueueKey.LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 throw new BizException(ErrorMessageEnum.ACQUIRE_LOCK_FAIL.getInfo());
 }
 //  Take out the expired tasks
 RScoredSortedSet<Object> bucketSet = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 long now = System.currentTimeMillis();
 Collection<Object> jobCollection = bucketSet.valueRange(0, false, now, true);
 if (CollectionUtils.isEmpty(jobCollection)) {
 return;
 }
 
 //  Move the expired tasks to the consumption queue ,zset  Delete the task in
 List<String> jobList = jobCollection.stream().map(String::valueOf).collect(Collectors.toList());
 RList<String> readyQueue = redissonClient.getList(RedisQueueKey.RD_LIST_TOPIC_PRE);

 if (CollectionUtils.isEmpty(jobList)) {
 return;
 }
 if (readyQueue.addAll(jobList)) {
 bucketSet.removeAllAsync(jobList);
 }

 } catch (InterruptedException e) {
 log.error(&quot;carryJobToQueue error&quot;, e);
 } finally {
 if (lock != null) {
 lock.unlock();
 }
 }
 }
}

4.5 Consumer thread ：

Start a consumption thread , The consuming thread will consume the expired tasks in the blocking queue , among  ConsumerService  You can use a strategic model , According to topic  Different business processes .

@Slf4j
@Component
public class ReadyQueueContext {

 @Autowired
 private RedissonClient redissonClient;

 @Autowired
 private ConsumerService consumerService;

 /**
 * TOPIC consumption
 */
 @PostConstruct
 public void startTopicConsumer() {
 TaskManager.doTask(this::runTopicThreads, " Turn on TOPIC Consumer thread ");
 }

 /**
 *  Turn on TOPIC Consumer thread
 *  Put all possible exceptions catch live , Make sure While(true) Can be uninterrupted
 */
 @SuppressWarnings("InfiniteLoopStatement")
 private void runTopicThreads() {
 while (true) {
 RLock lock = null;
 try {
 lock = redissonClient.getLock(RedisQueueKey.CONSUMER_TOPIC_LOCK);
 } catch (Exception e) {
 log.error("runTopicThreads getLock error", e);
 }
 try {
 if (lock == null) {
 continue;
 }
 //  Distributed lock time ratio Blpop More blocking time 1S, Avoid releasing the lock , The lock has been released overtime ,unlock Report errors
 boolean lockFlag = lock.tryLock(RedisQueueKey.LOCK_WAIT_TIME, RedisQueueKey.LOCK_RELEASE_TIME, TimeUnit.SECONDS);
 if (!lockFlag) {
 continue;
 }

 // 1.  obtain ReadyQueue Data to be consumed
 RBlockingQueue<String> queue = redissonClient.getBlockingQueue(RedisQueueKey.RD_LIST_TOPIC_PRE);
 String topicId = queue.poll(60, TimeUnit.SECONDS);
 if (StringUtils.isEmpty(topicId)) {
 continue;
 }

 // 2.  obtain job Meta information content
 RMap<String, Job> jobPoolMap = redissonClient.getMap(RedisQueueKey.JOB_POOL_KEY);
 Job job = jobPoolMap.get(topicId);

 // 3.  consumption
 FutureTask<Boolean> taskResult = TaskManager.doFutureTask(() -> consumerService.consumerMessage(job.getJobId(),job.getTopic(), job.getBody()), job.getTopic() + &quot;--> consumption JobId-->&quot; + job.getJobId());
 if (taskResult.get()) {
 // 3.1  Consumption success , Delete JobPool and DelayBucket Of job Information
 jobPoolMap.remove(topicId);
 } else {
 /**
 *  Retries are 0 Go straight back
 */
 if (job.getRetry() == 0) {
 return;
 }
 int retrySum = job.getRetry() + 1;
 // 3.2  Consumption failure , Then rejoin according to the strategy Bucket

 //  If the number of retries is greater than 5, Will jobPool Data deletion in , Persist to 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(&quot;next retryTime is [{}]&quot;, DateUtil.long2Str(nextTime));
 RScoredSortedSet<Object> delayBucket = redissonClient.getScoredSortedSet(RedisQueueKey.RD_ZSET_BUCKET_PRE);
 delayBucket.add(nextTime, topicId);
 // 3.3  Failed times to update meta information
 jobPoolMap.put(topicId, job);
 }
 } catch (Exception e) {
 log.error(&quot;runTopicThreads error&quot;, e);
 } finally {
 if (lock != null) {
 try {
 lock.unlock();
 } catch (Exception e) {
 log.error(&quot;runTopicThreads unlock error&quot;, e);
 }
 }
 }
 }
 }

About lingchuang group （Advance Intelligence Group）

Lingchuang group was founded in  2016  year , Committed to localized application through scientific and technological innovation , Transform and reshape the financial and retail industries , Create a service for consumers with diversified business layout 、 The ecosystem of enterprises and merchants . The group consists of two parts: enterprise business and consumer business , Enterprise business includes  ADVANCE.AI  and  Ginee, They are banks 、 Finance 、 Financial technology 、 Retail and e-commerce industry customers provide services based on  AI  Digital authentication technology 、 Risk management products and omni channel e-commerce service solutions ; Consumer business  Atome Financial  Including Asia's leading pay after pay platform  Atome  And digital financial services .

2021  year  9  month , Lingchuang group announced the completion of super  4  Billion dollars  D  Round of funding , After the completion of financing, the valuation of lingchuang group has exceeded  20  Billion dollars , Become one of the largest independent technology start-ups in Singapore .

Looking back  BREAK AWAY

Spring data JPA  Practice and principle analysis

How to solve the problem of... In the scenario of massive data update  Mysql  The deadlock problem

Enterprise class  APIs  Safety practice guide  （ It is suggested that junior and intermediate engineers collect ）

Cypress UI  Automated testing framework

serverless Make our operation and maintenance easier

▼  If you think this article is helpful to you , Inspired , Welcome to my collection ：

1、

give the thumbs-up 、 Focus on

Lingchuang group

, Get the latest technology sharing and company dynamics .

2、 Pay attention to our official account  &  Zhihu No 「 Lingchuang group  Advance Group」 Or visit

Official website

, Learn more about enterprise dynamics .