Delay - 如何用 Redis 打造一个延迟队列、广播（软件架构的设计）

文章1：设计概述
文章2：风险、问题、方案
通过前两篇文章的描述，将数据结构的使用和将会出现的问题及解决方案有所阐述，那么在本文中将会继续软件架构的设计，软件架构设计要满足设计原则，那么在这里就将从以下几个方面开始对架构的设计进行阐述：

1. 面向用户 1.1. 消息添加入口

public abstract class ProviderDelayJob {
    
    public abstract void addJob (String topic, Q job, boolean radio) throws DelayQueueException;
    
    public abstract Q delete (String topic, Long jobId) throws DelayQueueException;
}

此类为抽象类，实现类则是由延迟队列内部提供，本项目启动阶段会自动将所有依赖的 Spring bean 注入到系统 Spring 内，不需要用户手动添加，如果用户需要使用此类，直接在对应的业务类中是用 @Autowride 注入此抽象类即可。

1.2. 消费端 1.2.1 队列消费

public interface Callback {
    
    void delayListener (Q job, Channel channel);

    
    void finalRetry (String topic, Q job);
}


// 用户来继承本抽象类，实现 getTopic(), finalRetry(), delayListener(Q, Channel) 方法
public abstract class AbsDelayQueueListener implements Callback {

    
    public abstract String getTopic ();


    public AbsDelayQueueListener() {
        DelayQueueContextFactory.addListener(this);
    }

    @Override
    public void finalRetry(String topic, Q job) {
        log.warn("[self retry] 默认重试被调用：{}", JSON.toJSONString(job));
    }

    
    public QueueJob processorUnAck (String topic, Long jobId) {
        return DelayQueueContextFactory.processorUnAck(topic, jobId);
    }
}

用户使用到延迟队列中队列功能，那么就需要对添加的消息进行消费，要消费消息就必须继承 AbsDelayQueueListener 抽象类，并且实现 getTopic() ，finalRetry()，delayListener(Q, Channel) 三个抽象方法，当然 finalRetry() 方法是可以不用重写的，根据业务条件来。

1.2.2 广播消费

public abstract class AbsDelayRadioListener implements MessageListener {
    
    public abstract Set getChannels ();
    
    public abstract Class getClazz ();
}

该抽象类是广播的监听器，若开启了广播功能，且要使用广播功能，那么就必须继承本抽象类重写 getChannels () 与 getClazz () 与 onMessage(CharSequence, DelayQueueJob) 三个方法，且 getClazz () 是返回消息体的 class 对象，onMessage() 则是消费广播的方法，提供给用户处理业务逻辑。

2. 系统内架构设计

2.1 ContextFactory 上下文工厂

上下文工厂，顾名思义，是来创建上下文的一个工厂，是一个静态工厂，该工厂主要负责

类中所有监听器的扫描和注册（将监听器与线程做绑定关系）

 private void loadListeners() {
    // 每一个启动一个线程去执行
    Set>> entries = LISTENER_HOLDER.entrySet();
    for (Map.Entry> entry : entries) {
        if (entry.getKey().length() > KEY_MAX_LENGTH) {
            this.shutdown();
            this.queueThreadPoolFactory.getListenersThreadPool().shutdown();
            throw new DelayQueueException("TOPIC [" + entry.getKey() + "] 长度超过 " + KEY_MAX_LENGTH + " 的限制!");
        }
        // 每循环一次，将创建一个线程调用 doLoadListener() 
        this.queueThreadPoolFactory.getListenersThreadPool()
        		// 见 2.2.1 `ListenerContext` 监听上下文代码块
                .execute(() -> this.listenerContext.doLoadListener(entry));
    }
}

初始化所有的子 context ，包含 ListenerContext，TransferContext，RetryContext，AckContext，CrashContext 不同的 context 所做的事情不同，职责划分明确。

2.1.1 ListenerContext 监听上下文

