feat(demo-thread): add time wheel implementation

Author: helltractor

demo-thread/src/main/java/com/helltractor/demo/timewheel/Demo.java

@@ -0,0 +1,15 @@
+package com.helltractor.demo.timewheel;
+
+public class Demo {
+    
+    public static void main(String[] args) {
+        TimeWheel timeWheel = new TimeWheel();
+        for (int i = 0; i < 123; i++) {
+            final int index = i;
+            timeWheel.addDelayTask(() -> {
+                System.out.println(index);
+            }, 1000L * i);
+        }
+    }
+    
+}

demo-thread/src/main/java/com/helltractor/demo/timewheel/MpscTaskQueue.java

@@ -0,0 +1,54 @@
+package com.helltractor.demo.timewheel;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * A multi-producer single-consumer task queue for scheduling tasks in a time wheel.
+ */
+public class MpscTaskQueue {
+    
+    private final AtomicReference<TimeWheel.DelayTask> head = new AtomicReference<>(null);
+    
+    public void pushTask(TimeWheel.DelayTask delayTask) {
+        while (true) {
+            TimeWheel.DelayTask currentHead = head.get();
+            delayTask.next = currentHead;
+            if (head.compareAndSet(currentHead, delayTask)) {
+                return;
+            }
+        }
+    }
+    
+    public List<Runnable> removeAndReturnTasks(long tickTime) {
+        List<Runnable> result = new ArrayList<>();
+        TimeWheel.DelayTask prev = null;
+        TimeWheel.DelayTask current = head.get();
+        
+        while (current != null) {
+            if (current.deadline > tickTime) {
+                prev = current;
+                current = current.next;
+                continue;
+            }
+            TimeWheel.DelayTask next = current.next;
+            if (prev != null) {
+                prev.next = next;
+                current.next = null;
+                current = next;
+                continue;
+            }
+            if (head.compareAndSet(current, next)) {
+                result.add(current.task);
+                current.next = null;
+                current = next;
+                continue;
+            }
+            current = head.get();
+        }
+        
+        return result;
+    }
+    
+}

demo-thread/src/main/java/com/helltractor/demo/timewheel/TimeWheel.java

@@ -0,0 +1,99 @@
+package com.helltractor.demo.timewheel;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.LockSupport;
+
+/**
+ * A simple time wheel implementation for scheduling delayed tasks.
+ * Use MPSC task queues for each slot to allow concurrent task additions.
+ */
+public class TimeWheel {
+    
+    private final AtomicBoolean started;
+    private final CountDownLatch startTimeLatch;
+    private final Ticker ticker;
+    private final ExecutorService executor;
+    private final MpscTaskQueue[] wheel;
+    
+    // avoid multi-thread read/write issues
+    private volatile long startTime;
+    
+    public TimeWheel() {
+        this.started = new AtomicBoolean(false);
+        this.startTimeLatch = new CountDownLatch(1);
+        this.ticker = new Ticker();
+        this.executor = Executors.newFixedThreadPool(8);
+        this.wheel = new MpscTaskQueue[10];
+        for (int i = 0; i < wheel.length; i++) {
+            wheel[i] = new MpscTaskQueue();
+        }
+    }
+    
+    public void start() {
+        if (started.compareAndSet(false, true)) {
+            ticker.start();
+        }
+        try {
+            startTimeLatch.await();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+    
+    public void stop() {
+        if (started.compareAndSet(true, false)) {
+            LockSupport.unpark(ticker);
+        }
+    }
+    
+    public void addDelayTask(Runnable task, long delayMillis) {
+        start();
+        DelayTask delayTask = new DelayTask(task, delayMillis);
+        int index = Math.toIntExact((delayTask.deadline - startTime) / 100 % wheel.length);
+        MpscTaskQueue slot = wheel[index];
+        slot.pushTask(delayTask);
+    }
+    
+    static class DelayTask {
+        final Runnable task;
+        final long deadline;
+        DelayTask next;
+        
+        public DelayTask(Runnable task, long delayTime) {
+            this.task = task;
+            this.deadline = System.currentTimeMillis() + delayTime;
+        }
+    }
+    
+    class Ticker extends Thread {
+        
+        int tickCount = 0;
+        
+        @Override
+        public void run() {
+            startTime = System.currentTimeMillis();
+            startTimeLatch.countDown();
+            while (started.get()) {
+                long tickTime = startTime + (tickCount + 1) * 100L;
+                while (System.currentTimeMillis() <= tickTime) {
+                    LockSupport.parkUntil(tickTime);
+                    // check if stopped
+                    if (!started.get()) {
+                        return;
+                    }
+                }
+                int index = tickCount % wheel.length;
+                MpscTaskQueue queue = wheel[index];
+                List<Runnable> tasks = queue.removeAndReturnTasks(tickTime);
+                tasks.forEach(executor::execute);
+                tickCount++;
+            }
+            
+        }
+    }
+    
+}

demo-thread/src/test/java/com/helltractor/demo/timewheel/TimeWheelTest.java

@@ -0,0 +1,30 @@
+package com.helltractor.demo.timewheel;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+class TimerWheelTest {
+    
+    @Test
+    void testThreadSafe() throws InterruptedException {
+        TimeWheel timerWheel = new TimeWheel();
+        CountDownLatch countDownLatch = new CountDownLatch(10000);
+        AtomicInteger count = new AtomicInteger();
+        for (int i = 0; i < 10; i++) {
+            new Thread(() -> {
+                for (int i1 = 0; i1 < 1000; i1++) {
+                    timerWheel.addDelayTask(() -> {
+                        System.out.println(count.incrementAndGet());
+                        countDownLatch.countDown();
+                    }, 100);
+                }
+                System.out.println("添加结束");
+            }).start();
+        }
+        countDownLatch.await();
+    }
+    
+}