1. thread safe ---> addTask

2. to optimize the api

@@ -7,6 +7,7 @@ import java.util.HashSet;
 import java.util.Set;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.Lock;
@@ -38,7 +39,7 @@ public final class RingBufferWheel {
      */
      */
     private ExecutorService executorService;
     private ExecutorService executorService;
 
 
-    private AtomicInteger taskSize = new AtomicInteger();
+    private volatile int size = 0 ;
 
 
     /***
     /***
      * task stop sign
      * task stop sign
@@ -48,18 +49,19 @@ public final class RingBufferWheel {
     /**
     /**
      * task start sign
      * task start sign
      */
      */
-    private volatile boolean start = false ;
+    private volatile AtomicBoolean start = new AtomicBoolean(false);
 
 
     /**
     /**
      * total tick times
      * total tick times
      */
      */
-    private AtomicInteger tick = new AtomicInteger() ;
+    private AtomicInteger tick = new AtomicInteger();
 
 
     private Lock lock = new ReentrantLock();
     private Lock lock = new ReentrantLock();
     private Condition condition = lock.newCondition();
     private Condition condition = lock.newCondition();
 
 
     /**
     /**
      * Create a new delay task ring buffer by default size
      * Create a new delay task ring buffer by default size
+     *
      * @param executorService the business thread pool
      * @param executorService the business thread pool
      */
      */
     public RingBufferWheel(ExecutorService executorService) {
     public RingBufferWheel(ExecutorService executorService) {
@@ -71,6 +73,7 @@ public final class RingBufferWheel {
 
 
     /**
     /**
      * Create a new delay task ring buffer by custom buffer size
      * Create a new delay task ring buffer by custom buffer size
+     *
      * @param executorService the business thread pool
      * @param executorService the business thread pool
      * @param bufferSize      custom buffer size
      * @param bufferSize      custom buffer size
      */
      */
@@ -86,10 +89,14 @@ public final class RingBufferWheel {
 
 
     /**
     /**
      * Add a task into the ring buffer
      * Add a task into the ring buffer
+     *
      * @param task business task extends RingBufferWheel.Task
      * @param task business task extends RingBufferWheel.Task
      */
      */
     public void addTask(Task task) {
     public void addTask(Task task) {
         int key = task.getKey();
         int key = task.getKey();
+
+        try {
+            lock.lock();
             Set<Task> tasks = get(key);
             Set<Task> tasks = get(key);
 
 
             if (tasks != null) {
             if (tasks != null) {
@@ -105,35 +112,45 @@ public final class RingBufferWheel {
                 sets.add(task);
                 sets.add(task);
                 put(key, sets);
                 put(key, sets);
             }
             }
-
-        taskSize.incrementAndGet();
+            size ++ ;
+        }finally {
+            lock.unlock();
+        }
 
 
         start();
         start();
+
+
     }
     }
 
 
     /**
     /**
      * thread safe
      * thread safe
+     *
      * @return the size of ring buffer
      * @return the size of ring buffer
      */
      */
     public int taskSize() {
     public int taskSize() {
-        return taskSize.get();
+        return size;
     }
     }
 
 
     /**
     /**
      * Start background thread to consumer wheel timer, it will always run until you call method {@link #stop}
      * Start background thread to consumer wheel timer, it will always run until you call method {@link #stop}
      */
      */
     public void start() {
     public void start() {
-        if (!start){
+        if (!start.get()) {
+
+            if (start.compareAndSet(start.get(), true)) {
                 logger.info("delay task is starting");
                 logger.info("delay task is starting");
                 Thread job = new Thread(new TriggerJob());
                 Thread job = new Thread(new TriggerJob());
                 job.setName("consumer RingBuffer thread");
                 job.setName("consumer RingBuffer thread");
                 job.start();
                 job.start();
-            start = true ;
+                start.set(true);
+            }
+
         }
         }
     }
     }
 
 
     /**
     /**
      * Stop consumer ring buffer thread
      * Stop consumer ring buffer thread
+     *
      * @param force True will force close consumer thread and discard all pending tasks
      * @param force True will force close consumer thread and discard all pending tasks
      *              otherwise the consumer thread waits for all tasks to completes before closing.
      *              otherwise the consumer thread waits for all tasks to completes before closing.
      */
      */
@@ -144,7 +161,7 @@ public final class RingBufferWheel {
             executorService.shutdownNow();
             executorService.shutdownNow();
         } else {
         } else {
             logger.info("delay task is stopping");
             logger.info("delay task is stopping");
-            if (taskSize() > 0){
+            if (taskSize() > 0) {
                 try {
                 try {
                     lock.lock();
                     lock.lock();
                     condition.await();
                     condition.await();
@@ -202,11 +219,11 @@ public final class RingBufferWheel {
     private void size2Notify() {
     private void size2Notify() {
         try {
         try {
             lock.lock();
             lock.lock();
-            int size = taskSize.decrementAndGet();
+            size -- ;
             if (size == 0) {
             if (size == 0) {
                 condition.signal();
                 condition.signal();
             }
             }
-        }finally {
+        } finally {
             lock.unlock();
             lock.unlock();
         }
         }
     }
     }
@@ -225,7 +242,7 @@ public final class RingBufferWheel {
 
 
     private int mod(int target, int mod) {
     private int mod(int target, int mod) {
         // equals target % mod
         // equals target % mod
-        target = target + tick.get() ;
+        target = target + tick.get();
         return target & (mod - 1);
         return target & (mod - 1);
     }
     }
 
 
@@ -272,7 +289,7 @@ public final class RingBufferWheel {
         public void run() {
         public void run() {
             int index = 0;
             int index = 0;
             while (!stop) {
             while (!stop) {
-
+                try {
                     Set<Task> tasks = remove(index);
                     Set<Task> tasks = remove(index);
                     for (Task task : tasks) {
                     for (Task task : tasks) {
                         executorService.submit(task);
                         executorService.submit(task);
@@ -284,11 +301,12 @@ public final class RingBufferWheel {
 
 
                     //Total tick number of records
                     //Total tick number of records
                     tick.incrementAndGet();
                     tick.incrementAndGet();
-                try {
                     TimeUnit.SECONDS.sleep(1);
                     TimeUnit.SECONDS.sleep(1);
-                } catch (InterruptedException e) {
-                    logger.error("InterruptedException", e);
+
+                } catch (Exception e) {
+                    logger.error("Exception", e);
                 }
                 }
+
             }
             }
 
 
             logger.info("delay task is stopped");
             logger.info("delay task is stopped");

@@ -3,6 +3,9 @@ package com.crossoverjie.cim.common.data.construct;
 
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
@@ -18,10 +21,10 @@ public class RingBufferWheelTest {
 
 
     private static Logger logger = LoggerFactory.getLogger(RingBufferWheelTest.class) ;
     private static Logger logger = LoggerFactory.getLogger(RingBufferWheelTest.class) ;
 
 
-    public static void main(String[] args) throws InterruptedException {
-        test5();
+    public static void main(String[] args) throws Exception {
+
+        concurrentTest();
 
 
-        return;
     }
     }
 
 
     private static void test1() throws InterruptedException {
     private static void test1() throws InterruptedException {
@@ -150,7 +153,7 @@ public class RingBufferWheelTest {
 
 
 
 
     }
     }
-    private static void cuncrrentTest6() throws InterruptedException {
+    private static void concurrentTest() throws Exception {
         BlockingQueue<Runnable> queue = new LinkedBlockingQueue(10);
         BlockingQueue<Runnable> queue = new LinkedBlockingQueue(10);
         ThreadFactory product = new ThreadFactoryBuilder()
         ThreadFactory product = new ThreadFactoryBuilder()
                 .setNameFormat("msg-callback-%d")
                 .setNameFormat("msg-callback-%d")
@@ -161,25 +164,19 @@ public class RingBufferWheelTest {
         ExecutorService executorService = Executors.newFixedThreadPool(10) ;
         ExecutorService executorService = Executors.newFixedThreadPool(10) ;
         RingBufferWheel wheel = new RingBufferWheel(executorService) ;
         RingBufferWheel wheel = new RingBufferWheel(executorService) ;
 
 
+        for (int i = 0; i < 10; i++) {
             business.execute(new Runnable() {
             business.execute(new Runnable() {
                 @Override
                 @Override
                 public void run() {
                 public void run() {
-
+                    for (int i1 = 0; i1 < 30; i1++) {
+                        RingBufferWheel.Task task = new Job(i1) ;
+                        task.setKey(i1);
+                        wheel.addTask(task);
+                    }
                 }
                 }
             });
             });
-
-        for (int i = 0; i < 10; i++) {
-            RingBufferWheel.Task task = new Job(i) ;
-            task.setKey(i);
-            wheel.addTask(task);
         }
         }
 
 
-
-        TimeUnit.SECONDS.sleep(10);
-        RingBufferWheel.Task task = new Job(15) ;
-        task.setKey(15);
-        wheel.addTask(task);
-
         logger.info("task size={}",wheel.taskSize());
         logger.info("task size={}",wheel.taskSize());
 
 
         wheel.stop(false);
         wheel.stop(false);
@@ -209,4 +206,37 @@ public class RingBufferWheelTest {
         }
         }
 
 
     }
     }
+
+
+    public static void hashTimerTest(){
+
+        BlockingQueue<Runnable> queue = new LinkedBlockingQueue(10);
+        ThreadFactory product = new ThreadFactoryBuilder()
+                .setNameFormat("msg-callback-%d")
+                .setDaemon(true)
+                .build();
+        ThreadPoolExecutor business = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MILLISECONDS, queue,product);
+        HashedWheelTimer hashedWheelTimer = new HashedWheelTimer() ;
+
+        for (int i = 0; i < 10; i++) {
+            int finalI = i;
+            business.execute(new Runnable() {
+                @Override
+                public void run() {
+
+                    for (int i1 = 0; i1 < 10; i1++) {
+                        hashedWheelTimer.newTimeout(new TimerTask() {
+                            @Override
+                            public void run(Timeout timeout) throws Exception {
+                                logger.info("====" + finalI);
+                            }
+                        }, finalI,TimeUnit.SECONDS) ;
+                    }
+                }
+            });
+        }
+
+
+
+    }
 }
 }