DelayQueue之持久化方案

上一篇文章中提到了我们在项目中运用DelayQueue解决了一些需要延迟执行的任务，但是最近我们在生产环境上遇到了一个问题。重启服务器后，那些未执行的延迟任务就消失不见了。于是如何将延迟任务持久化就提上了日程。

关于DelayQueue的具体实现方案，已经在上一篇文章DelayQueue之通用组件中提到过了。本文就不再复述了。

本期的主题主要是探讨如何将延迟任务进行持久化。

何为延迟任务的持久化？顾名思义，就是将这些延迟任务的执行必要的数据，存储到数据库或redis等。

那么为何要进行持久化呢？很简单，因为延迟任务的数据是放在内存里,那么自己需要我们自己写持久化的备案以达到高可用，否则服务器故障宕机或新发布版本而导致服务器重启的时候，那么那些未执行的延迟任务数据将彻底丢失，这显然是我们不愿意见到的。

我目前采用的方案如下：
1、在需要使用到DelayQueue的地方，调用saveDelayTask方法，需要的参数有延迟任务函数策略工厂类的路由tag，执行方法所需的json格式的参数messageBody，延迟多久执行以秒为单位的delayTime。
2、任务调度每15秒去执行getNotCompletedMessageList方法。

大多数情况下，会在预计执行的时间点准时去执行processTask方法，那么异常状况下，如果服务器重启，那么定时任务调度会在一定时间后找到那些没有如期执行的延迟任务，通过定时任务调度的方式依次执行各自任务的processTask方法。
异常状态下，延迟任务执行会比预期执行时间有一定的延后，我设计的方案是目前我们可以允许的范围，这个大家可以酌情设置备选方案延后的时间。

核心代码如下，其他代码很简单就不一一公布了。

public void saveDelayTask(String tag, String messageBody, Long delayTime) {
    DelayTaskMessage delayTaskMessage = new DelayTaskMessage();
    delayTaskMessage.setTag(tag);
    LocalDateTime now = LocalDateTime.now();
    delayTaskMessage.setCreateTime(now);
    delayTaskMessage.setUpdateTime(now);
    delayTaskMessage.setDelayTime(delayTime);
    delayTaskMessage.setExpectedTime(now.plusSeconds(delayTime));
    delayTaskMessage.setMessageBody(messageBody);
    delayTaskMessage.setStatus(KafkaMessageStatusEnum.NOT_COMPLETE.getCode());
    int res = delayTaskMessageMapper.insertDelayTaskMessage(delayTaskMessage);
    if (res <= 0) {
        log.error("ybBrokerApp|insertDelayTaskMessage error, res<=0");
        throw new RuntimeException("insertDelayTaskMessage error, res<=0");
    }
    TaskMessage taskMessage = new TaskMessage(delayTime * 1000, messageBody,
            function -> this.processTask(delayTaskMessage));
    DelayQueue<TaskMessage> queue = taskManager.getQueue();
    queue.offer(taskMessage);
}

首先来分析一下，用来保存延迟任务的saveDelayTask方法。

tag是指延迟任务的标记，用于指定对应的策略类。

messageBody主要用于存储执行延迟任务的一些必要的数据，以json方法存储。

delayTime是延迟时间，默认以s为单位，主要是便于使用。

这个方法的主要功能是首先保存还未执行的延迟任务，自动根据延迟时间计算该延迟任务的预期执行时间，以便于后续的补偿算法跟踪，然后运用DelayQueue的特性，将这个延迟任务提交给延迟队列执行。

