江苏龙网

  1. 主页 > 生活百科 > >

TimingWheel 时间轮算法是如何实现的？( 二 )

时间轮算法

在这个构造器中有几个细节需要注意：

  1. 调用createWheel方法创建的wheel数组一定是2次方数，比如传入的ticksPerWheel是6，那么初始化的wheel长度一定是8 。这样做是为了让mask & tick 来计算出槽位
  2. tickDuration用的是纳秒
  3. 在构造里面并不会里面启动时间轮，而是要等到有第一个任务加入到时间轮的时候才启动 。在构造器里面会将工作线程worker封装成workerThread
放入任务到时间轮中public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {if (task == null) {throw new NullPointerException("task");}if (unit == null) {throw new NullPointerException("unit");}long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {pendingTimeouts.decrementAndGet();throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount+ ") is greater than or equal to maximum allowed pending "+ "timeouts (" + maxPendingTimeouts + ")");}// 如果时间轮没有启动，则启动start();// Add the timeout to the timeout queue which will be processed on the next tick.// During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;// Guard against overflow.//在delay为正数的情况下，deadline是不可能为负数//如果为负数，那么说明超过了long的最大值if (delay > 0 && deadline < 0) {deadline = Long.MAX_VALUE;}// 这里定时任务不是直接加到对应的格子中，而是先加入到一个队列里，然后等到下一个tick的时候，// 会从队列里取出最多100000个任务加入到指定的格子中HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);//Worker会去处理timeouts队列里面的数据timeouts.add(timeout);return timeout;}
  1. 如果时间轮没有启动，那么就调用start方法启动时间轮，启动时间轮之后会为startTime设置为当前时间
  2. 计算延迟时间deadline
  3. 将task任务封装到HashedWheelTimeout中，然后添加到timeouts队列中进行缓存
startprivate final CountDownLatchstartTimeInitialized= new CountDownLatch(1);public void start() {//workerState一开始的时候是0（WORKER_STATE_INIT），然后才会设置为1（WORKER_STATE_STARTED）switch (workerStateUpdater.get(this)) {case WORKER_STATE_INIT://使用cas来获取启动调度的权力，只有竞争到的线程允许来进行实例启动if (workerStateUpdater.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {//如果成功设置了workerState，那么就调用workerThread线程workerThread.start();}break;case WORKER_STATE_STARTED:break;case WORKER_STATE_SHUTDOWN:throw new IllegalStateException("cannot be started once stopped");default:throw new Error("Invalid WorkerState");}// 等待worker线程初始化时间轮的启动时间// Wait until the startTime is initialized by the worker.while (startTime == 0) {try {//这里使用countDownLauch来确保调度的线程已经被启动startTimeInitialized.await();} catch (InterruptedException ignore) {// Ignore - it will be ready very soon.}}}start方法会根据当前的workerState状态来启动时间轮 。并且用了startTimeInitialized来控制线程的运行，如果workerThread没有启动起来，那么newTimeout方法会一直阻塞在运行start方法中 。如果不阻塞，newTimeout方法会获取不到startTime 。
启动时间轮时间轮的启动在HashedWheelTimer的内部类Worker中 。调用workerThread#start方法会调用Worker的run方法启动时间轮 。
下面我们看时间轮启动做了什么，下面的分析不考虑任务被取消的情况 。
Worker#runpublic void run() {// Initialize the startTime.startTime = System.nanoTime();if (startTime == 0) {// We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.startTime = 1;}//HashedWheelTimer的start方法会继续往下运行// Notify the other threads waiting for the initialization at start().startTimeInitialized.countDown();do {//返回的是当前的nanoTime- startTime//也就是返回的是 每 tick 一次的时间间隔final long deadline = waitForNextTick();if (deadline > 0) {//算出时间轮的槽位int idx = (int) (tick & mask);//移除cancelledTimeouts中的bucket// 从bucket中移除timeoutprocessCancelledTasks();HashedWheelBucket bucket = wheel[idx];// 将newTimeout()方法中加入到待处理定时任务队列中的任务加入到指定的格子中transferTimeoutsToBuckets();bucket.expireTimeouts(deadline);tick++;}//校验如果workerState是started状态，那么就一直循环} while (workerStateUpdater.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);// Fill the unprocessedTimeouts so we can return them from stop() method.for (HashedWheelBucket bucket : wheel) {bucket.clearTimeouts(unprocessedTimeouts);}for (;;) {HashedWheelTimeout timeout = timeouts.poll();if (timeout == null) {break;}//如果有没有被处理的timeout，那么加入到unprocessedTimeouts对列中if (!timeout.isCancelled()) {unprocessedTimeouts.add(timeout);}}//处理被取消的任务processCancelledTasks();}


推荐阅读


上一篇：如何在Java中创建不可变类？

下一篇：抖音如何选品 抖音带货哪里选商品