Introducing new features.ring buffer delay task data struct

@@ -15,6 +15,11 @@
 
     <dependencies>
         <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+
+        <dependency>
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
         </dependency>

+package com.crossoverjie.cim.common.data.construct;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Function:Ring Queue, it can be used to delay task.
+ *
+ * @author crossoverJie
+ * Date: 2019-09-20 14:46
+ * @since JDK 1.8
+ */
+public final class RingBufferWheel {
+
+    private Logger logger = LoggerFactory.getLogger(RingBufferWheel.class);
+
+
+    private static final int STATIC_RING_SIZE = 64;
+
+    private Object[] ringBuffer;
+
+    private int bufferSize;
+
+    private ExecutorService executorService;
+
+    private AtomicInteger taskSize = new AtomicInteger();
+
+    private volatile boolean stop = false;
+
+    private Lock lock = new ReentrantLock();
+    private Condition condition = lock.newCondition();
+
+    public RingBufferWheel(ExecutorService executorService) {
+        this.executorService = executorService;
+        bufferSize = STATIC_RING_SIZE;
+        ringBuffer = new Object[bufferSize];
+    }
+
+
+    public void addTask(Task task) {
+        int key = task.getKey();
+        Set<Task> tasks = get(key);
+
+        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);
+            Set<Task> sets = new HashSet<>();
+            sets.add(task);
+            put(key, sets);
+        }
+
+        taskSize.incrementAndGet();
+
+    }
+
+    public int taskSize() {
+        return taskSize.get();
+    }
+
+    public void start() {
+        logger.info("delay task is starting");
+        Thread job = new Thread(new TriggerJob());
+        job.setName("consumer RingBuffer thread");
+        job.start();
+    }
+
+    public void stop(boolean force) {
+        if (force) {
+            logger.info("delay task is forced stop");
+            stop = true;
+            executorService.shutdownNow();
+        } else {
+            logger.info("delay task is stopping");
+            try {
+                lock.lock();
+                condition.await();
+                stop = true;
+            } catch (InterruptedException e) {
+                logger.error("InterruptedException", e);
+            } finally {
+                lock.unlock();
+            }
+            executorService.shutdown();
+        }
+
+
+    }
+
+
+    private Set<Task> get(int key) {
+        int index = mod(key, bufferSize);
+        return (Set<Task>) ringBuffer[index];
+    }
+
+    private void put(int key, Set<Task> tasks) {
+        int index = mod(key, bufferSize);
+        ringBuffer[index] = tasks;
+    }
+
+    private Set<Task> remove(int key) {
+        Set<Task> tempTask = new HashSet<>();
+        Set<Task> result = new HashSet<>();
+
+        Set<Task> tasks = (Set<Task>) ringBuffer[key];
+        if (tasks == null) {
+            return result;
+        }
+
+        for (Task task : tasks) {
+            if (task.getCycleNum() == 0) {
+                result.add(task);
+
+                size2Notify();
+            } else {
+                // decrement 1 cycle number and update origin data
+                task.setCycleNum(task.getCycleNum() - 1);
+                tempTask.add(task);
+            }
+        }
+
+        //update origin data
+        ringBuffer[key] = tempTask;
+
+        return result;
+    }
+
+    private void size2Notify() {
+        lock.lock();
+        int size = taskSize.decrementAndGet();
+        if (size == 0) {
+            condition.signal();
+        }
+        lock.unlock();
+    }
+
+    private int mod(int target, int mod) {
+        // equals target % mod
+        return target & (mod - 1);
+    }
+
+    private int cycleNum(int target, int mod) {
+        //equals target/mod
+        return target >> Integer.bitCount(mod - 1);
+    }
+
+    public abstract static class Task extends Thread {
+
+
+        private int cycleNum;
+
+        private int key;
+
+        @Override
+        public void run() {
+        }
+
+        public int getKey() {
+            return key;
+        }
+
+        public void setKey(int key) {
+            this.key = key;
+        }
+
+        public int getCycleNum() {
+            return cycleNum;
+        }
+
+        private void setCycleNum(int cycleNum) {
+            this.cycleNum = cycleNum;
+        }
+    }
+
+
+    private class TriggerJob implements Runnable {
+
+        @Override
+        public void run() {
+            int index = 0;
+            while (!stop) {
+
+                Set<Task> tasks = remove(index);
+                for (Task task : tasks) {
+                    executorService.submit(task);
+                }
+
+                if (++index > bufferSize - 1) {
+                    index = 0;
+                }
+
+                try {
+                    TimeUnit.SECONDS.sleep(1);
+                } catch (InterruptedException e) {
+                    logger.error("InterruptedException", e);
+                }
+            }
+
+            logger.info("delay task is stopped");
+        }
+    }
+}

