PowerJob的HashedWheelTimer工作流程源码解读

更新时间：2024年01月23日 09:06:26 作者：codecraft

这篇文章主要为大家介绍了PowerJob的HashedWheelTimer工作流程源码解读,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

序

本文主要研究一下PowerJob的HashedWheelTimer

Timer

tech/powerjob/server/common/timewheel/Timer.java

public interface Timer {
    /**
     * 调度定时任务
     */
    TimerFuture schedule(TimerTask task, long delay, TimeUnit unit);
    /**
     * 停止所有调度任务
     */
    Set<TimerTask> stop();
}

Timer接口定义了schedule方法，用于在指定时间之后调度TimerTask，它返回TimerFuture；stop方法返回未处理的TimerTask

TimerTask

@FunctionalInterface
public interface TimerTask extends Runnable {
}

TimerTask继承了Runnable

TimerFuture

tech/powerjob/server/common/timewheel/TimerFuture.java

public interface TimerFuture {

    TimerTask getTask();

    boolean cancel();

    boolean isCancelled();

    boolean isDone();
}

TimerFuture定于了getTask、cancel、isCancelled、isDone方法

HashedWheelTimer

tech/powerjob/server/common/timewheel/HashedWheelTimer.java

@Slf4j
public class HashedWheelTimer implements Timer {
    private final long tickDuration;
    private final HashedWheelBucket[] wheel;
    private final int mask;
    private final Indicator indicator;
    private final long startTime;
    private final Queue<HashedWheelTimerFuture> waitingTasks = Queues.newLinkedBlockingQueue();
    private final Queue<HashedWheelTimerFuture> canceledTasks = Queues.newLinkedBlockingQueue();
    private final ExecutorService taskProcessPool;
    public HashedWheelTimer(long tickDuration, int ticksPerWheel) {
        this(tickDuration, ticksPerWheel, 0);
    }
    /**
     * 新建时间轮定时器
     * @param tickDuration 时间间隔，单位毫秒（ms）
     * @param ticksPerWheel 轮盘个数
     * @param processThreadNum 处理任务的线程个数，0代表不启用新线程（如果定时任务需要耗时操作，请启用线程池）
     */
    public HashedWheelTimer(long tickDuration, int ticksPerWheel, int processThreadNum) {
        this.tickDuration = tickDuration;
        // 初始化轮盘，大小格式化为2的N次，可以使用 & 代替取余
        int ticksNum = CommonUtils.formatSize(ticksPerWheel);
        wheel = new HashedWheelBucket[ticksNum];
        for (int i = 0; i < ticksNum; i++) {
            wheel[i] = new HashedWheelBucket();
        }
        mask = wheel.length - 1;
        // 初始化执行线程池
        if (processThreadNum <= 0) {
            taskProcessPool = null;
        }else {
            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("HashedWheelTimer-Executor-%d").build();
            // 这里需要调整一下队列大小
            BlockingQueue<Runnable> queue = Queues.newLinkedBlockingQueue(8192);
            int core = Math.max(Runtime.getRuntime().availableProcessors(), processThreadNum);
            // 基本都是 io 密集型任务
            taskProcessPool = new ThreadPoolExecutor(core, 2 * core,
                    60, TimeUnit.SECONDS,
                    queue, threadFactory, RejectedExecutionHandlerFactory.newCallerRun("PowerJobTimeWheelPool"));
        }
        startTime = System.currentTimeMillis();
        // 启动后台线程
        indicator = new Indicator();
        new Thread(indicator, "HashedWheelTimer-Indicator").start();
    }
    //......
}

HashedWheelTimer实现了Timer接口，其构造器要求输入tickDuration、ticksPerWheel，它会将ticksPerWheel转换为2的N次，然后创建对应的HashedWheelBucket，若processThreadNum大于0则同时创建ThreadPoolExecutor用于处理任务，最后启动异步线程执行Indicator

schedule

@Override
    public TimerFuture schedule(TimerTask task, long delay, TimeUnit unit) {
        long targetTime = System.currentTimeMillis() + unit.toMillis(delay);
        HashedWheelTimerFuture timerFuture = new HashedWheelTimerFuture(task, targetTime);
        // 直接运行到期、过期任务
        if (delay <= 0) {
            runTask(timerFuture);
            return timerFuture;
        }
        // 写入阻塞队列，保证并发安全（性能进一步优化可以考虑 Netty 的 Multi-Producer-Single-Consumer队列）
        waitingTasks.add(timerFuture);
        return timerFuture;
    }