public int processTask(DelayTaskMessage param) {
    DelayTaskMessage delayTaskMessage = delayTaskMessageMapper.getDelayTaskMessageById(param.getId());
    try {
        if (null != delayTaskMessage && !Objects.equals(delayTaskMessage.getStatus(), KafkaMessageStatusEnum.NOT_COMPLETE.getCode())) {
            log.info("processTask executed already");
            return 1;
        }
        if (null != delayTaskMessage) {
            DelayTaskExecuteProcessor processor = processorFactory.getExecuteProcessor(delayTaskMessage.getTag());
            if (processor != null) {
                processor.execute(delayTaskMessage);
            } else {
                throw new RuntimeException("no such processor,tag=" + delayTaskMessage.getTag());
            }
            delayTaskMessage.setStatus(KafkaMessageStatusEnum.COMPLETE.getCode());
            delayTaskMessage.setExecutionTime(LocalDateTime.now());
            try {
                delayTaskMessage.setIpAddress(InetAddress.getLocalHost().getHostAddress());
            } catch (UnknownHostException ex) {
                log.error("Address.getLocalHost error", ex);
            }

            int res = delayTaskMessageMapper.updateDelayTaskMessageStatus(delayTaskMessage);
            if (res <= 0) {
                log.error("updateDelayTaskMessageStatus error res<=0");
                throw new RuntimeException("updateDelayTaskMessageStatus error");
            }
            return 1;
        } else {
            log.error("ybBrokerApp processTask error, delayTaskMessage is null delayTaskMessageId=", param.getId());
            return 0;
        }
    } catch (Exception e) {
        log.error("ybBrokerApp processTask error , param = " + param.toString() + "|", e);
        if (null != delayTaskMessage) {
            delayTaskMessage.setStatus(KafkaMessageStatusEnum.FAIL.getCode());
            delayTaskMessage.setErrorStack(e.getMessage());
            try {
                delayTaskMessage.setIpAddress(InetAddress.getLocalHost().getHostAddress());
            } catch (UnknownHostException ex) {
                log.error("Address.getLocalHost error", ex);
            }
            delayTaskMessageMapper.updateDelayTaskMessageStatus(delayTaskMessage);
        }
        return 0;
    }
}

然后是核心的处理延迟任务的processTask方法。

1、根据id,在数据库寻找到对应需要执行的延迟任务的持久化数据。
2、如果这条持久化数据非空且状态不是未执行的状态，那么提示该任务已经被执行过，防止重复执行。这里的status,主要有三种状态，未执行，已执行成功和执行失败。
3、如果这条持久化数据非空，且是未执行的状态，那么找到tag对应的策略类执行对应的execute方法。
4、将执行方法的ip地址记录下来，便于后续分析，同时将这条持久化数据的状态改为已执行成功的状态。
5、如果执行失败，将ip地址记录下来，同时把数据的状态改为执行失败。

在第4步中为何要先执行方法，后改状态呢？我是这么想的，延迟任务的执行方法是外部写的，在组件设计的时候无法控制会不会出现异常，而修改持久化数据的状态的方法是可控的，所以从这个角度上，我觉得先执行方法再修改状态更合理些。

需要注意的是，这个方法没有考虑并发的情况，是因为我在补偿方案里又额外延迟了一段时间，出现并发的情况非常小，不是很有必要考虑这种情况。

public List<DelayTaskMessage> getNotCompletedMessageList(int total, int index) {
        LocalDateTime expectedTime = LocalDateTime.now().plusSeconds(15L);
        List<DelayTaskMessage> delayTaskMessageList = delayTaskMessageMapper.getNotCompletedMessageList(expectedTime,total, index);
        if (CollectionUtils.isEmpty(delayTaskMessageList)) {
            return Lists.newArrayList();
        }
        return delayTaskMessageList;
    }

最后是补偿方案的落实，我是在定时任务中去保证延迟任务一定会被执行至少一次的。

我的设计是每隔15s去遍历一下那些过了预期执行时间+15s依然未执行的的延迟任务。然后将这些列表中的延迟任务重新调用processTask方法。

如果最终是通过补偿方案执行的延迟任务会比预期执行时间还要晚执行15到30s。目前在我们的项目中，这个额外延迟是可以被接收的。大家还是要根据实际情况酌情修改这个额外延迟的时间。

以上是我针对DelayQueue设计的的持久化方案，如果大家有更好的意见，可以一起讨论哦。