public void signalPublish(String key, Record value) throws Exception {
        // 注释：步骤1 如果当前不是 leader，直接将该请求给 leader
        if (!isLeader()) {
            JSONObject params = new JSONObject();
            params.put("key", key);
            params.put("value", value);
            Map<String, String> parameters = new HashMap<>(1);
            parameters.put("key", key);

            raftProxy.proxyPostLarge(getLeader().ip, API_PUB, params.toJSONString(), parameters);
            return;
        }

        try {
            OPERATE_LOCK.lock();
            long start = System.currentTimeMillis();
            final Datum datum = new Datum();
            datum.key = key;
            datum.value = value;
            if (getDatum(key) == null) {
                datum.timestamp.set(1L);
            } else {
                datum.timestamp.set(getDatum(key).timestamp.incrementAndGet());
            }

            JSONObject json = new JSONObject();
            json.put("datum", datum);
            json.put("source", peers.local());
            // 注释：本节点发布 service 发布事件
            onPublish(datum, peers.local());

            final String content = JSON.toJSONString(json);

            final CountDownLatch latch = new CountDownLatch(peers.majorityCount());
            for (final String server : peers.allServersIncludeMyself()) {
                // 注释：这里应该是避免特殊情况：步骤1 之前正在选举 leader，在这里之间选举出了 leader
                if (isLeader(server)) {
                    latch.countDown();
                    continue;
                }
                final String url = buildURL(server, API_ON_PUB);
                HttpClient.asyncHttpPostLarge(url, Arrays.asList("key=" + key), content, new AsyncCompletionHandler<Integer>() {
                    @Override
                    public Integer onCompleted(Response response) throws Exception {
                        if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                            Loggers.RAFT.warn("[RAFT] failed to publish data to peer, datumId={}, peer={}, http code={}",
                                datum.key, server, response.getStatusCode());
                            return 1;
                        }
                        latch.countDown();
                        return 0;
                    }

                    @Override
                    public STATE onContentWriteCompleted() {
                        return STATE.CONTINUE;
                    }
                });

            }   
            // 注释：超时 5s 就算失败，但是其实本节点已经执行了 service 创建事件
            if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) {
                // only majority servers return success can we consider this update success
                Loggers.RAFT.error("data publish failed, caused failed to notify majority, key={}", key);
                throw new IllegalStateException("data publish failed, caused failed to notify majority, key=" + key);
            }

            long end = System.currentTimeMillis();
            Loggers.RAFT.info("signalPublish cost {} ms, key: {}", (end - start), key);
        } finally {
            OPERATE_LOCK.unlock();
        }
    }

// 注释：master 选举
GlobalExecutor.registerMasterElection(new MasterElection());
// 注释：节点间心跳
GlobalExecutor.registerHeartbeat(new HeartBeat());

public class MasterElection implements Runnable {
    @Override
    public void run() {
        try {

            if (!peers.isReady()) {
                return;
            }

            RaftPeer local = peers.local();
            // 注释：每次递减0.5s，直到小于0.5s，初始是15~20s的范围，这个任务0.5s执行一次，说明选举在没心跳最坏的情况
            // 是15~20s进行一次选举，即发投票。但是心跳任务每次都会重置这个 leaderDueMs 为 15s
            local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;

            if (local.leaderDueMs > 0) {
                return;
            }

            // reset timeout
            // 注释：重置选举时间间隔
            local.resetLeaderDue();
            // 注释：重置心跳时间间隔为5s
            local.resetHeartbeatDue();

            sendVote();
        } catch (Exception e) {
            Loggers.RAFT.warn("[RAFT] error while master election {}", e);
        }

    }
}

public class HeartBeat implements Runnable {
    @Override
    public void run() {
        try {

            if (!peers.isReady()) {
                return;
            }

            RaftPeer local = peers.local();
            // 注释：5s以上发一次心跳（初始随机0~5s）
            local.heartbeatDueMs -= GlobalExecutor.TICK_PERIOD_MS;
            if (local.heartbeatDueMs > 0) {
                return;
            }
            // 注释：重置心跳间隔
            local.resetHeartbeatDue();
            // 注释：发送心跳，里面重置了 leaderDueMs 时间，即当心跳一直正常时，不会发起master选举
            sendBeat();
        } catch (Exception e) {
            Loggers.RAFT.warn("[RAFT] error while sending beat {}", e);
        }

    }
}

WORD_SEPARATOR_CHAR = '~'
LINE_SEPARATOR_CHAR = '*'