Java 时间轮机制实现

潇凝子潇

127人浏览 · 2026-01-25 17:48:07

潇凝子潇 · 2026-01-25 17:48:07 发布

1. 核心组件定义

TimerTask 接口

/**
 * 定时任务接口
 */
public interface TimerTask {
    /**
     * 任务执行方法
     * @param timeout 超时对象
     */
    void run(Timeout timeout);
}

Timeout 接口

/**
 * 超时接口
 */
public interface Timeout {
    /**
     * 获取任务
     * @return 定时任务
     */
    TimerTask task();
    
    /**
     * 是否已过期
     * @return true表示已过期
     */
    boolean isExpired();
    
    /**
     * 是否已取消
     * @return true表示已取消
     */
    boolean isCancelled();
    
    /**
     * 取消任务
     * @return true表示取消成功
     */
    boolean cancel();
}

2. 定时任务实体类

TimerTaskEntry - 定时任务条目

/**
 * 定时任务条目，用于维护任务之间的链表关系
 */
class TimerTaskEntry implements Comparable<TimerTaskEntry> {
    private volatile boolean cancelled = false;
    private TimerTaskEntry next;
    private TimerTaskEntry prev;
    private final TimerTask timerTask;
    private final long expirationMs;

    public TimerTaskEntry(TimerTask timerTask, long expirationMs) {
        this.timerTask = timerTask;
        this.expirationMs = expirationMs;
    }

    public boolean isCancelled() {
        return cancelled;
    }

    public void cancel() {
        cancelled = true;
    }

    public TimerTask timerTask() {
        return timerTask;
    }

    public long expirationMs() {
        return expirationMs;
    }

    /**
     * 从链表中移除当前节点
     */
    public void remove() {
        synchronized (this) {
            if (next != null) {
                next.prev = prev;
            }
            if (prev != null) {
                prev.next = next;
            }
            next = null;
            prev = null;
        }
    }

    @Override
    public int compareTo(TimerTaskEntry o) {
        return Long.compare(this.expirationMs, o.expirationMs);
    }
}

3. 任务列表管理

TimerTaskList - 定时任务链表

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 定时任务链表，用于管理同一时间槽中的多个任务
 */
class TimerTaskList {
    private final AtomicInteger taskCounter = new AtomicInteger(0);
    private final TimerTaskEntry sentinal = new TimerTaskEntry(null, -1); // 哨兵节点

    public TimerTaskList() {
        sentinal.next = sentinal;
        sentinal.prev = sentinal;
    }

    /**
     * 添加任务到链表
     * @param timerTaskEntry 任务条目
     * @return true表示添加成功
     */
    public boolean add(TimerTaskEntry timerTaskEntry) {
        boolean done = false;
        while (!done) {
            timerTaskEntry.remove();
            synchronized (this) {
                if (timerTaskEntry.isCancelled()) {
                    done = true;
                } else {
                    // 添加到链表头部
                    timerTaskEntry.next = sentinal.next;
                    timerTaskEntry.prev = sentinal;
                    sentinal.next.prev = timerTaskEntry;
                    sentinal.next = timerTaskEntry;
                    taskCounter.incrementAndGet();
                    done = true;
                }
            }
        }
        return true;
    }

    /**
     * 移除任务
     * @param timerTaskEntry 任务条目
     * @return true表示移除成功
     */
    public boolean remove(TimerTaskEntry timerTaskEntry) {
        synchronized (this) {
            if (timerTaskEntry.isCancelled()) {
                return false;
            }
            timerTaskEntry.remove();
            int currentCount = taskCounter.decrementAndGet();
            if (currentCount == 0) {
                // 如果链表为空，重置哨兵节点
                sentinal.next = sentinal;
                sentinal.prev = sentinal;
            }
            return true;
        }
    }

    /**
     * 清空链表
     */
    public void clear() {
        synchronized (this) {
            // 重置哨兵节点
            sentinal.next = sentinal;
            sentinal.prev = sentinal;
            taskCounter.set(0);
        }
    }

    /**
     * 过期任务
     * @param currentTime 当前时间
     * @return 过期任务列表
     */
    public java.util.List<TimerTaskEntry> expiredTimeouts(long currentTime) {
        java.util.List<TimerTaskEntry> expired = new java.util.ArrayList<>();
        synchronized (this) {
            TimerTaskEntry entry = sentinal.next;
            while (entry != sentinal) {
                TimerTaskEntry next = entry.next;
                if (entry.expirationMs <= currentTime) {
                    entry.remove();
                    taskCounter.decrementAndGet();
                    if (!entry.isCancelled()) {
                        expired.add(entry);
                    }
                }
                entry = next;
            }
        }
        return expired;
    }

    public int size() {
        return taskCounter.get();
    }
}

4. 时间槽（Bucket）

Bucket - 时间槽容器

import java.util.concurrent.atomic.AtomicLong;

/**
 * 时间槽，用于存储特定时间范围内的任务
 */
