关于监控线程任务的超时管理，利用DelayQueue堵塞队列判断是否子任务线程是否超时。

淡淡的烟草味﹌ 2023-10-10 14:41 5阅读 0赞

首先DelayQueue队列的元素必须是Delayed接口的元素，所以需要创建一个实现Delayed接口的类对象。其中必须重写

getDelay方法与compareTo方法，因为DelayQueue队列就是通过其getDelay方法去判断该元素是否超时

FutureTaskHolder类

public class FutureTaskHolder implements Delayed {
    
        private final Integer deploymentTaskId;
    
        private final DeploymentTaskLogger taskLogger;
    
        private final FutureTask<Void> futureTask;
    
        private final int timeoutThreshold;
    
        private final long endTime;
    
        /**
         * @param deploymentTaskId    部署任务id
         * @param futureTask       future任务
         * @param timeoutThreshold 超时阈值，单位min
         * @param taskLogger      日志记录器
         */
        public FutureTaskHolder(Integer deploymentTaskId, FutureTask<Void> futureTask, int timeoutThreshold, DeploymentTaskLogger taskLogger) {
            this.deploymentTaskId = deploymentTaskId;
            this.futureTask = futureTask;
            this.timeoutThreshold = timeoutThreshold;
            this.taskLogger = taskLogger;
            endTime = System.currentTimeMillis() + (long) timeoutThreshold * 1000 * 60;
        }
    
        @Override
        public long getDelay(TimeUnit unit) {
            //System.out.println("单位" + unit.toString());
            //return unit.convert(diff, unit);
            return endTime - System.currentTimeMillis();
        }
    
        @Override
        public int compareTo(Delayed o) {
            FutureTaskHolder futureTaskHolder = (FutureTaskHolder) o;
            long diff = this.endTime - futureTaskHolder.endTime;
            return diff <= 0 ? -1 : 1;
        }
    
        public int getTimeoutThreshold() {
            return timeoutThreshold;
        }
    
        public FutureTask<Void> getFutureTask() {
            return futureTask;
        }
    
        public Integer getDeploymentTaskId() {
            return deploymentTaskId;
        }
    
        public DeploymentTaskLogger getTaskLogger() {
            return taskLogger;
        }
    }

监控管理工具类：

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.FutureTask;
    
    @Component
    public class FutureTaskManager {
    
    
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
        private final DelayQueue<FutureTaskHolder> delayQueue = new DelayQueue<>();
        @Autowired
        private ChkDeploymentManager deploymentTaskManager;
    
        public void addFutureTaskHolder(FutureTaskHolder futureTaskHolder){
            delayQueue.add(futureTaskHolder);
        }
    
        @PostConstruct
        public void init(){
            //开启线程，从延迟队列中取CompleteFutureHolder，如果超时还未完成，则设置部署任务状态为失败
            new Thread(() -> {
                while (true){
                    try {
                        FutureTaskHolder futureTaskHolder = delayQueue.take();
                        FutureTask<Void> futureTask = futureTaskHolder.getFutureTask();
                        if (!futureTask.isDone()){
                            logger.warn("部署任务超时,deploymentTaskId:{}", futureTaskHolder.getDeploymentTaskId());
                            DeploymentTaskLogger taskLogger = futureTaskHolder.getTaskLogger();
                            taskLogger.log("部署任务超时");
                            deploymentTaskManager.updateDeploymentTaskStatus(futureTaskHolder.getDeploymentTaskId(), DeploymentTaskStatus.FAIL, taskLogger);
                            /// TODO: 2022/7/21 如何取消正在运行的任务
                            futureTask.cancel(true);
                        }
                    } catch (InterruptedException e) {
                        logger.warn("取消超时任务失败,msg:{}", e.getMessage());
                        throw new RuntimeException(e);
                    }
                }
            }).start();
        }
    
    
    }

使用

//加入manager，监控是否超过部署时间
     completeFutureManager.addFutureTaskHolder(new FutureTaskHolder(id, futureTask, deploymentTemplate.getDeploymentTimeOut(), taskLogger));

当任务从堵塞队列中出来后还未完成的话，则提示任务超时，且进行数据库更新，因为该线程是守护线程，所以可以一直判断。