schedule方法先计算目标时间，然后创建对应的HashedWheelTimerFuture，若delay小于等于0则执行runTask，否则添加到waitingTasks

stop

@Override
    public Set<TimerTask> stop() {
        indicator.stop.set(true);
        taskProcessPool.shutdown();
        while (!taskProcessPool.isTerminated()) {
            try {
                Thread.sleep(100);
            }catch (Exception ignore) {
            }
        }
        return indicator.getUnprocessedTasks();
    }

stop方法先设置indicator的stop为true，然后执行taskProcessPool.shutdown()，等待关闭，最后返回indicator.getUnprocessedTasks()

HashedWheelBucket

private final class HashedWheelBucket extends LinkedList<HashedWheelTimerFuture> {
        public void expireTimerTasks(long currentTick) {
            removeIf(timerFuture -> {
                // processCanceledTasks 后外部操作取消任务会导致 BUCKET 中仍存在 CANCELED 任务的情况
                if (timerFuture.status == HashedWheelTimerFuture.CANCELED) {
                    return true;
                }
                if (timerFuture.status != HashedWheelTimerFuture.WAITING) {
                    log.warn("[HashedWheelTimer] impossible, please fix the bug");
                    return true;
                }
                // 本轮直接调度
                if (timerFuture.totalTicks <= currentTick) {
                    if (timerFuture.totalTicks < currentTick) {
                        log.warn("[HashedWheelTimer] timerFuture.totalTicks < currentTick, please fix the bug");
                    }
                    try {
                        // 提交执行
                        runTask(timerFuture);
                    }catch (Exception ignore) {
                    } finally {
                        timerFuture.status = HashedWheelTimerFuture.FINISHED;
                    }
                    return true;
                }
                return false;
            });
        }
    }
    private void runTask(HashedWheelTimerFuture timerFuture) {
        timerFuture.status = HashedWheelTimerFuture.RUNNING;
        if (taskProcessPool == null) {
            timerFuture.timerTask.run();
        }else {
            taskProcessPool.submit(timerFuture.timerTask);
        }
    }

HashedWheelBucket继承了LinkedList，其泛型为HashedWheelTimerFuture，它提供了expireTimerTasks方法，通过removeIf删除status为CANCELED、status不为WAITING，以及执行runTask(注意这里忽略了异常)之后标记status为FINISHED的元素；runTask先标记为RUNNING，对于taskProcessPool为null则直接执行，否则提交到taskProcessPool

HashedWheelTimerFuture

tech/powerjob/server/common/timewheel/HashedWheelTimer.java

private final class HashedWheelTimerFuture implements TimerFuture {
        // 预期执行时间
        private final long targetTime;
        private final TimerTask timerTask;
        // 所属的时间格，用于快速删除该任务
        private HashedWheelBucket bucket;
        // 总圈数
        private long totalTicks;
        // 当前状态 0 - 初始化等待中，1 - 运行中，2 - 完成，3 - 已取消
        private int status;
        // 状态枚举值
        private static final int WAITING = 0;
        private static final int RUNNING = 1;
        private static final int FINISHED = 2;
        private static final int CANCELED = 3;
        public HashedWheelTimerFuture(TimerTask timerTask, long targetTime) {
            this.targetTime = targetTime;
            this.timerTask = timerTask;
            this.status = WAITING;
        }
        @Override
        public TimerTask getTask() {
            return timerTask;
        }
        @Override
        public boolean cancel() {
            if (status == WAITING) {
                status = CANCELED;
                canceledTasks.add(this);
                return true;
            }
            return false;
        }
        @Override
        public boolean isCancelled() {
            return status == CANCELED;
        }
        @Override
        public boolean isDone() {
            return status == FINISHED;
        }
    }

HashedWheelTimerFuture实现了TimerFuture接口，它定义了WAITING、RUNNING、FINISHED、CANCELED状态；初始状态为WAITING，对于WAITING状态的可以设置为CANCELED，并添加到canceledTasks；isCancelled判断状态是不是CANCELED，isDone判断状态是不是FINISHED

getUnprocessedTasks

