详解基于Redis 延时队列的优化方案

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();

}