+package com.crossoverjie.cim.common;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ * Date: 2019-09-23 14:21
+ * @since JDK 1.8
+ */
+public class CommonTest {
+
+    @Test
+    public void test() throws InterruptedException {
+
+
+        System.out.println(is2(9));
+
+        System.out.println(Integer.bitCount(64-1));
+
+        int target = 1569312600 ;
+        int mod = 64 ;
+        System.out.println(target % mod);
+
+        System.out.println(mod(target,mod));
+        System.out.println("============");
+
+        System.out.println(cycleNum(256,64)) ;
+
+        cycle();
+    }
+
+
+    private int mod(int target, int mod){
+        // equals target % mod
+        return target & (mod -1) ;
+    }
+
+    private int cycleNum(int target,int mod){
+        //equals target/mod
+        return target >> Integer.bitCount(mod-1) ;
+    }
+
+    private boolean is2(int target){
+        if (target < 0){
+            return false ;
+        }
+
+        int value = target & (target - 1) ;
+        if (value != 0){
+            return false ;
+        }
+
+        return true ;
+    }
+
+
+    private void cycle() throws InterruptedException {
+        int index = 0  ;
+        while (true){
+            System.out.println("=======" + index);
+
+            if (++index > 63){
+                index = 0 ;
+            }
+            TimeUnit.MILLISECONDS.sleep(200);
+        }
+    }
+
+}

+package com.crossoverjie.cim.common.data.construct;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+public class RingBufferWheelTest {
+
+    private static Logger logger = LoggerFactory.getLogger(RingBufferWheelTest.class) ;
+
+    public static void main(String[] args) throws InterruptedException {
+        test4();
+
+        return;
+    }
+
+    private static void test1() throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(2) ;
+
+        Task task = new Task() ;
+        task.setKey(10);
+        RingBufferWheel wheel = new RingBufferWheel(executorService) ;
+        wheel.addTask(task) ;
+
+        task = new Task() ;
+        task.setKey(74);
+        wheel.addTask(task) ;
+
+        wheel.start();
+
+        while (true){
+            logger.info("task size={}" , wheel.taskSize());
+            TimeUnit.SECONDS.sleep(1);
+        }
+    }
+    private static void test2() throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(2) ;
+
+        Task task = new Task() ;
+        task.setKey(10);
+        RingBufferWheel wheel = new RingBufferWheel(executorService) ;
+        wheel.addTask(task) ;
+
+        task = new Task() ;
+        task.setKey(74);
+        wheel.addTask(task) ;
+
+        wheel.start();
+
+//        new Thread(() -> {
+//            while (true){
+//                logger.info("task size={}" , wheel.taskSize());
+//                try {
+//                    TimeUnit.SECONDS.sleep(1);
+//                } catch (InterruptedException e) {
+//                    e.printStackTrace();
+//                }
+//            }
+//        }).start();
+
+        TimeUnit.SECONDS.sleep(12);
+        wheel.stop(true);
+
+
+    }
+    private static void test3() throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(2) ;
+
+        Task task = new Task() ;
+        task.setKey(10);
+        RingBufferWheel wheel = new RingBufferWheel(executorService) ;
+        wheel.addTask(task) ;
+
+        task = new Task() ;
+        task.setKey(74);
+        wheel.addTask(task) ;
+
+        wheel.start();
+
+
+        TimeUnit.SECONDS.sleep(2);
+        wheel.stop(false);
+
+
+    }
+    private static void test4() throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(2) ;
+
+        RingBufferWheel wheel = new RingBufferWheel(executorService) ;
+
+        for (int i = 0; i < 65; i++) {
+            Job task = new Job(i) ;
+            task.setKey(i);
+            wheel.addTask(task);
+        }
+
+        wheel.start();
+
+        logger.info("task size={}",wheel.taskSize());
+
+        wheel.stop(false);
+
+
+    }
+
+
+    private static class Task extends RingBufferWheel.Task{
+
+        @Override
+        public void run() {
+            logger.info("================");
+        }
+    }
+
+    private static class Job extends RingBufferWheel.Task{
+
+        private int num ;
+
+        public Job(int num) {
+            this.num = num;
+        }
+
+        @Override
+        public void run() {
+            logger.info("number={}" , num);
+        }
+    }
+}
\ No newline at end of file