public Set<TimerTask> getUnprocessedTasks() {
            try {
                latch.await();
            }catch (Exception ignore) {
            }
            Set<TimerTask> tasks = Sets.newHashSet();
            Consumer<HashedWheelTimerFuture> consumer = timerFuture -> {
                if (timerFuture.status == HashedWheelTimerFuture.WAITING) {
                    tasks.add(timerFuture.timerTask);
                }
            };
            waitingTasks.forEach(consumer);
            for (HashedWheelBucket bucket : wheel) {
                bucket.forEach(consumer);
            }
            return tasks;
        }

getUnprocessedTasks会等待Indicator的while循环结束，然后遍历所有的HashedWheelBucket找出状态还是WAITING的任务

Indicator

private class Indicator implements Runnable {
        private long tick = 0;
        private final AtomicBoolean stop = new AtomicBoolean(false);
        private final CountDownLatch latch = new CountDownLatch(1);
        @Override
        public void run() {
            while (!stop.get()) {
                // 1. 将任务从队列推入时间轮
                pushTaskToBucket();
                // 2. 处理取消的任务
                processCanceledTasks();
                // 3. 等待指针跳向下一刻
                tickTack();
                // 4. 执行定时任务
                int currentIndex = (int) (tick & mask);
                HashedWheelBucket bucket = wheel[currentIndex];
                bucket.expireTimerTasks(tick);
                tick ++;
            }
            latch.countDown();
        }
        /**
         * 模拟指针转动，当返回时指针已经转到了下一个刻度
         */
        private void tickTack() {
            // 下一次调度的绝对时间
            long nextTime = startTime + (tick + 1) * tickDuration;
            long sleepTime = nextTime - System.currentTimeMillis();
            if (sleepTime > 0) {
                try {
                    Thread.sleep(sleepTime);
                }catch (Exception ignore) {
                }
            }
        }
        /**
         * 处理被取消的任务
         */
        private void processCanceledTasks() {
            while (true) {
                HashedWheelTimerFuture canceledTask = canceledTasks.poll();
                if (canceledTask == null) {
                    return;
                }
                // 从链表中删除该任务（bucket为null说明还没被正式推入时间格中，不需要处理）
                if (canceledTask.bucket != null) {
                    canceledTask.bucket.remove(canceledTask);
                }
            }
        }
        /**
         * 将队列中的任务推入时间轮中
         */
        private void pushTaskToBucket() {
            while (true) {
                HashedWheelTimerFuture timerTask = waitingTasks.poll();
                if (timerTask == null) {
                    return;
                }
                // 总共的偏移量
                long offset = timerTask.targetTime - startTime;
                // 总共需要走的指针步数
                timerTask.totalTicks = offset / tickDuration;
                // 取余计算 bucket index
                int index = (int) (timerTask.totalTicks & mask);
                HashedWheelBucket bucket = wheel[index];
                // TimerTask 维护 Bucket 引用，用于删除该任务
                timerTask.bucket = bucket;
                if (timerTask.status == HashedWheelTimerFuture.WAITING) {
                    bucket.add(timerTask);
                }
            }
        }
        public Set<TimerTask> getUnprocessedTasks() {
            try {
                latch.await();
            }catch (Exception ignore) {
            }
            Set<TimerTask> tasks = Sets.newHashSet();
            Consumer<HashedWheelTimerFuture> consumer = timerFuture -> {
                if (timerFuture.status == HashedWheelTimerFuture.WAITING) {
                    tasks.add(timerFuture.timerTask);
                }
            };
            waitingTasks.forEach(consumer);
            for (HashedWheelBucket bucket : wheel) {
                bucket.forEach(consumer);
            }
            return tasks;
        }
    }

Indicator实现了Runnable接口，其run方法在stop为false的时候循环执行，pushTaskToBucket、processCanceledTasks、tickTack、expireTimerTasks

pushTaskToBucket

private void pushTaskToBucket() {
            while (true) {
                HashedWheelTimerFuture timerTask = waitingTasks.poll();
                if (timerTask == null) {
                    return;
                }
                // 总共的偏移量
                long offset = timerTask.targetTime - startTime;
                // 总共需要走的指针步数
                timerTask.totalTicks = offset / tickDuration;
                // 取余计算 bucket index
                int index = (int) (timerTask.totalTicks & mask);
                HashedWheelBucket bucket = wheel[index];
                // TimerTask 维护 Bucket 引用，用于删除该任务
                timerTask.bucket = bucket;
                if (timerTask.status == HashedWheelTimerFuture.WAITING) {
                    bucket.add(timerTask);
                }
            }
        }