class Bucket {
    private final TimerTaskList taskList = new TimerTaskList();
    private final AtomicLong expiration = new AtomicLong(-1L);
    private volatile Bucket next = null;

    /**
     * 添加任务到时间槽
     * @param timeout 定时任务条目
     */
    public void addTask(TimerTaskEntry timeout) {
        boolean done = taskList.add(timeout);
        if (!done) {
            return;
        }

        // 更新过期时间
        long bucketExpiration = expiration.get();
        if (timeout.expirationMs() < bucketExpiration || bucketExpiration == -1L) {
            expiration.set(timeout.expirationMs());
        }
    }

    /**
     * 过期任务
     * @param currentTime 当前时间
     * @return 过期任务列表
     */
    public java.util.List<TimerTaskEntry> expiredTimeouts(long currentTime) {
        // 设置过期时间为-1，表示已处理
        expiration.set(-1L);
        return taskList.expiredTimeouts(currentTime);
    }

    public long getExpiration() {
        return expiration.get();
    }

    public boolean isEmpty() {
        return taskList.size() == 0;
    }
}

5. 时间轮核心实现

TimingWheel - 时间轮

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 时间轮实现
 */
public class TimingWheel {
    private final long tickMs; // 每个时间槽的间隔时间
    private final int wheelSize; // 时间轮大小
    private final long interval; // 时间轮总时间跨度
    private final AtomicLong currentTime; // 当前时间指针
    private final java.util.List<Bucket> buckets; // 时间槽数组
    
    // 下一层时间轮，用于处理超过当前时间轮范围的任务
    private final TimingWheel overflowWheel;
    
    // 时间槽索引
    private final AtomicLong tick = new AtomicLong(0);

    public TimingWheel(long tickMs, int wheelSize, long startTime, TimingWheel overflowWheel) {
        this.tickMs = tickMs;
        this.wheelSize = wheelSize;
        this.interval = tickMs * wheelSize;
        this.currentTime = new AtomicLong(startTime - (startTime % tickMs));
        this.overflowWheel = overflowWheel;
        
        this.buckets = new java.util.ArrayList<>(wheelSize);
        for (int i = 0; i < wheelSize; i++) {
            buckets.add(new Bucket());
        }
    }

    public TimingWheel(long tickMs, int wheelSize, long startTime) {
        this(tickMs, wheelSize, startTime, null);
    }

    /**
     * 添加定时任务
     * @param timerTaskEntry 任务条目
     * @return true表示添加成功
     */
    public boolean add(TimerTaskEntry timerTaskEntry) {
        long expiration = timerTaskEntry.expirationMs();
        
        if (timerTaskEntry.isCancelled()) {
            return false;
        }

        long calculatedExpiration = expiration - currentTime.get();
        
        if (calculatedExpiration < tickMs) {
            // 任务将在下一个tick内过期，立即执行
            return false;
        } else if (calculatedExpiration < interval) {
            // 计算时间槽位置
            long virtualId = expiration / tickMs;
            int index = (int) (virtualId % wheelSize);
            
            Bucket bucket = buckets.get(index);
            bucket.addTask(timerTaskEntry);
            
            // 更新时间槽过期时间
            long newExpiration = bucket.getExpiration();
            if (newExpiration > 0) {
                long startOfBucket = (virtualId * tickMs);
                if (tick.compareAndSet(startOfBucket, startOfBucket + tickMs)) {
                    // 更新下一层时间轮
                    if (overflowWheel != null) {
                        overflowWheel.add(bucket);
                    }
                }
            }
            return true;
        } else {
            // 任务超出了当前时间轮范围，添加到上一层时间轮
            if (overflowWheel != null) {
                return overflowWheel.add(timerTaskEntry);
            } else {
                return false;
            }
        }
    }

    /**
     * 推进时间指针
     * @param time 新时间
     */
    public void advanceClock(long time) {
        if (time >= currentTime.get() + tickMs) {
            currentTime.set(time - (time % tickMs));
            
            // 推进上一层时间轮
            if (overflowWheel != null) {
                overflowWheel.advanceClock(currentTime.get());
            }
        }
    }

    public long currentTime() {
        return currentTime.get();
    }
}

6. 定时器实现

HashedWheelTimer - 哈希时间轮定时器

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 哈希时间轮定时器
 */
public class HashedWheelTimer implements Timer {
    private static final int WORKER_STATE_INIT = 0;
    private static final int WORKER_STATE_STARTED = 1;
    private static final int WORKER_STATE_SHUTDOWN = 2;
    
    private final AtomicBoolean shutdown = new AtomicBoolean(false);
    private final AtomicInteger workerState = new AtomicInteger(WORKER_STATE_INIT);
    private final Thread workerThread;
    
    // 时间轮
    private final TimingWheel timingWheel;
    
    // 任务队列
    private final BlockingQueue<HashedWheelTimeout> timeouts = new LinkedBlockingQueue<>();
    
