HashedWheelTimer源码浅析

傷城~ 2022-11-18 02:03 228阅读 0赞

HashedWheelTimer由HashedWheelBucket数组组成，通过newTimeout方法提交任务并启动Worker工作线程，工作线程会不断将新的Timeout加到HashedWheelBucket中，并执行HashedWheelBucket中的Timeout。

### 1.Timeout ###

通过Timeout可以获取关联的Timer和TimerTask。

public interface Timeout { 
        Timer timer();
        TimerTask task();
        boolean isExpired();
        boolean isCancelled();
        boolean cancel();
    }

### 2.HashedWheelTimeout ###

HashedWheelTimeout还可以获取前一个或者后一个Timeout。

private static final class HashedWheelTimeout implements Timeout { 
    
    	private final HashedWheelTimer timer;
        private final TimerTask task;
        private final long deadline;
        private volatile int state = ST_INIT;
        long remainingRounds;
        HashedWheelTimeout next;
        HashedWheelTimeout prev;
        HashedWheelBucket bucket;
        
        HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) { 
            this.timer = timer;
            this.task = task;
            this.deadline = deadline;
        }
        
        @Override
        public boolean cancel() { 
            if (!compareAndSetState(ST_INIT, ST_CANCELLED)) { 
                return false;
            }
            // 将Timeout加入队列中
            timer.cancelledTimeouts.add(this);
            return true;
        }
        
        // 移除该Timeout并且pendingTimeouts减一
        void remove() { 
            HashedWheelBucket bucket = this.bucket;
            if (bucket != null) { 
                bucket.remove(this);
            } else { 
                timer.pendingTimeouts.decrementAndGet();
            }
        }
        
        // 执行任务
        public void expire() { 
            if (!compareAndSetState(ST_INIT, ST_EXPIRED)) { 
                return;
            }
    
            try { 
                task.run(this);
            } catch (Throwable t) { 
                if (logger.isWarnEnabled()) { 
                    logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
                }
            }
        }
    }

### 3.HashedWheelBucket ###

HashedWheelBucket是用来存储HashedWheelTimeouts的双向链表。

private static final class HashedWheelBucket { 
    
        private HashedWheelTimeout head;
        private HashedWheelTimeout tail;
        
        // 将Timeout加到链表中，每次从加到链表后面
        void addTimeout(HashedWheelTimeout timeout) { 
            assert timeout.bucket == null;
            timeout.bucket = this;
            if (head == null) { 
                head = tail = timeout;
            } else { 
                tail.next = timeout;
                timeout.prev = tail;
                tail = timeout;
            }
        }
        
        // 处理所有的Timeout
        void expireTimeouts(long deadline) { 
            HashedWheelTimeout timeout = head;
    
            while (timeout != null) { 
                HashedWheelTimeout next = timeout.next;
                if (timeout.remainingRounds <= 0) { 
                    next = remove(timeout);
                    if (timeout.deadline <= deadline) { 
                        timeout.expire();
                    } else { 
                        throw new IllegalStateException(String.format(
                            "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                } else if (timeout.isCancelled()) { 
                   	// 移除timeout并返回其下一个节点
                    next = remove(timeout);
                } else { 
                    timeout.remainingRounds--;
                }
                timeout = next;
            }
        }
        
    }

### 4.HashedWheelTimer ###

public class HashedWheelTimer implements Timer { 
        
        // 工作线程
        private final Worker worker = new Worker();
        private final Thread workerThread;
        
        // 工作状态 0 - init, 1 - started, 2 - shut down
        private volatile int workerState;
        
        // Timeout队列
        private final Queue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
        private final Queue<HashedWheelTimeout> cancelledTimeouts = new LinkedBlockingQueue<>();
        
        // Timeouts数量
        private final AtomicLong pendingTimeouts = new AtomicLong(0);
        private final long maxPendingTimeouts;
    }

public HashedWheelTimer(
            ThreadFactory threadFactory,
            long tickDuration, TimeUnit unit, int ticksPerWheel,
            long maxPendingTimeouts) { 
    
        if (threadFactory == null) { 
            throw new NullPointerException("threadFactory");
        }
        if (unit == null) { 
            throw new NullPointerException("unit");
        }
        if (tickDuration <= 0) { 
            throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
        }
        if (ticksPerWheel <= 0) { 
            throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
        }
    
        // 创建Bucket数组，初始化时间轮
        wheel = createWheel(ticksPerWheel);
        mask = wheel.length - 1;
    
        this.tickDuration = unit.toNanos(tickDuration);
    
        if (this.tickDuration >= Long.MAX_VALUE / wheel.length) { 
            throw new IllegalArgumentException(String.format(
                    "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
                    tickDuration, Long.MAX_VALUE / wheel.length));
        }
        // 创建工作线程
        workerThread = threadFactory.newThread(worker);
    
        this.maxPendingTimeouts = maxPendingTimeouts;
    
        if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
                WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) { 
            reportTooManyInstances();
        }
    }

// 提交新任务并返回关联的Timeout
    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 + ")");
        }
    
        // 如果工作状态是0改为1并启动工作线程。等待工作线程初始化startTime.
        start();
    
        long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
    
        if (delay > 0 && deadline < 0) { 
            deadline = Long.MAX_VALUE;
        }
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        timeouts.add(timeout);
        return timeout;
    }

### 5.Worker ###

private final class Worker implements Runnable { 
            private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
            private long tick;   
    }

public void run() { 
        // 初始化startTime.
        startTime = System.nanoTime();
        if (startTime == 0) { 
            // 0表示没初始化，初始化了就不是0
            startTime = 1;
        }
    
        // 通知start()方法中等待初始化startTime的线程
        startTimeInitialized.countDown();
    
        // 只要工作状态是1就一直循环
        do { 
            final long deadline = waitForNextTick();
            if (deadline > 0) { 
                int idx = (int) (tick & mask);
                // 移除取消的任务
                processCancelledTasks();
                // 获取要处理的Bucket
                HashedWheelBucket bucket =
                        wheel[idx];
                // 从timeouts中取出Timeout并加入到对应的Bucket中
                transferTimeoutsToBuckets();
                // 执行Bucket中的任务
                bucket.expireTimeouts(deadline);
                tick++;
            }
        } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
    
        // 清理Bucket并将不是expired或cancelled的Timeout加入到unprocessedTimeouts中
        for (HashedWheelBucket bucket : wheel) { 
            bucket.clearTimeouts(unprocessedTimeouts);
        }
        for (; ; ) { 
            HashedWheelTimeout timeout = timeouts.poll();
            if (timeout == null) { 
                break;
            }
            if (!timeout.isCancelled()) { 
                unprocessedTimeouts.add(timeout);
            }
        }
        processCancelledTasks();
    }