Merge pull request #33 from crossoverJie/cim-1.0.4

cim 1.0.4

Author: crossoverJie

cim-client/src/main/java/com/crossoverjie/cim/client/handle/CIMClientHandle.java

@@ -1,5 +1,6 @@
 package com.crossoverjie.cim.client.handle;
 
+import com.crossoverjie.cim.client.service.ShutDownMsg;
 import com.crossoverjie.cim.client.thread.ReConnectJob;
 import com.crossoverjie.cim.client.util.SpringBeanFactory;
 import com.crossoverjie.cim.common.constant.Constants;
@@ -37,6 +38,8 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
 
     private ScheduledExecutorService scheduledExecutorService ;
 
+    private ShutDownMsg shutDownMsg ;
+
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
@@ -71,11 +74,20 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        LOGGER.info("客户端断开了，重新连接！");
+
+        if (shutDownMsg == null){
+            shutDownMsg = SpringBeanFactory.getBean(ShutDownMsg.class) ;
+        }
+
+        //用户主动退出，不执行重连逻辑
+        if (shutDownMsg.checkStatus()){
+            return;
+        }
 
         if (scheduledExecutorService == null){
             scheduledExecutorService = SpringBeanFactory.getBean("scheduledTask",ScheduledExecutorService.class) ;
         }
+        LOGGER.info("客户端断开了，重新连接！");
         // TODO: 2019-01-22 后期可以改为不用定时任务，连上后就关闭任务 节省性能。
         scheduledExecutorService.scheduleAtFixedRate(new ReConnectJob(ctx),0,10, TimeUnit.SECONDS) ;
     }

cim-client/src/main/java/com/crossoverjie/cim/client/service/ShutDownMsg.java

@@ -0,0 +1,26 @@
+package com.crossoverjie.cim.client.service;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ * Date: 2019-02-27 16:17
+ * @since JDK 1.8
+ */
+@Component
+public class ShutDownMsg {
+    private boolean isCommand ;
+
+    /**
+     * 置为用户主动退出状态
+     */
+    public void shutdown(){
+        isCommand = true ;
+    }
+
+    public boolean checkStatus(){
+        return isCommand ;
+    }
+}

cim-client/src/main/java/com/crossoverjie/cim/client/service/impl/command/QueryHistoryCommand.java