    // 超时计数器
    private final AtomicLong timeoutCounter = new AtomicLong(0);
    
    // 执行器
    private final ExecutorService taskExecutor;

    public HashedWheelTimer() {
        this(100, 512, System.currentTimeMillis());
    }

    public HashedWheelTimer(long tickDuration, int ticksPerWheel, long startTime) {
        this.timingWheel = new TimingWheel(tickDuration, ticksPerWheel, startTime);
        this.taskExecutor = Executors.newCachedThreadPool();
        
        this.workerThread = new Thread(new Worker(), "HashedWheelTimerWorker");
        workerThread.setDaemon(true);
    }

    @Override
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        
        if (shutdown.get()) {
            throw new IllegalStateException("cannot schedule task after timer has been shut down");
        }

        long deadline = System.currentTimeMillis() + unit.toMillis(delay);
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        
        timeouts.offer(timeout);
        return timeout;
    }

    @Override
    public Set<Timeout> stop() {
        if (!shutdown.compareAndSet(false, true)) {
            return Collections.emptySet();
        }

        boolean interrupted = false;
        while (workerState.get() != WORKER_STATE_SHUTDOWN) {
            try {
                workerThread.interrupt();
                workerThread.join(100);
            } catch (InterruptedException e) {
                interrupted = true;
            }
        }

        if (interrupted) {
            Thread.currentThread().interrupt();
        }

        return Collections.emptySet();
    }

    private class Worker implements Runnable {
        private long tick = 0;

        @Override
        public void run() {
            if (!workerState.compareAndSet(WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                return;
            }

            while (!shutdown.get()) {
                // 处理新任务
                fetchFromBucket();
                
                // 推进时间轮
                timingWheel.advanceClock(System.currentTimeMillis());
                
                // 获取过期任务
                java.util.List<TimerTaskEntry> expired = getExpiredTimeouts();
                
                // 执行过期任务
                for (TimerTaskEntry entry : expired) {
                    HashedWheelTimeout timeout = (HashedWheelTimeout) entry.timerTask();
                    if (!timeout.isCancelled()) {
                        taskExecutor.execute(() -> timeout.task().run(timeout));
                    }
                }
                
                try {
                    Thread.sleep(1); // 简单的等待机制
                } catch (InterruptedException e) {
                    break;
                }
            }
            
            workerState.set(WORKER_STATE_SHUTDOWN);
        }
        
        private void fetchFromBucket() {
            HashedWheelTimeout timeout;
            while ((timeout = timeouts.poll()) != null) {
                if (!timeout.isCancelled()) {
                    if (!timingWheel.add(timeout.getEntry())) {
                        // 任务应该立即执行
                        if (!timeout.isCancelled()) {
                            taskExecutor.execute(() -> timeout.task().run(timeout));
                        }
                    }
                }
            }
        }
        
        private java.util.List<TimerTaskEntry> getExpiredTimeouts() {
            // 获取当前时间槽中的过期任务
            return new java.util.ArrayList<>(); // 简化实现
        }
    }

    private static final class HashedWheelTimeout implements Timeout {
        private final HashedWheelTimer timer;
        private final TimerTask task;
        private final long deadline;
        private final TimerTaskEntry entry;
        private volatile boolean cancelled = false;

        HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
            this.timer = timer;
            this.task = task;
            this.deadline = deadline;
            this.entry = new TimerTaskEntry(this, deadline);
        }

        @Override
        public TimerTask task() {
            return task;
        }

        @Override
        public boolean isExpired() {
            return System.currentTimeMillis() > deadline;
        }

        @Override
        public boolean isCancelled() {
            return cancelled;
        }

        @Override
        public boolean cancel() {
            if (!cancelled) {
                cancelled = true;
                entry.cancel();
                return true;
            }
            return false;
        }
        
        public TimerTaskEntry getEntry() {
            return entry;
        }
    }
}

7. Timer 接口

import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 定时器接口
 */
public interface Timer {
    /**
     * 创建新的定时任务
     * @param task 任务
     * @param delay 延迟时间
     * @param unit 时间单位
     * @return 超时对象
     */
    Timeout newTimeout(TimerTask task, long delay, TimeUnit unit);
    
    /**
     * 停止定时器
     * @return 未执行的任务集合
     */
    Set<Timeout> stop();
}

8. 使用示例

public class TimerExample {
    public static void main(String[] args) {
        // 创建时间轮定时器
        Timer timer = new HashedWheelTimer(10, 512, System.currentTimeMillis());
        
        // 创建一个简单的定时任务
        TimerTask task = new TimerTask() {
            @Override
            public void run(Timeout timeout) {
                System.out.println("定时任务执行，当前时间：" + System.currentTimeMillis());
            }
        };
        
        // 添加定时任务，延迟2秒后执行
        Timeout timeout = timer.newTimeout(task, 2, TimeUnit.SECONDS);
        
        // 等待任务执行
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        // 停止定时器
        timer.stop();
    }
}