揭開 Raft 的神秘面紗，和ApacheRatis 了解Raft 組件的使用
相比 Paxos， Raft 一直以來就是以易于理解著稱。今天我們以一年 Raft 使用者的角度，來看一下，別人根據 Raft 論文實現了之后，我們一般要怎么樣使用。
俗話說，要想知道梨子的味道，就要親口嘗一嘗，沒吃過豬肉，也要見一見豬跑。否則別人再怎么樣形容，你可能還以為是像貓狗一類毛茸茸。
在 Raft 官網里長長的列表就能發現，實現 Raft 的框架目前不少。Java 里我大概看了螞蟻的 SOFARaft 和 Apache 的 Ratis。這次我們以 Ratis 為例，揭開面紗，來看看到底要怎樣使用。
當然，下面具體提到的例子，也是這些組件中自帶的 example。
一、編譯
github下載 Ratis 直接 mvn clean package 即可，如果編譯過程中出錯，可以先clean install ratis-proto
二、示例
Ratis 自帶的示例有三個：
- arithmetic
- counter
- filestore
在 ratis-examples 模塊中，對于 arithmetic 和 filestore比較方便，可以通過main/bin目錄下的 shell 腳本快速啟動 Server 和 Client 來進行測試。
對于Raft，咱們都知道是需要多實例組成集群才能測試，你啟動一個實例沒啥用，連選主都成問題。Bin 目錄下的 start-all 支持 example 的名稱以及對應的命令。比如 filestore server 代表是啟動 filestore 這個應用的server。對應的命令參數會在相應example里的 cli 中解析。同時會一次性啟動三個server，組成一個集群并在周期內完成選舉。
而對于 counter 這個示例，并沒有相應的腳本來快速啟動三個server，這個我們可以通過命令行或者在IDE里以參數的形式啟動。
三、分析
下面我們來示例里看下 Raft Server 是怎樣工作的。
對于 counter 示例來說，我們啟動的時候，需要傳入一個參數，代表當前的server是第幾個，目的在于，要從 peers 列表中得知該用哪個IP + 端口去啟動它。這里我們能發現，這個 peers 列表，是在代碼內提前設置好的。當然你說動態配置啥的，也沒啥問題，另外兩個示例是通過shell 腳本里common 中的配置傳入的。
所以，第一步我們看到， Raft Server 在啟動的時候，會通過「配置」的形式，來知道 peer 之間的存在，這樣才能彼此通信，讓別人給自己投票或者給別人投票，完成 Term 內的選舉。另外，才能接收到 Leader 傳過來的 Log ，并且應用到本地。
第二步，我們來看下 Client 和 集群之間是如何通信的。整個 Raft 集群可能有多個實例，我們知道必須通過 Leader 來完成寫操作。那怎樣知道誰是Leader?有什么辦法?
一般常見的思路有：
- 在寫之前，先去集群內查一下，誰是 Leader，然后再寫
- 隨機拿一個寫，不行再換一個，不停的試，總會有一個成功。
當然方式二這樣試下去效率不太高。所以會在這個隨機試一次之后，集群會將當前的 Leader 信息返回給 Client，然后 Client 直接通過這個建立連接進行通信即可。
在 Ratis 里， Client 調用非 Leader 節點會收到 Server 拋出的一個異常，異常中會包含一個稱為 suggestLeader 的信息，表示當前正確的 Leader，按這個連上去就行。當然，如果如果在此過程中發生的 Leader 的變更，那就會有一個新的suggestLeader 返回來，再次重試。
我們來看 Counter 這個示例中的實現。
Server 和 Client 的共用的Common 代碼中，包含 peers 的聲明
- public final class CounterCommon {
- public static final List<RaftPeer> PEERS = new ArrayList<>(3);
- static {
- PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n1"), "127.0.0.1:6000"));
- PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n2"), "127.0.0.1:6001"));
- PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n3"), "127.0.0.1:6002"));
- }
這里聲明了三個節點。
通過命令行啟動時，會直接把index 傳進來, index 取值1-3。
- java -cp *.jar org.apache.ratis.examples.counter.server.CounterServer {serverIndex}
然后在Server 啟動的時候，拿到對應的配置信息。
- //find current peer object based on application parameter
- RaftPeer currentPeer =
- CounterCommon.PEERS.get(Integer.parseInt(args[0]) - 1);
再設置存儲目錄
- //set the storage directory (different for each peer) in RaftProperty object
- File raftStorageDir = new File("./" + currentPeer.getId().toString());
- RaftServerConfigKeys.setStorageDir(properties,
- Collections.singletonList(raftStorageDir))
重點看這里，每個 Server 都會有一個狀態機「CounterStateMachine」，平時我們的「業務邏輯」都放到這里
- //create the counter state machine which hold the counter value
- CounterStateMachine counterStateMachine = new CounterStateMachine();
客戶端發送的命令，會在這個狀態機中被執行，同時這些命令又以Log 的形式復制給其它節點，各個節點的Log 又會在它自己的狀態機里執行，從而保證各個節點狀態的一致。
最后根據這些配置，生成 Raft Server 實例并啟動。
- //create and start the Raft server
- RaftServer server = RaftServer.newBuilder()
- .setGroup(CounterCommon.RAFT_GROUP)
- .setProperties(properties)
- .setServerId(currentPeer.getId())
- .setStateMachine(counterStateMachine)
- .build();
- server.start();
CounterStateMachine 里，應用計數的這一小段代碼，我們看先檢查了命令是否合法，然后執行命令
- //check if the command is valid
- String logData = entry.getStateMachineLogEntry().getLogData()
- .toString(Charset.defaultCharset());
- if (!logData.equals("INCREMENT")) {
- return CompletableFuture.completedFuture(
- Message.valueOf("Invalid Command"));
- }
- //update the last applied term and index
- final long index = entry.getIndex();
- updateLastAppliedTermIndex(entry.getTerm(), index);
- //actual execution of the command: increment the counter
- counter.incrementAndGet();
- //return the new value of the counter to the client
- final CompletableFuture<Message> f =
- CompletableFuture.completedFuture(Message.valueOf(counter.toString()));
- //if leader, log the incremented value and it's log index
- if (trx.getServerRole() == RaftProtos.RaftPeerRole.LEADER) {
- LOG.info("{}: Increment to {}", index, counter.toString());
- }
我們再來看 Client 的實現。
和 Server 類似，通過配置屬性，創建一個實例
- private static RaftClient buildClient() {
- RaftProperties raftProperties = new RaftProperties();
- RaftClient.Builder builder = RaftClient.newBuilder()
- .setProperties(raftProperties)
- .setRaftGroup(CounterCommon.RAFT_GROUP)
- .setClientRpc(
- new GrpcFactory(new Parameters())
- .newRaftClientRpc(ClientId.randomId(), raftProperties));
- return builder.build();
- }
然后就可以向Server發送命令開工了。
- raftClient.send(Message.valueOf("INCREMENT"));
Counter 的狀態機支持INCREMENT 和 GET 兩個命令。所以example 最后執行了一個 GET 的命令來獲取最終的計數結果
- RaftClientReply count = raftClient.sendReadOnly(Message.valueOf("GET"));
四、內部部分實現
RaftClientImpl 里，初期會從peers列表中選一個，當成leader 去請求。
- RaftClientImpl(ClientId clientId, RaftGroup group, RaftPeerId leaderId,
- RaftClientRpc clientRpc, RaftProperties properties, RetryPolicy retryPolicy) {
- this.clientId = clientId;
- this.clientRpc = clientRpc;
- this.peers = new ConcurrentLinkedQueue<>(group.getPeers());
- this.groupId = group.getGroupId();
- this.leaderId = leaderId != null? leaderId
- : !peers.isEmpty()? peers.iterator().next().getId(): null;
- ...
- }
之后，會根據server 返回的不同異常分別處理。
- private RaftClientReply sendRequest(RaftClientRequest request) throws IOException {
- RaftClientReply reply;
- try {
- reply = clientRpc.sendRequest(request);
- } catch (GroupMismatchException gme) {
- throw gme;
- } catch (IOException ioe) {
- handleIOException(request, ioe);
- }
- reply = handleLeaderException(request, reply, null);
- reply = handleRaftException(reply, Function.identity());
- return reply;
- }
比如在 handleLeaderException 中，又分幾種情況，因為通過Client 來和 Server 進行通訊的時候，會隨機從peers里選擇一個，做為leader去請求，如果 Server 返回異常，說它不是leader，就用下面的代碼，隨機從另外的peer里選擇一個再去請求。
- final RaftPeerId oldLeader = request.getServerId();
- final RaftPeerId curLeader = leaderId;
- final boolean stillLeader = oldLeader.equals(curLeader);
- if (newLeader == null && stillLeader) {
- newLeader = CollectionUtils.random(oldLeader,
- CollectionUtils.as(peers, RaftPeer::getId));
- }
- static <T> T random(final T given, Iterable<T> iteration) {
- Objects.requireNonNull(given, "given == null");
- Objects.requireNonNull(iteration, "iteration == null");
- final List<T> list = StreamSupport.stream(iteration.spliterator(), false)
- .filter(e -> !given.equals(e))
- .collect(Collectors.toList());
- final int size = list.size();
- return size == 0? null: list.get(ThreadLocalRandom.current().nextInt(size));
- }
是不是感覺很低效。如果這個時候，server 返回的信息里，告訴client 誰是 leader，那client 直接連上去就可以了是吧。
- /**
- * @return null if the reply is null or it has
- * {@link NotLeaderException} or {@link LeaderNotReadyException}
- * otherwise return the same reply.
- */
- RaftClientReply handleLeaderException(RaftClientRequest request, RaftClientReply reply,
- Consumer<RaftClientRequest> handler) {
- if (reply == null || reply.getException() instanceof LeaderNotReadyException) {
- return null;
- }
- final NotLeaderException nle = reply.getNotLeaderException();
- if (nle == null) {
- return reply;
- }
- return handleNotLeaderException(request, nle, handler);
- }
- RaftClientReply handleNotLeaderException(RaftClientRequest request, NotLeaderException nle,
- Consumer<RaftClientRequest> handler) {
- refreshPeers(nle.getPeers());
- final RaftPeerId newLeader = nle.getSuggestedLeader() == null ? null
- : nle.getSuggestedLeader().getId();
- handleIOException(request, nle, newLeader, handler);
- return null;
- }
我們會看到，在異常的信息中，如果能夠提取出一個 suggestedLeader，這時候就會做為新的leaderId來使用，下次直接連接了。
本文轉載自微信公眾號「Tomcat那些事兒」，可以通過以下二維碼關注。轉載本文請聯系Tomcat那些事兒公眾號。
