揭开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 的声明

 
 
 
 
    
  
  
  
  
  1. public final class CounterCommon { 
    2. 
  
  
  
  
  2.   public static final List PEERS = new ArrayList<>(3); 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4.   static { 
    5. 
  
  
  
  
  5.     PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n1"), "127.0.0.1:6000")); 
    6. 
  
  
  
  
  6.     PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n2"), "127.0.0.1:6001")); 
    7. 
  
  
  
  
  7.     PEERS.add(new RaftPeer(RaftPeerId.getRaftPeerId("n3"), "127.0.0.1:6002")); 
    8. 
  
  
  
  
  8.   } 
    9.

这里声明了三个节点。

通过命令行启动时，会直接把index 传进来, index 取值1-3。

 
 
 
 
    
  
  
  
  
  1. java -cp *.jar org.apache.ratis.examples.counter.server.CounterServer {serverIndex} 
    2.

然后在Server 启动的时候，拿到对应的配置信息。

 
 
 
 
    
  
  
  
  
  1. //find current peer object based on application parameter 
    2. 
  
  
  
  
  2.     RaftPeer currentPeer = 
    3. 
  
  
  
  
  3.         CounterCommon.PEERS.get(Integer.parseInt(args[0]) - 1); 
    4.

再设置存储目录

 
 
 
 
    
  
  
  
  
  1. //set the storage directory (different for each peer) in RaftProperty object 
    2. 
  
  
  
  
  2.     File raftStorageDir = new File("./" + currentPeer.getId().toString()); 
    3. 
  
  
  
  
  3.     RaftServerConfigKeys.setStorageDir(properties, 
    4. 
  
  
  
  
  4.         Collections.singletonList(raftStorageDir)) 
    5.

重点看这里，每个 Server 都会有一个状态机「CounterStateMachine」，平时我们的「业务逻辑」都放到这里

 
 
 
 
    
  
  
  
  
  1. //create the counter state machine which hold the counter value 
    2. 
  
  
  
  
  2.     CounterStateMachine counterStateMachine = new CounterStateMachine(); 
    3.

客户端发送的命令，会在这个状态机中被执行，同时这些命令又以Log 的形式复制给其它节点，各个节点的Log 又会在它自己的状态机里执行，从而保证各个节点状态的一致。

 

最后根据这些配置，生成 Raft Server 实例并启动。

 
 
 
 
    
  
  
  
  
  1. //create and start the Raft server 
    2. 
  
  
  
  
  2.     RaftServer server = RaftServer.newBuilder() 
    3. 
  
  
  
  
  3.         .setGroup(CounterCommon.RAFT_GROUP) 
    4. 
  
  
  
  
  4.         .setProperties(properties) 
    5. 
  
  
  
  
  5.         .setServerId(currentPeer.getId()) 
    6. 
  
  
  
  
  6.         .setStateMachine(counterStateMachine) 
    7. 
  
  
  
  
  7.         .build(); 
    8. 
  
  
  
  
  8.     server.start(); 
    9.

CounterStateMachine 里，应用计数的这一小段代码，我们看先检查了命令是否合法，然后执行命令

 
 
 
 
    
  
  
  
  
  1. //check if the command is valid 
    2. 
  
  
  
  
  2.     String logData = entry.getStateMachineLogEntry().getLogData() 
    3. 
  
  
  
  
  3.         .toString(Charset.defaultCharset()); 
    4. 
  
  
  
  
  4.     if (!logData.equals("INCREMENT")) { 
    5. 
  
  
  
  
  5.       return CompletableFuture.completedFuture( 
    6. 
  
  
  
  
  6.           Message.valueOf("Invalid Command")); 
    7. 
  
  
  
  
  7.     } 
    8. 
  
  
  
  
  8.     //update the last applied term and index 
    9. 
  
  
  
  
  9.     final long index = entry.getIndex(); 
    10. 
  
  
  
  
  10.     updateLastAppliedTermIndex(entry.getTerm(), index); 
    11. 
  
  
  
  
  11.  
    12. 
  
  
  
  
  12.     //actual execution of the command: increment the counter 
    13. 
  
  
  
  
  13.     counter.incrementAndGet(); 
    14. 
  
  
  
  
  14.  
    15. 
  
  
  
  
  15.     //return the new value of the counter to the client 
    16. 
  
  
  
  
  16.     final CompletableFuture f = 
    17. 
  
  
  
  
  17.         CompletableFuture.completedFuture(Message.valueOf(counter.toString())); 
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19.     //if leader, log the incremented value and it's log index 
    20. 
  
  
  
  
  20.     if (trx.getServerRole() == RaftProtos.RaftPeerRole.LEADER) { 
    21. 
  
  
  
  
  21.       LOG.info("{}: Increment to {}", index, counter.toString()); 
    22. 
  
  
  
  
  22.     } 
    23.

我们再来看 Client 的实现。

和 Server 类似，通过配置属性，创建一个实例

 
 
 
 
    
  
  
  
  
  1. private static RaftClient buildClient() { 
    2. 
  
  
  
  
  2.     RaftProperties raftProperties = new RaftProperties(); 
    3. 
  
  
  
  
  3.     RaftClient.Builder builder = RaftClient.newBuilder() 
    4. 
  
  
  
  
  4.         .setProperties(raftProperties) 
    5. 
  
  
  
  
  5.         .setRaftGroup(CounterCommon.RAFT_GROUP) 
    6. 
  
  
  
  
  6.         .setClientRpc( 
    7. 
  
  
  
  
  7.             new GrpcFactory(new Parameters()) 
    8. 
  
  
  
  
  8.                 .newRaftClientRpc(ClientId.randomId(), raftProperties)); 
    9. 
  
  
  
  
  9.     return builder.build(); 
    10. 
  
  
  
  
  10.   } 
    11.

然后就可以向Server发送命令开工了。

 
 
 
 
    
  
  
  
  
  1. raftClient.send(Message.valueOf("INCREMENT")); 
    2.

Counter 的状态机支持INCREMENT 和 GET 两个命令。所以example 最后执行了一个 GET 的命令来获取最终的计数结果

 
 
 
 
    
  
  
  
  
  1. RaftClientReply count = raftClient.sendReadOnly(Message.valueOf("GET")); 
    2.

四、内部部分实现

RaftClientImpl 里，初期会从peers列表中选一个，当成leader 去请求。

 
 
 
 
    
  
  
  
  
  1. RaftClientImpl(ClientId clientId, RaftGroup group, RaftPeerId leaderId, 
    2. 
  
  
  
  
  2.       RaftClientRpc clientRpc, RaftProperties properties, RetryPolicy retryPolicy) { 
    3. 
  
  
  
  
  3.     this.clientId = clientId; 
    4. 
  
  
  
  
  4.     this.clientRpc = clientRpc; 
    5. 
  
  
  
  
  5.     this.peers = new ConcurrentLinkedQueue<>(group.getPeers()); 
    6. 
  
  
  
  
  6.     this.groupId = group.getGroupId(); 
    7. 
  
  
  
  
  7.     this.leaderId = leaderId != null? leaderId 
    8. 
  
  
  
  
  8.         : !peers.isEmpty()? peers.iterator().next().getId(): null; 
    9. 
  
  
  
  
  9.     ... 
    10. 
  
  
  
  
  10.   } 
    11.

之后，会根据server 返回的不同异常分别处理。

 
 
 
 
    
  
  
  
  
  1. private RaftClientReply sendRequest(RaftClientRequest request) throws IOException { 
    2. 
  
  
  
  
  2.     RaftClientReply reply; 
    3. 
  
  
  
  
  3.     try { 
    4. 
  
  
  
  
  4.       reply = clientRpc.sendRequest(request); 
    5. 
  
  
  
  
  5.     } catch (GroupMismatchException gme) { 
    6. 
  
  
  
  
  6.       throw gme; 
    7. 
  
  
  
  
  7.     } catch (IOException ioe) { 
    8. 
  
  
  
  
  8.       handleIOException(request, ioe); 
    9. 
  
  
  
  
  9.     } 
    10. 
  
  
  
  
  10.     reply = handleLeaderException(request, reply, null); 
    11. 
  
  
  
  
  11.     reply = handleRaftException(reply, Function.identity()); 
    12. 
  
  
  
  
  12.     return reply; 
    13. 
  
  
  
  
  13.   } 
    14.

比如在 handleLeaderException 中，又分几种情况，因为通过Client 来和 Server 进行通讯的时候，会随机从peers里选择一个，做为leader去请求，如果 Server 返回异常，说它不是leader，就用下面的代码，随机从另外的peer里选择一个再去请求。

 
 
 
 
    
  
  
  
  
  1. final RaftPeerId oldLeader = request.getServerId(); 
    2. 
  
  
  
  
  2.     final RaftPeerId curLeader = leaderId; 
    3. 
  
  
  
  
  3.     final boolean stillLeader = oldLeader.equals(curLeader); 
    4. 
  
  
  
  
  4.     if (newLeader == null && stillLeader) { 
    5. 
  
  
  
  
  5.       newLeader = CollectionUtils.random(oldLeader, 
    6. 
  
  
  
  
  6.           CollectionUtils.as(peers, RaftPeer::getId)); 
    7. 
  
  
  
  
  7.     } 
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9.  static  T random(final T given, Iterable iteration) { 
    10. 
  
  
  
  
  10.     Objects.requireNonNull(given, "given == null"); 
    11. 
  
  
  
  
  11.     Objects.requireNonNull(iteration, "iteration == null"); 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13.     final List list = StreamSupport.stream(iteration.spliterator(), false) 
    14. 
  
  
  
  
  14.         .filter(e -> !given.equals(e)) 
    15. 
  
  
  
  
  15.         .collect(Collectors.toList()); 
    16. 
  
  
  
  
  16.     final int size = list.size(); 
    17. 
  
  
  
  
  17.     return size == 0? null: list.get(ThreadLocalRandom.current().nextInt(size)); 
    18. 
  
  
  
  
  18.   } 
    19.

是不是感觉很低效。如果这个时候，server 返回的信息里，告诉client 谁是 leader，那client 直接连上去就可以了是吧。

 
 
 
 
    
  
  
  
  
  1. /** 
    2. 
  
  
  
  
  2.    * @return null if the reply is null or it has 
    3. 
  
  
  
  
  3.    * {@link NotLeaderException} or {@link LeaderNotReadyException} 
    4. 
  
  
  
  
  4.    * otherwise return the same reply. 
    5. 
  
  
  
  
  5.    */ 
    6. 
  
  
  
  
  6.   RaftClientReply handleLeaderException(RaftClientRequest request, RaftClientReply reply, 
    7. 
  
  
  
  
  7.                                         Consumer handler) { 
    8. 
  
  
  
  
  8.     if (reply == null || reply.getException() instanceof LeaderNotReadyException) { 
    9. 
  
  
  
  
  9.       return null; 
    10. 
  
  
  
  
  10.     } 
    11. 
  
  
  
  
  11.     final NotLeaderException nle = reply.getNotLeaderException(); 
    12. 
  
  
  
  
  12.     if (nle == null) { 
    13. 
  
  
  
  
  13.       return reply; 
    14. 
  
  
  
  
  14.     } 
    15. 
  
  
  
  
  15.     return handleNotLeaderException(request, nle, handler); 
    16. 
  
  
  
  
  16.   }
    17. 
 
 
 
 
 
 
 
 
    
  
  
  
  
  1. RaftClientReply handleNotLeaderException(RaftClientRequest request, NotLeaderException nle, 
    2. 
  
  
  
  
  2.       Consumer handler) { 
    3. 
  
  
  
  
  3.     refreshPeers(nle.getPeers()); 
    4. 
  
  
  
  
  4.     final RaftPeerId newLeader = nle.getSuggestedLeader() == null ? null 
    5. 
  
  
  
  
  5.         : nle.getSuggestedLeader().getId(); 
    6. 
  
  
  
  
  6.     handleIOException(request, nle, newLeader, handler); 
    7. 
  
  
  
  
  7.     return null; 
    8. 
  
  
  
  
  8.   } 
    9.

我们会看到，在异常的信息中，如果能够提取出一个 suggestedLeader，这时候就会做为新的leaderId来使用，下次直接连接了。

本文转载自微信公众号「Tomcat那些事儿」，可以通过以下二维码关注。转载本文请联系Tomcat那些事儿公众号。

当前题目：揭开Raft的神秘面纱，和ApacheRatis了解Raft组件的使用
标题路径：http://www.shufengxianlan.com/qtweb/news28/39728.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

标签优化知识

各行业网站

马边网站建设　　 　瑞杰管理　　 　医药医疗网站建设方案　　 　成都定制网站　　 　成都众诚伟业　　 　成都包装设计公司　　 　成都标志设计　　 　达州做网站　　 　网站制作价格　　 　微信公众号开发　　 　成都400电话　　 　艾名斯线缆　　 　网站优化排名推广　　 　温江做网站　　 　成都商城app开发　　 　无边发光字制作　　 　宜宾网站设计　　 　娄底尾呼妄服装　　 　免备案空间　　 　网站建设改版