pushTaskToBucket通过waitingTasks.poll()拉取任务，若为null直接返回，否则通过timerTask.targetTime与startTime计算offset，再根据tickDuration计算需要走的步数，然后计算并获取目标HashedWheelBucket，然后将timerTask添加到bucket中

processCanceledTasks

private void processCanceledTasks() {
            while (true) {
                HashedWheelTimerFuture canceledTask = canceledTasks.poll();
                if (canceledTask == null) {
                    return;
                }
                // 从链表中删除该任务（bucket为null说明还没被正式推入时间格中，不需要处理）
                if (canceledTask.bucket != null) {
                    canceledTask.bucket.remove(canceledTask);
                }
            }
        }

processCanceledTasks会执行canceledTasks.poll()拉取canceledTask，若canceledTask.bucket不为null则将canceledTask从该bucket中移除

tickTack

private void tickTack() {
            // 下一次调度的绝对时间
            long nextTime = startTime + (tick + 1) * tickDuration;
            long sleepTime = nextTime - System.currentTimeMillis();
            if (sleepTime > 0) {
                try {
                    Thread.sleep(sleepTime);
                }catch (Exception ignore) {
                }
            }
        }

tickTack模拟指针移动，它先计算nextTime，再计算需要sleep多久，然后执行Thread.sleep(sleepTime)

小结

PowerJob定义了Timer接口，并提供了HashedWheelTimer的实现，它定义了waitingTasks、canceledTasks两个LinkedBlockingQueue(无界队列)，同时还支持定义任务处理线程池的core线程数；它通过Indicator线程来处理时间轮的转动及任务处理，Indicator循环将waitingTasks的任务放入到对应的bucket，然后模拟时间轮等待，然后通过bucket.expireTimerTasks(tick)处理到期任务，最后再递增tick。

以上就是PowerJob的HashedWheelTimer工作流程源码解读的详细内容，更多关于PowerJob HashedWheelTimer的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

通过pipeline配置sonar自动化实现过程解析
这篇文章主要介绍了通过pipeline配置sonar自动化实现过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-11-11
基于OpenCv与JVM实现加载保存图像功能(JAVA 图像处理)
openCv有一个名imread的简单函数，用于从文件中读取图像，本文给大家介绍JAVA 图像处理基于OpenCv与JVM实现加载保存图像功能，感兴趣的朋友一起看看吧
2022-01-01
finally 一定会执行(实例代码)
下面小编就为大家带来一篇finally 一定会执行(实例代码)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-07-07
SpringBoot修改子模块Module的jdk版本的方法附修改原因
这篇文章主要介绍了SpringBoot修改子模块Module的jdk版本的方法附修改原因,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-04-04
RabbitMQ下载与安装教程
RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件),这篇文章给大家介绍RabbitMQ下载与安装教程,感兴趣的朋友跟随小编一起看看吧
2024-02-02
浅谈Java线程间通信之wait/notify
下面小编就为大家带来一篇浅谈Java线程间通信之wait/notify。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-06-06
Java微信授权登录小程序接口流程
微信授权登录小程序的流程是一个涉及前端和后端交互的过程,主要目的是让用户能够使用微信账号快速登录小程序,避免重复输入用户名和密码,下面给大家介绍Java微信授权登录小程序接口流程,感兴趣的朋友跟随小编一起看看吧
2024-08-08
解析Java中的默认方法
这篇文章主要介绍了Java中的默认方法,包括继承和调用等Java入门学习中的基础知识,需要的朋友可以参考下
2015-07-07
Springboot处理CORS跨域请求的三种方法
这篇文章主要介绍了Springboot处理CORS跨域请求的三种方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-06-06
在IDEA中maven配置MyBatis的流程详解
刚学完javaweb,对自己的Dao层代码很不满意的话,可得来学学MyBatis.学习MyBatis既可以改进JDBC的使用,实现Dao层也会变得很简便,下面我将介绍IDEA中maven配置MyBatis简单流程,需要的朋友可以参考下
2021-06-06