系统中任何实现了 AbsSelayQueueListener 的接口都被成为监听器，想要所有的监听器都能消费自己监听的 TOPIC 的消息，那么都要被统一管理起来，后续有消息触发消费的时候，可以准确的找到监听器的实现类，从而调用 delayListener 方法进行业务处理。

// ContextFactory 调用， 2.1 进入
public  void doLoadListener(Map.Entry> entry) {
    final String self = Constant.colon.concat(entry.getKey());
    final String transferLockKey = RedisKeyUtil.getLockKey(this.properties.getAppName(), RedisKeyUtil.IN_TRANSIT_KEY).concat(self);
    final String readyKey = RedisKeyUtil.getReadyKey(properties, entry.getKey());
    
    while (shutdown) {
        // 从下层 QueueOperation 从读取超时的消息
        Set selfTopics = this.delayQueueOperation.distributedSafeReadReadySelfJob(transferLockKey, readyKey);
        if (CollectionUtils.isEmpty(selfTopics)) {
        	// 任何一个线程进入后，没有消费的 TOPIC 的时候就进入睡眠
            TransferListenSynchronizer.await(entry.getKey());
            continue;
        }
        // 有消息，则调用本类中的处理方法， 代码如下
        this.processSelfMetadata(selfTopics, entry.getValue());
    }
    log.info("[listen] {} 线程结束延迟队列的监听.....", Thread.currentThread());
}

处理自己 TOPIC 超时的消息

private  void processSelfMetadata(Set selfTopics,
														  // 一直持有实现类的引用，从而可以轻易调用实现方法
                                                          AbsDelayQueueListener self) {
    Iterator iterator = selfTopics.iterator();
    while (iterator.hasNext()){
        // 宕机等待善后，不往下走
        if (!shutdown) {
            break;
        }
        String selfTopic = iterator.next();
        // 读取元数据 metadata
        Q job = (Q) this.delayQueueOperation.readQueueMetadataJob(selfTopic);
        if (Objects.isNull(job)) {
        	// 读不到数据，说明有误，直接删除错误的数据
            this.delayQueueOperation.delNotFountMetadata(selfTopic, self.getTopic());
            continue;
        }
        // 深克隆，防止用户在失败的方法中操作内容覆盖掉正确的内容
        Q deep = (Q) new DeepCopyUtils().copy(job);
        try {
            // 创建 ack rejoin 标识对象
            Channel channel = new Channel(QueueAckContext.getAckTag());
            // 执行用户回调
			this.distributedSafeTimeoutExecuteCallbackMethod(self, channel, job);
            if (log.isDebugEnabled()) {
                log.debug("[listen] topic:{} 消费了消息：{}", selfTopic, JSON.toJSONString(job));
            }
            // 处理成功后处理 ack 和 rejoin 逻辑
            this.ackContext.processAck(self, deep, job, selfTopic);
            iterator.remove();
        } catch (Exception ex) {
            log.error("[listen] 回调业务执行异常：{}", ex.getMessage(), ex);
            // 上面调用业务失败后，直接调用重试机制
            this.retryContext.retry(selfTopic, deep, self, job.getRetryInterval());
        }
    }
}

回调继承 AbsDelayQueueListener#delayListener 的方法

private  void distributedSafeTimeoutExecuteCallbackMethod(AbsDelayQueueListener self,
                                                                                     final Channel channel, final Q job) throws Exception {
    // ack 标识上推
    final ThreadLocal listenThread = QueueAckContext.getAckTag();
    final Duration waitTime = this.properties.getConsumeWaitTime();
    final long lockWaitTime = this.properties.getLockWaitTime();

    // 等待时间为 0 的时候, 不走超时
    if (waitTime.getSeconds() <= 0) {
        try {
            self.delayListener(job, channel);
        } catch (Exception ex) {
            log.error("[listen] 业务执行失败, 异常:{}", ex.getMessage(), ex);
            throw ex;
        }
    }

    // 设置超时执行的时间走超时逻辑
    try {
        FutureTask ft = new FutureTask<>(() -> {
            self.delayListener(job, channel);
            // 返回被执行线程所拥有的状态码,放入主线程中
            return QueueAckContext.removeAck();
        });
        // 异步执行
        this.queueThreadPoolFactory.getTimeoutThreadPool().execute(ft);
        // 超时线程 ack 标识指向监听线程
        listenThread.set(ft.get(waitTime.getSeconds(), TimeUnit.SECONDS));
    } catch (Exception ex) {
        log.error("[listen] 超时业务执行失败, 异常:{}", ex.getMessage(), ex);
        throw ex;
    }
}

