Merge pull request #58 from crossoverJie/cim-1.0.7

cim 1.0.7

@@ -40,7 +40,6 @@ public class DelayMsgCommand implements InnerCommand {
         RingBufferWheel.Task task = new DelayMsgJob(message) ;
         RingBufferWheel.Task task = new DelayMsgJob(message) ;
         task.setKey(delayTime);
         task.setKey(delayTime);
         ringBufferWheel.addTask(task);
         ringBufferWheel.addTask(task);
-        ringBufferWheel.start();
         echoService.echo(EmojiParser.parseToUnicode(msg));
         echoService.echo(EmojiParser.parseToUnicode(msg));
     }
     }
 
 

@@ -3,10 +3,13 @@ package com.crossoverjie.cim.common.data.construct;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.HashSet;
+import java.util.Map;
 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,25 +41,32 @@ public final class RingBufferWheel {
      */
      */
     private ExecutorService executorService;
     private ExecutorService executorService;
 
 
-    private AtomicInteger taskSize = new AtomicInteger();
+    private volatile int size = 0;
 
 
     /***
     /***
-     * task running sign
+     * task stop sign
      */
      */
     private volatile boolean stop = false;
     private volatile boolean stop = false;
 
 
-    private volatile boolean start = false ;
+    /**
+     * task start sign
+     */
+    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();
 
 
+    private AtomicInteger taskId = new AtomicInteger();
+    private Map<Integer, Task> taskMap = new HashMap<>(16);
+
     /**
     /**
      * 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) {
@@ -68,8 +78,9 @@ 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
      */
      */
     public RingBufferWheel(ExecutorService executorService, int bufferSize) {
     public RingBufferWheel(ExecutorService executorService, int bufferSize) {
         this(executorService);
         this(executorService);
@@ -82,56 +93,110 @@ public final class RingBufferWheel {
     }
     }
 
 
     /**
     /**
-     * Add a task into the ring buffer
-     * @param task business task extends RingBufferWheel.Task
+     * Add a task into the ring buffer(thread safe)
+     *
+     * @param task business task extends {@link Task}
      */
      */
-    public void addTask(Task task) {
+    public int addTask(Task task) {
         int key = task.getKey();
         int key = task.getKey();
-        Set<Task> tasks = get(key);
+        int id;
 
 
-        if (tasks != null) {
-            int cycleNum = cycleNum(key, bufferSize);
-            task.setCycleNum(cycleNum);
-            tasks.add(task);
-        } else {
+        try {
+            lock.lock();
             int index = mod(key, bufferSize);
             int index = mod(key, bufferSize);
-            int cycleNum = cycleNum(key, bufferSize);
-            task.setCycleNum(index);
-            task.setCycleNum(cycleNum);
-            Set<Task> sets = new HashSet<>();
-            sets.add(task);
-            put(key, sets);
+            task.setIndex(index);
+            Set<Task> tasks = get(index);
+
+            if (tasks != null) {
+                int cycleNum = cycleNum(key, bufferSize);
+                task.setCycleNum(cycleNum);
+                tasks.add(task);
+            } else {
+                int cycleNum = cycleNum(key, bufferSize);
+                task.setCycleNum(index);
+                task.setCycleNum(cycleNum);
+                Set<Task> sets = new HashSet<>();
+                sets.add(task);
+                put(key, sets);
+            }
+            id = taskId.incrementAndGet();
+            taskMap.put(id, task);
+            size++;
+        } finally {
+            lock.unlock();
         }
         }
 
 
-        taskSize.incrementAndGet();
+        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
      * @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.get()) {
+
+            if (start.compareAndSet(start.get(), true)) {
+                logger.info("delay task is starting");
+                Thread job = new Thread(new TriggerJob());
+                job.setName("consumer RingBuffer thread");
+                job.start();
+                start.set(true);
+            }
 
 
-        if (!start){
-            logger.info("delay task is starting");
-            Thread job = new Thread(new TriggerJob());
-            job.setName("consumer RingBuffer thread");
-            job.start();
-            start = 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.
      */
      */
@@ -142,7 +207,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();
@@ -160,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];
         return (Set<Task>) ringBuffer[index];
     }
     }
 
 
@@ -200,11 +264,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();
         }
         }
     }
     }
@@ -223,7 +287,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);
     }
     }
 
 
