揭開 Raft 的神祕面紗，和ApacheRatis 瞭解Raft 組件的使用

以前寫過一些關於 Raft 原理的文章。javascript

圖解分佈式一致性算法
php

Raft協議詳解
css

相比 Paxos， Raft 一直以來就是以易於理解著稱。今天咱們以一年 Raft 使用者的角度，來看一下，別人根據 Raft 論文實現了以後，咱們通常要怎麼樣使用。java

俗話說，要想知道梨子的味道，就要親口嘗一嘗，沒吃過豬肉，也要見一見豬跑。不然別人再怎麼樣形容，你可能還覺得是像貓狗一類毛茸茸。nginx

在 Raft 官網裏長長的列表就能發現，實現 Raft 的框架目前很多。Java 裏我大概看了螞蟻的 SOFARaft 和 Apache 的 Ratis。此次咱們以 Ratis 爲例，揭開面紗，來看看到底要怎樣使用。git

固然，下面具體提到的例子，也是這些組件中自帶的 example。github

1、編譯web

github下載 Ratis 直接 mvn clean package 便可，若是編譯過程當中出錯，能夠先clean install ratis-proto算法

2、示例sql

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裏以參數的形式啓動。

3、分析

下面咱們來示例裏看下 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"));

4、內部部分實現

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來使用，下次直接鏈接了。