2.1.2 TransferContext 搬运上下文

结合第一篇文档的数据结构剖析中，所谓的搬运，就是将消息 zset 中超时的数据，搬运到不同机器所指向的 set 队列内

private void transfer() {
    // {redis-delay-queue}:timeout_table
    final String timeoutKey = RedisKeyUtil.getTopKey(this.properties.getAppName(), RedisKeyUtil.SET_KEY_PREFIX);
    //  {redis-delay-queue}:ready_table
    final String readyKey = RedisKeyUtil.getTopKey(this.properties.getAppName(), RedisKeyUtil.READY_KEY_PREFIX);
    
    this.queueThreadPoolFactory.getTransferThreadPool().execute(() -> {
        while (shutdown) {
            try {
            	// 搬运完毕一次后，扫描以下还有多少剩余未搬运的，通过剩余的来计算睡眠的时间
                long remain = this.delayQueueOperation.distributedSafeTransfer(readyKey, timeoutKey);
                // 智能睡眠
                SmartSleepUtil.sleep(remain);
            } catch (Exception ex) {
                log.error("[transfer] 搬运发生异常, ex: {}", ex.getMessage(), ex);
            }
        }
        log.info("[transfer] 停止.........");
    });
}

2.1.3 RetryContext 重试上下文

重试，当业务处理失败后会进入到 RetryContext 中处理重试，在 ListenerContext 中调用 distributedSafeTimeoutExecuteCallbackMethod 方法失败的情况，会进入重试，重试代码如下：

public  void retry(String selfTopic, Q deep, AbsDelayQueueListener self, int retryInterval) {

    // 重试间隔时间不允许小于 0
    if (retryInterval <= 0) {
        log.warn("[retry] 重试间隔时间 {} 小于或等于 0，取默认值：{}", retryInterval, deep.getRetryInterval());
        retryInterval = deep.getRetryInterval();
    }

    if (deep.getRetry() >= (this.properties.getRetry())) {
        log.warn("[retry] TOPIC: {} 已经被重试 {} 次，不再重试", self.getTopic(), deep.getRetry());
        // 删除消息 ready metadata un_transfer
        this.delayQueueOperation.retryFailDelete(self.getTopic(), deep.getJobId());
        // 调用用户提供的最终通知
        self.finalRetry(self.getTopic(), deep);
        return;
    }

    // 设置重试次数 + 1， 并保存到 metadata 中
    deep.setRetry(deep.getRetry() + 1);
    boolean updateJob = this.delayQueueOperation.updateMetadataJob(selfTopic, deep);
    if (updateJob) {
        log.info("[retry] topic：{} 重试次数：{}，metadata：{}", selfTopic, deep.getRetry(), JSON.toJSONString(deep));
        deep.setDelayTime(System.currentTimeMillis() + retryInterval);
        this.delayQueueOperation.consumeFailRollback(self.getTopic(), deep);
    }
}

2.1.4 AckContext 消息确认上下文

消息确认机制，保证消息能成功被消费，如果没有 ack，那么消息将会进入 un_ack 队列中进行保存，处理逻辑如下：

