raft算法重构

Author: beston123

src/main/java/code/distribution/raft/NodeManageService.java

@@ -0,0 +1,61 @@
+package code.distribution.raft;
+
+import code.distribution.raft.enums.NodeEventType;
+import code.distribution.raft.enums.RoleType;
+import code.distribution.raft.model.NodeEvent;
+import code.util.NamedThreadFactory;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 〈节点管理服务〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author beston
+ * @date 2020/12/08
+ */
+public class NodeManageService implements IService {
+
+    private final RaftNodeServer nodeServer;
+
+    private boolean stop;
+
+    public NodeManageService(RaftNodeServer nodeServer) {
+        this.nodeServer = nodeServer;
+    }
+
+    private final ScheduledExecutorService nodeManagerTimer = new ScheduledThreadPoolExecutor(1,
+            new NamedThreadFactory("NodeManagerTimer-"), new ThreadPoolExecutor.CallerRunsPolicy());
+
+    @Override
+    public void start() {
+        stop = false;
+        nodeManagerTimer.scheduleAtFixedRate(() -> {
+            while (!stop) {
+                if (nodeServer.getNode().getRole() == RoleType.LEADER) {
+                    try {
+                        NodeEvent nodeEvent = RaftClusterManager.newNodeQueue().take();
+                        System.out.println("Handle node event:" + nodeEvent);
+                        if (nodeEvent.getEventType() == NodeEventType.ADD) {
+                            nodeServer.addNode(nodeEvent.getNodeId());
+                        } else if (nodeEvent.getEventType() == NodeEventType.REMOVE) {
+                            nodeServer.removeNode(nodeEvent.getNodeId());
+                        }
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            }
+        }, RaftConst.SCHEDULER_DELAY_MS, RaftConst.SLEEP_DEVIATION_MS, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void close() {
+        stop = true;
+        nodeManagerTimer.shutdownNow();
+    }
+
+}

src/main/java/code/distribution/raft/RaftClusterManager.java

@@ -0,0 +1,107 @@
+package code.distribution.raft;
+
+import code.distribution.raft.model.NodeCommand;
+import code.distribution.raft.model.NodeEvent;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
+/**
+ * 〈集群管理〉<p>
+ * 〈维护集群节点的上下线〉
+ *
+ * @author zixiao
+ * @date 2019/3/7
+ */
+public class RaftClusterManager {
+
+    /**
+     * 所有节点
+     */
+    private static Set<String> allNodes = new HashSet<>();
+
+    /**
+     * 除自身外的索引节点
+     */
+    private static Set<String> otherNodes = new HashSet<>();
+
+    /**
+     * 新变更节点队列
+     */
+    private static BlockingQueue<NodeEvent> newNodeQueue = new LinkedBlockingQueue<>();
+
+    /**
+     * 自身节点
+     * 如果有值，则本节点是raft服务端
+     * 如果为null，则本节点是raft客户端
+     */
+    private static String selfId;
+
+    public static void config(String[] nodeIds){
+        for (String nodeId : nodeIds) {
+            allNodes.add(nodeId);
+        }
+    }
+
+    public static void config(RaftConfig raftConfig){
+        config(raftConfig.parseClusterNodes());
+        if(raftConfig.getNodeId() != null){
+            selfId = raftConfig.getNodeId();
+            otherNodes = allNodes.stream().filter(nodeId -> !nodeId.equals(selfId)).collect(Collectors.toSet());
+        }
+    }
+
+    public static List<String> allNodes(){
+        return new ArrayList<>(allNodes);
+    }
+
+    public static Set<String> otherNodes(){
+        return otherNodes;
+    }
+
+    public static BlockingQueue<NodeEvent> newNodeQueue(){
+        return newNodeQueue;
+    }
+
+    public static int nodeNum(){
+       return allNodes.size();
+    }
+
+    public static boolean exist(String nodeId){
+        return allNodes().contains(nodeId);
+    }
+
+    public static boolean addNode(String nodeId){
+        synchronized (allNodes) {
+            //是否为raft服务端
+            if (selfId != null && !nodeId.equals(selfId)) {
+                otherNodes.add(nodeId);
+            }
+            boolean success = allNodes.add(nodeId);
+            if (success) {
+                newNodeQueue.add(NodeEvent.buildAdd(nodeId));
+            }
+            return success;
+        }
+    }
+
+    public static boolean removeNode(String nodeId){
+        synchronized (allNodes) {
+            //是否为raft服务端
+            if (selfId != null) {
+                otherNodes.remove(nodeId);
+            }
+            boolean success = allNodes.remove(nodeId);
+            if (success) {
+                newNodeQueue.add(NodeEvent.buildRemove(nodeId));
+            }
+            return success;
+        }
+    }
+
+}

src/main/java/code/distribution/raft/RaftConfig.java

@@ -13,6 +13,11 @@
 @Data
 public class RaftConfig {
 
+    /**
+     * 机器节点Id
+     */
+    private String nodeId;
+
     /**
      * 集群节点配置，以英文逗号,或分号;分隔
      */

src/main/java/code/distribution/raft/RaftConst.java

@@ -11,13 +11,23 @@ public interface RaftConst {
     /**
      * 心跳定时器
      */
-    long HEARTBEAT_MS = 100;
+    long HEARTBEAT_MS = 150;
 
     /**
      * 选举定时器
      * 必须要大于5倍心跳定时，建议是10倍关系
      */
-    long ELECTION_TIMEOUT_MS = 1000;
+    long ELECTION_TIMEOUT_MS = 1500;
+
+    /**
+     * 休眠时间
+     */
+    long SLEEP_DEVIATION_MS = 50;
+
+    /**
+     * 定时器启动延时
+     */
+    long SCHEDULER_DELAY_MS = 3000;
 
     /**
      * 空对象的term值
@@ -27,7 +37,7 @@ public interface RaftConst {
     /**
      * 重试日志复制
      */
-    long RETRY_APPEND_MS = 3000;
+    long RETRY_APPEND_MS = 1000;
 
     /**
      * 换行符