Redis 延时队列

zadd delayqueue 1614608094000 taskinfo

zrangebyscore delayqueue 0 timestamp limit 0 1000

@XxlScheduled(cron = "0/5 * * * * ?", name = "scan business1 delayqueue")
public void scanBusiness1() {
    // 某业务逻辑的zset延迟队列对应的key
    String zsetKey = "delayqueue:business1";
    while (true) {
        // 筛选score值小于当前时间戳的元素，一次最多拉取1000条
        Set<String> tasks = stringRedisTemplate.opsForZSet().rangeByScore(zsetKey, 0, System.currentTimeMillis(), 0, 1000);
        if (CollectionUtils.isEmpty(tasks)) {
            // 当前时间下已没有需要执行的任务，结束本次扫描
            return;
        }
        for (String task : tasks) {
            // 先删除，再执行，确保多线程环境下执行的唯一性
            Boolean delete = stringRedisTemplate.delete(task);
            if (delete) {
                // 删除成功后，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
                rabbitTemplate.convertAndSend("exchange_business1", "routekey_business1", task);
            }
        }
    }
}

@Scheduled(cron = "0/5 * * * * ?")
public void scanBusiness1() {
    // 队列数量M，考虑动态配置，保持和机器数量基本一致
    int M = 10;
    // redis自增key，用于负载均衡
    String incrKey = "incrkey:delayqueue:business1";
    // 每台机器执行时，从不同的zset中拉取消息，尽量确保不同机器访问不同zset
    String zsetKey = "delayqueue:business1:" + (stringRedisTemplate.opsForValue().increment(incrKey) % M);
    while (true) {
        // 此处逻辑和前面代码一致，省略。。。
    }
}

local zsetKey = 'delayqueue'
local timestamp = 1614608094000
local items = redis.call('zrangebyscore',zsetKey,0,timestamp,'limit',0,1)
if #items == 0 then
    return ''
else
    redis.call('zremrangebyrank',zsetKey,0,0)
    return items[1]
end

/**
 * 基于ZSET实现消息延迟处理，score存储执行时间点，到达时间点即会向指定队列发送该消息；
 * 定义一个继承本类的bean即可；
 */
public abstract class AbstractDelayedMsgScanTrigger implements Runnable, DisposableBean {

    private static final RedisScript<String> TRY_GET_AND_DEL_SCRIPT;
    static {
        // 获取并删除的lua脚本，使用spring提供的api
        String sb = "local items = redis.call('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1)\n" +
                "if #items == 0 then\n" +
                "\treturn ''\n" +
                "else\n" +
                "\tredis.call('zremrangebyrank',KEYS[1],0,0)\n" +
                "\treturn items[1]\n" +
                "end";
        // 自有工具类，只要能创建出spring包下的 RedisScript 的实现类对象均可
        TRY_GET_AND_DEL_SCRIPT = RedisScriptHelper.createScript(sb, String.class);
    }

    private final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(getThreadNum(), getThreadNum(),
            0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getThreadNamePrefix()));
    private volatile boolean quit = false;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void startScan() {
        // bean构建完成后，启动若干执行线程
        int threadNum = getThreadNum();
        for (int i = 0; i < threadNum; i++) {
            EXECUTOR.execute(this);
        }
    }

    @Override
    public void run() {
        while (!quit) {
            try {
                // 循环，采用lua获取当前需要执行的任务并将其从redis中删除
                String msg = stringRedisTemplate.execute(TRY_GET_AND_DEL_SCRIPT,
                        Lists.newArrayList(getDelayedMsgSourceKey()), String.valueOf(System.currentTimeMillis()));
                if (StringUtils.isNotBlank(msg)) {
                    // 消息不为空，表示获取任务成功，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
                    rabbitTemplate.convertAndSend(getSendExchange(), getSendRoutingKey(), msg);
                } else {
                    // 获取不到任务，表示当前时间点下暂无需要执行的任务，则线程休眠1S（可视情况调整）
                    SleepUtils.sleepSeconds(1);
                }
            } catch (Exception e) {
                Logs.MSG.error("delayed msg scan error, sourceKey:{}", getDelayedMsgSourceKey(), e);
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        quit = true;
    }

    public void setQuit(boolean quit) {
        this.quit = quit;
    }

    /**
     * 获取消息的工作线程数量
     */
    protected abstract int getThreadNum();

    /**
     * 线程名称前缀，方便问题定位
     */
    protected abstract String getThreadNamePrefix();

    /**
     * 存放延迟消息的ZSET队列名
     */
    protected abstract String getDelayedMsgSourceKey();

    /**
     * 消息到达执行时间点时将其通过指定 exchange 发送到实时消费队列中
     */
    protected abstract String getSendExchange();

    /**
     * 消息到达执行时间点时将其通过指定 routingKey 发送到实时消费队列中
     */
    protected abstract String getSendRoutingKey();

}