java + redis zset实现延迟队列(定时到期执行任务)

createh51周前 (03-27)技术教程2

在Redis中,zet作为有序集合,可以利用其有序的特性,将任务添加到zset中,将任务的到期时间作为score,利用zset的默认有序特性,zrangewithscores可以获取score值最小的元素(也就是最近到期的任务),判断系统时间与该任务的到期时间大小,如果达到到期时间,就执行业务,并删除该到期任务,继续判断下一个元素,如果没有到期,就sleep一段时间(比如1秒),如果集合为空,也sleep一段时间。

1. 添加依赖


            redis.clients
            jedis
            3.3.0
        

2. 测试代码

package com.demo;


import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;

/**
 * 基于redis的延迟队列
 */
public class RedisDelayQueue {

    public static void main(String[] args) {

        System.out.println("begin time:" + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));

        RedisProduceThread produceThread=new RedisProduceThread();
        produceThread.start();

        RedisConsumeThread consumeThread=new RedisConsumeThread();
        consumeThread.start();

    }

    public static class DelayTask {

        /* 触发时间*/
        private long time;
        private String name;

        public long getTime() {
            return time;
        }

        public void setTime(long time) {
            this.time = time;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }
    }

    // 添加任务线程
    public static class RedisProduceThread extends Thread {

        public RedisProduceThread() {
        }

        @Override
        public void run() {

            Jedis jedis = new Jedis("127.0.0.1",6379);

            while (true)
            {
                long timeMillis = System.currentTimeMillis();

                Random rnd = new Random();
                int i = rnd.nextInt(30);

                double delay = timeMillis / 1000 + i;

                jedis.zadd("myzset", delay, "item-" + i);

                Double doubleDelay = delay;
                long longDelay = doubleDelay.longValue();

                System.out.println("添加业务:item-" + i + ",添加时间:" + timeMillis / 1000 + " ,到期时间:" + longDelay + ",延迟时间:" + i + " 秒");

                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }


        }
    }

    // 读取到期任务线程
    public static class RedisConsumeThread extends Thread {

        public RedisConsumeThread() {
        }

        @Override
        public void run() {

            Jedis jedis = new Jedis("127.0.0.1",6379);

            while (true) {

                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                // 从redis读取时间最小的数据
                long timestamp = System.currentTimeMillis() / 1000;

                Set myzset = jedis.zrangeWithScores("myzset", 0, 1);

                // 如果读取记录为空
                if(myzset.isEmpty())
                {
                    // 延时1秒
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }

                Iterator iterator = myzset.iterator();

                while (iterator.hasNext())
                {
                    Tuple tuple = iterator.next();
                    String item = tuple.getElement();
                    Double score = tuple.getScore();

                    // 如果当前记录到期
                    if(timestamp >= score)
                    {
                        long lscore = score.longValue();

                        // 执行业务处理
                        System.out.println("到期业务:" + item + " ,到期时间:" + lscore + ",系统时间:" + timestamp);

                        // 处理完成后,删除当前记录
                        jedis.zrem("myzset", item);

                        // 继续循环读取下一条
                    }
                    else
                    {
                        // 最小记录未到期,延时1秒
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                }


            }

        }
    }

}

3. 执行测试

添加业务:item-1,添加时间:1645515070 ,到期时间:1645515071,延迟时间:1 秒
到期业务:item-5 ,到期时间:1645515069,系统时间:1645515070
到期业务:item-1 ,到期时间:1645515071,系统时间:1645515071
添加业务:item-5,添加时间:1645515073 ,到期时间:1645515078,延迟时间:5 秒
到期业务:item-15 ,到期时间:1645515073,系统时间:1645515074
添加业务:item-23,添加时间:1645515076 ,到期时间:1645515099,延迟时间:23 秒
添加业务:item-11,添加时间:1645515079 ,到期时间:1645515090,延迟时间:11 秒
到期业务:item-5 ,到期时间:1645515078,系统时间:1645515079
添加业务:item-5,添加时间:1645515082 ,到期时间:1645515087,延迟时间:5 秒
添加业务:item-7,添加时间:1645515085 ,到期时间:1645515092,延迟时间:7 秒
添加业务:item-29,添加时间:1645515088 ,到期时间:1645515117,延迟时间:29 秒
到期业务:item-20 ,到期时间:1645515087,系统时间:1645515088
到期业务:item-5 ,到期时间:1645515087,系统时间:1645515088
到期业务:item-11 ,到期时间:1645515090,系统时间:1645515090

可以看到添加业务的时间加上延迟时间就是业务到期时间,在业务到期的下一秒,就输出了到期提示。

可以根据业务量的大小,每次读取的数据可以是一条数据,也可以是多条数据。一般情况下,每秒做一次检查可以满足大多数的业务需要,特殊情况下,可以将sleep的时间缩小(比如500ms或者300ms),这样可以做到更大的精确性。

相关文章

Spring Boot 架构下的订单自动取消机制:定时任务篇

引言在电子商务领域,确保交易流程的顺畅和高效至关重要。一个常见的场景是,用户生成订单后,系统会给予一定的支付时间窗口,如果在这个窗口内用户未完成支付,订单应当自动取消,以避免资源锁定和库存占用。本文将...

定时任务优化总结(从半个小时优化到秒级)

整体优化思路:1. 按需查询、2. 分小批次游标查询、3. JED场景下按数据库分片分组更新 、4. 精准定位要处理的数据、5.负载均衡业务背景:站外广告投放平台在做推广管理状态优化重构的时候,引入了...

老板喊你设计一个高效的定时任务系统

【51CTO.com原创稿件】今天想跟大家一起探讨一个听起来很简单的话题:定时任务机制。图片来自 Pexels无非就是一个计时器,到了指定时间就开始跑呗。too young,要是这么简单我还说啥呢,干...

探索动态执行的计划任务-DynamicSchedule

背景在现代软件开发中,计划任务是一种常见的需求。无论是定时发送邮件、定期清理缓存,还是执行数据同步,计划任务都能帮助我们自动化这些重复性工作。最近有一个需求,用户想要自己设定定时时间,来动态的执行定时...