@@ -237,6 +301,7 @@ public final class RingBufferWheel {
      */
      */
     public abstract static class Task extends Thread {
     public abstract static class Task extends Thread {
 
 
+        private int index;
 
 
         private int cycleNum;
         private int cycleNum;
 
 
@@ -261,6 +326,14 @@ public final class RingBufferWheel {
         private void setCycleNum(int cycleNum) {
         private void setCycleNum(int cycleNum) {
             this.cycleNum = cycleNum;
             this.cycleNum = cycleNum;
         }
         }
+
+        public int getIndex() {
+            return index;
+        }
+
+        private void setIndex(int index) {
+            this.index = index;
+        }
     }
     }
 
 
 
 
@@ -270,23 +343,24 @@ 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);
+                    for (Task task : tasks) {
+                        executorService.submit(task);
+                    }
 
 
-                Set<Task> tasks = remove(index);
-                for (Task task : tasks) {
-                    executorService.submit(task);
-                }
-
-                if (++index > bufferSize - 1) {
-                    index = 0;
-                }
+                    if (++index > bufferSize - 1) {
+                        index = 0;
+                    }
 
 
-                //Total tick number of records
-                tick.incrementAndGet();
-                try {
+                    //Total tick number of records
+                    tick.incrementAndGet();
                     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");

 package com.crossoverjie.cim.common.data.construct;
 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;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
 
 
 public class RingBufferWheelTest {
 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 {
-        test6();
+    public static void main(String[] args) throws Exception {
+
+        test7();
 
 
-        return;
     }
     }
 
 
     private static void test1() throws InterruptedException {
     private static void test1() throws InterruptedException {
@@ -29,8 +38,6 @@ public class RingBufferWheelTest {
         task.setKey(74);
         task.setKey(74);
         wheel.addTask(task) ;
         wheel.addTask(task) ;
 
 
-        wheel.start();
-
         while (true){
         while (true){
             logger.info("task size={}" , wheel.taskSize());
             logger.info("task size={}" , wheel.taskSize());
             TimeUnit.SECONDS.sleep(1);
             TimeUnit.SECONDS.sleep(1);
@@ -75,11 +82,9 @@ public class RingBufferWheelTest {
         wheel.addTask(task) ;
         wheel.addTask(task) ;
 
 
         task = new Task() ;
         task = new Task() ;
-        task.setKey(74);
+        task.setKey(60);
         wheel.addTask(task) ;
         wheel.addTask(task) ;
 
 
-        wheel.start();
-
 
 
         TimeUnit.SECONDS.sleep(2);
         TimeUnit.SECONDS.sleep(2);
         wheel.stop(false);
         wheel.stop(false);
@@ -116,8 +121,6 @@ public class RingBufferWheelTest {
             wheel.addTask(task);
             wheel.addTask(task);
         }
         }
 
 
-        wheel.start();
-
         logger.info("task size={}",wheel.taskSize());
         logger.info("task size={}",wheel.taskSize());
 
 
         wheel.stop(false);
         wheel.stop(false);
@@ -135,13 +138,69 @@ public class RingBufferWheelTest {
             wheel.addTask(task);
             wheel.addTask(task);
         }
         }
 
 
-        wheel.start();
-
-        TimeUnit.SECONDS.sleep(10);
+        TimeUnit.SECONDS.sleep(5);
         RingBufferWheel.Task task = new Job(15) ;
         RingBufferWheel.Task task = new Job(15) ;
         task.setKey(15);
         task.setKey(15);
         wheel.addTask(task);
         wheel.addTask(task);
-        wheel.start();
+
+        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()
+                .setNameFormat("msg-callback-%d")
+                .setDaemon(true)
+                .build();
+        ThreadPoolExecutor business = new ThreadPoolExecutor(4, 4, 1, TimeUnit.MILLISECONDS, queue,product);
+
+        ExecutorService executorService = Executors.newFixedThreadPool(10) ;
+        RingBufferWheel wheel = new RingBufferWheel(executorService) ;
+
+        for (int i = 0; i < 10; i++) {
+            business.execute(new Runnable() {
+                @Override
+                public void run() {
+                    for (int i1 = 0; i1 < 30; i1++) {
+                        RingBufferWheel.Task task = new Job(i1) ;
+                        task.setKey(i1);
+                        wheel.addTask(task);
+                    }
+                }
+            });
+        }
 
 
         logger.info("task size={}",wheel.taskSize());
         logger.info("task size={}",wheel.taskSize());
 
 
@@ -172,4 +231,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) ;
+                    }
+                }
+            });
+        }
+
+
+
+    }
 }
 }