@@ -25,6 +25,9 @@ public class QueryHistoryCommand implements InnerCommand {
     @Override
     public void process(String msg) {
         String[] split = msg.split(" ");
+        if (split.length < 2){
+            return;
+        }
         String res = msgLogger.query(split[1]);
         System.out.println(res);
     }

cim-client/src/main/java/com/crossoverjie/cim/client/service/impl/command/ShutDownCommand.java

@@ -4,6 +4,7 @@
 import com.crossoverjie.cim.client.service.InnerCommand;
 import com.crossoverjie.cim.client.service.MsgLogger;
 import com.crossoverjie.cim.client.service.RouteRequest;
+import com.crossoverjie.cim.client.service.ShutDownMsg;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,9 +37,14 @@ public class ShutDownCommand implements InnerCommand {
     @Resource(name = "callBackThreadPool")
     private ThreadPoolExecutor executor;
 
+
+    @Autowired
+    private ShutDownMsg shutDownMsg ;
+
     @Override
     public void process(String msg) {
         LOGGER.info("系统关闭中。。。。");
+        shutDownMsg.shutdown();
         routeRequest.offLine();
         msgLogger.stop();
         executor.shutdown();

cim-client/src/main/resources/application.properties

@@ -45,8 +45,8 @@ cim.clear.route.request.url=http://45.78.28.220:8083/offLine
 #cim.clear.route.request.url=http://localhost:8083/offLine
 
 # 客户端唯一ID
-cim.user.id=1545574841528
-cim.user.userName=zhangsan
+cim.user.id=1551267098213
+cim.user.userName=test3
 
 # 回调线程队列大小
 cim.callback.thread.queue.size = 2

cim-common/src/main/java/com/crossoverjie/cim/common/data/construct/SortArrayMap.java

@@ -0,0 +1,128 @@
+package com.crossoverjie.cim.common.data.construct;
+
+import java.util.Arrays;
+import java.util.Comparator;
+
+/**
+ * Function:根据 key 排序的 Map
+ *
+ * @author crossoverJie
+ * Date: 2019-02-25 18:17
+ * @since JDK 1.8
+ */
+public class SortArrayMap {
+
+    /**
+     * 核心数组
+     */
+    private Node[] buckets;
+
+    private static final int DEFAULT_SIZE = 10;
+
+    /**
+     * 数组大小
+     */
+    private int size = 0;
+
+    public SortArrayMap() {
+        buckets = new Node[DEFAULT_SIZE];
+    }
+
+    /**
+     * 写入数据
+     * @param key
+     * @param value
+     */
+    public void add(Long key, String value) {
+        checkSize(size + 1);
+        Node node = new Node(key, value);
+        buckets[size++] = node;
+    }
+
+    /**
+     * 校验是否需要扩容
+     * @param size
+     */
+    private void checkSize(int size) {
+        if (size >= buckets.length) {
+            //扩容自身的 3/2
+            int oldLen = buckets.length;
+            int newLen = oldLen + (oldLen >> 1);
+            buckets = Arrays.copyOf(buckets, newLen);
+        }
+    }
+
+    /**
+     * 顺时针取出数据
+     * @param key
+     * @return
+     */
+    public String firstNodeValue(long key) {
+        if (size == 0){
+            return null ;
+        }
+        for (Node bucket : buckets) {
+            if (bucket == null){
+                continue;
+            }
+            if (bucket.key >= key) {
+                return bucket.value;
+            }
+        }
+
+        return buckets[0].value;
+
+    }
+
+    /**
+     * 排序
+     */
+    public void sort() {
+        Arrays.sort(buckets, 0, size, new Comparator<Node>() {
+            @Override
+            public int compare(Node o1, Node o2) {
+                if (o1.key > o2.key) {
+                    return 1;
+                } else {
+                    return -1;
+                }
+            }
+        });
+    }
+
+    public void print() {
+        for (Node bucket : buckets) {
+            if (bucket == null) {
+                continue;
+            }
+            System.out.println(bucket.toString());
+        }
+    }
+
+    public int size() {
+        return size;
+    }
+
+    /**
+     * 数据节点
+     */
+    private class Node {
+        public Long key;
+        public String value;
+
+        public Node(Long key, String value) {
+            this.key = key;
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return "Node{" +
+                    "key=" + key +
+                    ", value='" + value + '\'' +
+                    '}';
+        }
+
+    }
+
+}

cim-common/src/main/java/com/crossoverjie/cim/common/route/algorithm/RouteHandle.java

@@ -0,0 +1,21 @@
+package com.crossoverjie.cim.common.route.algorithm;
+
+import java.util.List;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ * Date: 2019-02-27 00:31
+ * @since JDK 1.8
+ */
+public interface RouteHandle {
+
+    /**
+     * 再一批服务器里进行路由
+     * @param values
+     * @param key
+     * @return
+     */
+    String routeServer(List<String> values,String key) ;
+}

cim-common/src/main/java/com/crossoverjie/cim/common/route/algorithm/consistenthash/AbstractConsistentHash.java

@@ -0,0 +1,84 @@
+package com.crossoverjie.cim.common.route.algorithm.consistenthash;
+
+import java.io.UnsupportedEncodingException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+
+/**
+ * Function:一致性 hash 算法抽象类
+ *
+ * @author crossoverJie
+ * Date: 2019-02-27 00:35
+ * @since JDK 1.8
+ */
+public abstract class AbstractConsistentHash {
+
+    /**
+     * 新增节点
+     * @param key
+     * @param value
+     */
+    protected abstract void add(long key,String value);
+
+    /**
+     * 排序节点，数据结构自身支持排序可以不用重写
+     */
+    protected void sort(){}
+
+    /**
+     * 根据当前的 key 通过一致性 hash 算法的规则取出一个节点
+     * @param value
+     * @return
+     */
+    protected abstract String getFirstNodeValue(String value);
+
+    /**
+     * 传入节点列表以及客户端信息获取一个服务节点
+     * @param values
+     * @param key
+     * @return
+     */
+    public String process(List<String> values,String key){
+
+        for (String value : values) {
+            add(hash(value), value);
+        }
+        sort();
+
+        return getFirstNodeValue(key) ;
+    }
+
+    /**
+     * hash 运算
+     * @param value
+     * @return
+     */
+    public Long hash(String value){
+        MessageDigest md5;
+        try {
+            md5 = MessageDigest.getInstance("MD5");
+        } catch (NoSuchAlgorithmException e) {
+            throw new RuntimeException("MD5 not supported", e);
+        }
+        md5.reset();
+        byte[] keyBytes = null;
+        try {
+            keyBytes = value.getBytes("UTF-8");
+        } catch (UnsupportedEncodingException e) {
+            throw new RuntimeException("Unknown string :" + value, e);
+        }
+
+        md5.update(keyBytes);
+        byte[] digest = md5.digest();
+
+        // hash code, Truncate to 32-bits
+        long hashCode = ((long) (digest[3] & 0xFF) << 24)
+                | ((long) (digest[2] & 0xFF) << 16)
+                | ((long) (digest[1] & 0xFF) << 8)
+                | (digest[0] & 0xFF);
+
+        long truncateHashCode = hashCode & 0xffffffffL;
+        return truncateHashCode;
+    }
+}

cim-common/src/main/java/com/crossoverjie/cim/common/route/algorithm/consistenthash/ConsistentHashHandle.java

@@ -0,0 +1,25 @@
+package com.crossoverjie.cim.common.route.algorithm.consistenthash;
+
+import com.crossoverjie.cim.common.route.algorithm.RouteHandle;
+
+import java.util.List;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ * Date: 2019-02-27 00:33
+ * @since JDK 1.8
+ */
+public class ConsistentHashHandle implements RouteHandle {
+    private AbstractConsistentHash hash ;
+
+    public void setHash(AbstractConsistentHash hash) {
+        this.hash = hash;
+    }
+
+    @Override
+    public String routeServer(List<String> values, String key) {
+        return hash.process(values, key);
+    }
+}