Documenting source code.

@@ -3,7 +3,9 @@ package com.crossoverjie.cim.common.data.construct;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -39,7 +41,7 @@ public final class RingBufferWheel {
      */
     private ExecutorService executorService;
 
-    private volatile int size = 0 ;
+    private volatile int size = 0;
 
     /***
      * task stop sign
@@ -59,6 +61,9 @@ public final class RingBufferWheel {
     private Lock lock = new ReentrantLock();
     private Condition condition = lock.newCondition();
 
+    private AtomicInteger taskId = new AtomicInteger();
+    private Map<Integer, Task> taskMap = new HashMap<>(16);
+
     /**
      * Create a new delay task ring buffer by default size
      *
@@ -88,23 +93,25 @@ public final class RingBufferWheel {
     }
 
     /**
-     * Add a task into the ring buffer
+     * Add a task into the ring buffer(thread safe)
      *
-     * @param task business task extends RingBufferWheel.Task
+     * @param task business task extends {@link Task}
      */
-    public void addTask(Task task) {
+    public int addTask(Task task) {
         int key = task.getKey();
+        int id;
 
         try {
             lock.lock();
-            Set<Task> tasks = get(key);
+            int index = mod(key, bufferSize);
+            task.setIndex(index);
+            Set<Task> tasks = get(index);
 
             if (tasks != null) {
                 int cycleNum = cycleNum(key, bufferSize);
                 task.setCycleNum(cycleNum);
                 tasks.add(task);
             } else {
-                int index = mod(key, bufferSize);
                 int cycleNum = cycleNum(key, bufferSize);
                 task.setCycleNum(index);
                 task.setCycleNum(cycleNum);
@@ -112,18 +119,57 @@ public final class RingBufferWheel {
                 sets.add(task);
                 put(key, sets);
             }
-            size ++ ;
-        }finally {
+            id = taskId.incrementAndGet();
+            taskMap.put(id, task);
+            size++;
+        } finally {
             lock.unlock();
         }
 
         start();
 
+        return id;
+    }
+
+
+    /**
+     * Cancel task by taskId
+     * @param id unique id through {@link #addTask(Task)}
+     * @return
+     */
+    public boolean cancel(int id) {
+
+        boolean flag = false;
+        Set<Task> tempTask = new HashSet<>();
+
+        try {
+            lock.lock();
+            Task task = taskMap.get(id);
+            if (task == null) {
+                return false;
+            }
+
+            Set<Task> tasks = get(task.getIndex());
+            for (Task tk : tasks) {
+                if (tk.getKey() == task.getKey() && tk.getCycleNum() == task.getCycleNum()) {
+                    size--;
+                    flag = true;
+                } else {
+                    tempTask.add(tk);
+                }
 
             }
+            //update origin data
+            ringBuffer[task.getIndex()] = tempTask;
+        } finally {
+            lock.unlock();
+        }
+
+        return flag;
+    }
 
     /**
-     * thread safe
+     * Thread safe
      *
      * @return the size of ring buffer
      */
@@ -179,8 +225,7 @@ public final class RingBufferWheel {
     }
 
 
-    private Set<Task> get(int key) {
-        int index = mod(key, bufferSize);
+    private Set<Task> get(int index) {
         return (Set<Task>) ringBuffer[index];
     }
 
@@ -219,7 +264,7 @@ public final class RingBufferWheel {
     private void size2Notify() {
         try {
             lock.lock();
-            size -- ;
+            size--;
             if (size == 0) {
                 condition.signal();
             }
@@ -256,6 +301,7 @@ public final class RingBufferWheel {
      */
     public abstract static class Task extends Thread {
 
+        private int index;
 
         private int cycleNum;
 
@@ -280,6 +326,14 @@ public final class RingBufferWheel {
         private void setCycleNum(int cycleNum) {
             this.cycleNum = cycleNum;
         }
+
+        public int getIndex() {
+            return index;
+        }
+
+        private void setIndex(int index) {
+            this.index = index;
+        }
     }
 
 

@@ -2,7 +2,6 @@ package com.crossoverjie.cim.common.data.construct;
 
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.Timeout;
 import io.netty.util.TimerTask;
@@ -23,7 +22,7 @@ public class RingBufferWheelTest {
 
     public static void main(String[] args) throws Exception {
 
-        test6();
+        test7();
 
     }
 
@@ -147,9 +146,38 @@ public class RingBufferWheelTest {
         logger.info("task size={}",wheel.taskSize());
 
         wheel.stop(false);
+    }
+
+    private static void test7() throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(2) ;
 
+        RingBufferWheel wheel = new RingBufferWheel(executorService,512) ;
 
+        for (int i = 0; i < 10; i++) {
+            RingBufferWheel.Task task = new Job(i) ;
+            task.setKey(i);
+            wheel.addTask(task);
         }
+
+        RingBufferWheel.Task task = new Job(15) ;
+        task.setKey(15);
+        int cancel = wheel.addTask(task);
+
+        new Thread(() -> {
+            boolean flag = wheel.cancel(cancel);
+            logger.info("cancel task={}",flag) ;
+        }).start();
+
+        RingBufferWheel.Task task1 = new Job(20) ;
+        task1.setKey(20);
+        wheel.addTask(task1) ;
+
+        logger.info("task size={}",wheel.taskSize());
+
+        wheel.stop(false);
+    }
+
+
     private static void concurrentTest() throws Exception {
         BlockingQueue<Runnable> queue = new LinkedBlockingQueue(10);
         ThreadFactory product = new ThreadFactoryBuilder()