public void processAck (AbsDelayQueueListener self, final Q deep, final Q job, final String selfTopic) {
    final int tag = QueueAckContext.removeAck();
    final boolean ack = (tag & 2) == 2, rejoin = (tag & 4) == 4;
    // 啥也没处理, 需要 un_ack
    if (ack) {
        // ack 后, 删除元数据 与 消费队列
        this.delayQueueOperation.postConsumerDelete(self.getTopic(), deep.getJobId());
        if (log.isDebugEnabled()) {
            log.debug("[ack] TOPIC: {} 业务回调成功, JOB: {}", self.getTopic(), deep.getJobId());
        }
    } else {
        // 只做了 ack, 不需要重入, 才 put un ack
        if (!rejoin) {
            log.warn("[ack] 消息未被 ack， topic: {}, JOB:{}", self.getTopic(), JSON.toJSONString(deep));
            // 从待消费的队列移动到未确认的队列
            this.delayQueueOperation.putUnAckJob(selfTopic, self.getTopic(), deep);
        }
    }
    // 需要重新入队列
    if (rejoin) {
        job.setRejoin(true);
        // 如果用户设置了大于当前时间, 就使用用户的, 如果没有, 就使用自己的
        if (job.getDelayTime() < System.currentTimeMillis()) {
            job.setDelayTime(Constant.expire);
        }
        String newKey = RedisKeyUtil.getSecondKey(self.getTopic(), job.getJobId());
        if (!Objects.equals(newKey, selfTopic)) {
            throw new DelayQueueException("[rejoin] 请勿修改jobId后重新入队，原始TOPIC:"
                    + selfTopic + ", 当前TOPIC：" + newKey);
        }
        // 添加到主队列，尝试分布式重试
        this.delayQueueOperation.addJob(self.getTopic(), job);
        // 删除 un ack
        this.delayQueueOperation.deleteUnAckJob(self.getTopic(), job.getJobId());
        log.info("[rejoin] 消息重新入队, 等待过期消费：topic：{}， JOB：{}", selfTopic, JSON.toJSONString(job));
        // 重入队列后，需要立马唤醒搬运线程检查消息的过期情况
        SmartSleepUtil.immediatelyArouse();
    }
}

Channel 对象

public class Channel {

    
    private final ThreadLocal ackTag;

    public Channel(ThreadLocal ackTag) {
        this.ackTag = ackTag;
    }

    public void ack () {
        this.ack(true);
    }

    
    // 1 << 1           0000 0010   2
    // 1 << 1 | 1 << 2  0000 0110   6
    public void ack (boolean ack) {
        if (!ack) {
            return;
        }
        ackTag.set((ackTag.get() == 1 << 2) ? (1 << 2 | 1 << 1) : (1 << 1));
    }

    @SuppressWarnings("all")
    public void unack () {
        ack(false);
    }


    public void rejoin() {
        rejoin(true);
    }

    
    // 1 << 2           0000 0100   4
    // 1 << 2 | 1 << 1  0000 0110   6
    public void rejoin (boolean rejoin) {
        if (!rejoin) {
            return;
        }
        ackTag.set((ackTag.get() == 1 << 1) ? (1 << 1 | 1 << 2) : (1 << 2));
    }
}

2.1.5 CrashContext 关机善后处理上下文

在文章《风险、问题、方案》中，提及到单机宕机后消息如何回滚的问题，那么这里就通过 CrashContext 完成的，实现原理主要是依赖于 spring boot 中关机前发布的 ContextCloseEvent 事件，监听这个事件来对自己未消费的小重新放入到 Zset 集合中

public class CrashHandlerContext implements ApplicationListener {

    private final DelayQueueContextFactory factory;
    private final RedisDelayQueueProperties properties;

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        this.factory.shutdown();
        if (properties.getEnableCrash()) {
            factory.crashRollback();
        }
    }
}

public void crashRollback () {
    // 处理所有消费者所在的队列
    Set listeners = LISTENER_HOLDER.keySet();
    this.delayQueueOperation.listenerAftermathRestore(listeners);
}

3. 总结：

通过对各大核心上下文的源码展示，相信读者心中已经有了一定的思路了，如果有兴趣的话，也可以自己写一个，由于此项目是贡献给公司的，所以很遗憾不能开源。

Delay - 如何用 Redis 打造一个延迟队列、广播（软件架构的设计）

Java相关栏目本月热门文章