一致性算法Raft的原理简介与源码初探

一、Raft算法背景

Raft是一种共识算法，旨在替代Paxos。它通过逻辑分离比Paxos更容易理解，但它也被正式证明是安全的，并提供了一些额外的功能（维基百科）。它通过日志复制来实现的一致性，提供了和(多重)Paxos 算法相同的功能和性能，但是它的算法结构和 Paxos 是不同的，因此Raft 算法更容易理解和应用。Raft 有几个关键模块：领导人选举、日志复制和安全性，同时它通过更强的一致性来减少算法状态的数量。从用户研究的结果可以证明，对于学生而言，Raft 算法比 Paxos 算法更容易学习。Raft 算法还允许集群成员的动态变更，它利用大多数原则来保证安全性。

二、与Paxos算法的比较

Paxos是著名的一致性算法（维基百科），微信也在其系统中有大量的使用（微信 PaxosStore：深入浅出 Paxos 算法协议），但有两个致命的缺点。

第一个是 Paxos 太难以理解。

它的完整的解释晦涩难懂；很少有人能完全理解，只有少数人成功的读懂了它。

第二个是它难以在实际环境中实现。

其中一个原因是，对于多决策 Paxos （multi-Paxos），大家还没有一个一致同意的算法。Lamport 的描述大部分都是有关于单决策 Paxos （single-decree Paxos）；他仅仅描述了实现多决策的可能的方法，缺少许多细节。有许多实现 Paxos 和优化 Paxos 的尝试，但是他们都和 Lamport 的描述有些出入。而且，Paxos 算法的结构也不是十分易于构建实践的系统；单决策分解也会产生其他的结果。例如，独立的选择一组日志条目然后合并成一个序列化的日志并没有带来太多的好处，仅仅增加了不少复杂性。围绕着日志来设计一个系统是更加简单高效的；新日志条目以严格限制的顺序增添到日志中去。
另一个问题是，Paxos 使用了一种对等的点对点的方式作为它的核心（尽管它最终提议了一种弱领导人的方法来优化性能）。在只有一个决策会被制定的简化世界中是很有意义的，但是很少有现实的系统使用这种方式。如果有一系列的决策需要被制定，首先选择一个领导人，然后让他去协调所有的决议，会更加简单快速。

三、Raft算法简介

Raft协议主要有三种角色：

Leader（领导者）
Follower（跟随者）
Candidate（候选人）

每个服务器节点会在这三种状态之间通过特定条件不断变换：
服务器状态
跟随者只响应来自其他服务器的请求。如果跟随者接收不到消息，那么他就会变成候选人并发起一次选举。获得集群中大多数选票的候选人将成为领导者。领导人一直都会是领导人直到自己宕机了。
Raft算法论文主要分为三部分进行描述论证：领导人选举、日志复制、安全性。

领导者选举

领导者选举主要按照如下原则进行。
其中上图的"Term"指代领导者的任期时长，Raft 把时间分割成任意长度的任期（如下图）。

任期用连续的整数标记。
每一段任期从一次选举开始，一个或者多个候选人尝试成为领导者。有可能存在一个任期会以没有领导人结束，但一个新的任期（和一次新的选举）会很快重新开始。
每个候选人、跟随者都有投票权，且每个投票者只能投票给那些日志与自身一样或更新的候选人（这称之为选举限制，与后面的日志复制与安全性论证有关）
获取多数投票的候选人成为当前任期的领导者，然后他会向其他的服务器发送心跳消息来建立自己的权威并且阻止新的领导人的产生。
Raft采用随机数的方式，使得服务器由“跟随者”变为“候选人”无规律性，从而加速领导人的选出。
Raft 保证了在一个给定的任期内，最多只有一个领导者。

日志复制

一旦一个领导人被选举出来，他就开始为客户端提供服务，且Raft算法主要靠日志执行。其中日志复制的主要原则如下：

客户端每一个请求都被领导人作为一条新的日志条目附加到日志中去，然后的发起附加条目 RPCs 给其他的服务器，让他们复制这条日志条目。只有领导者有权发送复制日志的指令给其他服务器。
在领导者将创建的日志条目复制到大多数的服务器上的时候，领导者就可以决定将其进行提交，只有被提交的日志才会被认为是有效的并最终返回给客户端。
在非领导者服务器上的未提交日志，可能被新领导者发送来的日志删除并覆盖，所以存在丢失未提交日志的情况。
领导者从来不会覆盖或者删除自己的日志，只会在变为跟随者时（通常因为失联、崩溃等异常情况后再恢复才会变为跟随者）才可能被覆盖未提交的日志。
只有领导人当前任期的日志条目才能通过计算数目来进行提交。一旦当前任期的日志条目以这种方式被提交，那么由于日志匹配原则（Log Matching Property），之前的日志条目也都会被间接的提交。（这称之为提交之前任期内的日志条目，与后面的安全性论证有关）

这个日志复制机制展示了Raft的一致性特性：只要大部分的服务器是正常的，Raft 能够接受、复制并且应用新的日志条目。在通常情况下，一条新的日志条目可以在一轮 RPC 内完成在集群的大多数服务器上的复制；并且一个速度很慢的追随者并不会影响整体的性能。

安全性

安全性的论证部分主要是通过以下两个前面提到限制规则来保证算法的安全性：
1.选举限制
2.提交之前任期内的日志条目
具体论证方式采用了反证法，感兴趣的同学可以参考原论文以及其译文。

一图了解Raft算法

点击打开动画图

四、Raft源码初探

Raft如此优秀，我们当然想了解它是如何实现的了，在java语言的实现中使用较广的要数 Atomix 的开源分布式容错框架了。

1.代码布局

从github（https://github.com/atomix/atomix ）上下载源码后，主要分如下几个模块

agent 代理机模块，主要用于部署服务时使用。
bin 生成 javadocs的脚本目录。
cluster 集群相关功能模块，包括集群节点定义、链接、集群间通讯协议等功能。
core 分布式框架的核心功能模块，包括各种在分布式情况下实现的一致性容器功能，如集合、列表、选举、取号、时钟等等功能。
dist 打包信息
docs 文档目录
primitive 各种基本数据类型与接口定义模块
protocols 协议功能模块，包括raft算法实现
rest 提供REST服务功能模块
storage 增强配置、存储日志功能模块
tests 测试模块
utils 其他工具类模块

2. 从测试入手看源码使用

Atomix的文档虽然不算详细，但其开源代码还是非常规范的，且提供了tests测试模块，这大大降低了我们上手的难度。直接从RaftPerformanceTest.java入手，经过分析主要分三部分：服务器端、客户端、代理执行。

2.1 创建服务器端

// 创建nodes个服务端
public void createServers(int nodes){
  List<RaftServer> servers = new ArrayList<>();
  CountDownLatch latch = new CountDownLatch(nodes);

  for (int i = 0; i < nodes; i++) {

   // 设置集群节点Member(包括 IP地址、端口)
   Address address = Address.from("localhost", ++port);
   Member member = Member.builder(MemberId.from(String.valueOf(++nextId)))
     .withAddress(address)
     .build();

   // 保存映射关系方便后续调用
   addressMap.put(member.id(), address);
   members.add(member);

   // 创建Raft服务端
   RaftServer server = createServer(members.get(i), Lists.newArrayList(members));
   server.bootstrap(members.stream().map(Member::id).collect(Collectors.toList())).thenRun(latch::countDown);
   servers.add(server);
  }
 }

 /**
  * Creates a Raft server.
  */
 private RaftServer createServer(Member member, List<Node> members) {

  // 创建Raft服务端协议（包括Netty集群协议初始化）
  RaftServerProtocol protocol;
  ManagedMessagingService messagingService;
  messagingService = (ManagedMessagingService) new NettyMessagingService("test", member.address(), new MessagingConfig())
    .start()
    .join();
  messagingServices.add(messagingService);
  protocol = new RaftServerMessagingProtocol(messagingService, protocolSerializer, addressMap::get);

  // 集群服务启动实例
  BootstrapService bootstrapService = new BootstrapService() {
   @Override
   public MessagingService getMessagingService() {
    return messagingService;
   }

   @Override
   public UnicastService getUnicastService() {
    return new UnicastServiceAdapter();
   }

   @Override
   public BroadcastService getBroadcastService() {
    return new BroadcastServiceAdapter();
   }
  };

  // Raft服务端Builder初始化
  RaftServer.Builder builder = RaftServer.builder(member.id())
    .withProtocol(protocol)
    .withThreadModel(ThreadModel.SHARED_THREAD_POOL)
    .withMembershipService(new DefaultClusterMembershipService(
      member,
      Version.from("1.0.0"),
      new DefaultNodeDiscoveryService(bootstrapService, member, new BootstrapDiscoveryProvider(members)),
      bootstrapService,
      new HeartbeatMembershipProtocol(new HeartbeatMembershipProtocolConfig())))
    .withStorage(RaftStorage.builder()
      .withStorageLevel(StorageLevel.DISK)
      .withDirectory(new File(String.format("target/perf-logs/%s", member.id())))
      .withNamespace(storageNamespace)
      .withMaxSegmentSize(1024 * 1024 * 64)
      .withDynamicCompaction()
      .withFlushOnCommit(false)
      .build());
  // 创建Raft服务端实例
  RaftServer server = builder.build();
  servers.add(server);
  return server;
 }

2.2 创建客户端

/**
  * Creates a Raft client.
  */
 private RaftClient createClient() throws Exception {
  // 设置客户端集群节点Member(包括 IP地址、端口)
  Address address = Address.from("localhost", ++port);
  Member member = Member.builder(MemberId.from(String.valueOf(++nextId)))
    .withAddress(address)
    .build();

  // 保存映射关系方便后续调用
  addressMap.put(member.id(), address);

  // 创建客户端Raft协议
  RaftClientProtocol protocol;
  MessagingService messagingService = new NettyMessagingService("test", member.address(), new MessagingConfig()).start().join();
  protocol = new RaftClientMessagingProtocol(messagingService, protocolSerializer, addressMap::get);

  // 创建Raft客户端
  RaftClient client = RaftClient.builder()
    .withMemberId(member.id())
    .withPartitionId(PartitionId.from("test", 1))
    .withProtocol(protocol)
    .withThreadModel(ThreadModel.SHARED_THREAD_POOL)
    .build();

  // 链接到Raft服务端
  client.connect(members.stream().map(Member::id).collect(Collectors.toList())).join();
  clients.add(client);
  return client;
 }

2.3 创建代理执行端

/**
  * Creates a test session.
  */
 private SessionClient createProxy(RaftClient client) {
  return client.sessionBuilder("raft-performance-test", TestPrimitiveType.INSTANCE, new ServiceConfig())
    .withReadConsistency(READ_CONSISTENCY)
    .withCommunicationStrategy(COMMUNICATION_STRATEGY)
    .build();
 }
 /**
  * Runs operations for a single Raft proxy.
  */
 private void runProxy(SessionClient proxy, CompletableFuture<Void> future) {
  int count = totalOperations.incrementAndGet();
  // 通过递归调用反复执行 TOTAL_OPERATIONS 次操作
  if (count > TOTAL_OPERATIONS) {
   future.complete(null);
  } else if (count % 2 < 1) {
   proxy.execute(operation(PUT, clientSerializer.encode(Maps.immutableEntry(UUID.randomUUID().toString(), UUID.randomUUID().toString()))))
     .whenComplete((result, error) -> {
      if (error == null) {
       writeCount.incrementAndGet();
      }
      runProxy(proxy, future);
     });
  } else {
   proxy.execute(operation(GET, clientSerializer.encode(UUID.randomUUID().toString()))).whenComplete((result, error) -> {
    if (error == null) {
     readCount.incrementAndGet();
    }
    runProxy(proxy, future);
   });
  }
 }

2.4 简化后的测试调用流程

public void run(){
  int serverNum = 3;
  int clientNum = 5;
  RaftClient[] clients = new RaftClient[clientNum];
  SessionClient[] proxies = new SessionClient[clientNum];
  CompletableFuture<Void>[] futures = new CompletableFuture[clientNum];
  try {
   // 创建服务端
   this.createServers(serverNum);

   // 创建客户端、代理服务端
   for (int i = 0; i < clientNum; i++) {
    clients[i] = createClient();
    proxies[i] = createProxy(clients[i]).connect().join();
    CompletableFuture<Void> future = new CompletableFuture<>();
    futures[i] = future;
   }

   // 通过代理执行操作
   long startTime = System.currentTimeMillis();
   for (int i = 0; i < clients.length; i++) {
    runProxy(proxies[i], futures[i]);
   }

   CompletableFuture.allOf(futures).join();
   long endTime = System.currentTimeMillis();
   long runTime = endTime - startTime;
   System.out.println(String.format("readCount: %d/%d, writeCount: %d/%d, runTime: %dms",
     readCount.get(),
     TOTAL_OPERATIONS,
     writeCount.get(),
     TOTAL_OPERATIONS,
     runTime));

  }catch (Exception e){
   System.err.println("error: "+e.getMessage());
   e.printStackTrace();
  }
 }

3. 执行看看效果：

// 调整以下2个参数，方便快速产生效果
// 执行操作次数
private static final int TOTAL_OPERATIONS = 1000;
// 读写操作比例
private static final int WRITE_RATIO = 5;

18:25:08.334 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5001
18:25:08.342 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:09.056 [raft-server-1] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{1} - Transitioning to FOLLOWER
18:25:09.066 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5002
18:25:09.066 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:09.086 [raft-server-2] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{2} - Transitioning to FOLLOWER
18:25:09.094 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5003
18:25:09.094 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:09.111 [raft-server-3] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{3} - Transitioning to FOLLOWER
18:25:09.891 [raft-server-1] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{1} - Transitioning to CANDIDATE
18:25:09.893 [raft-server-1] INFO  i.a.p.raft.roles.CandidateRole - RaftServer{1}{role=CANDIDATE} - Starting election
18:25:09.925 [raft-server-1] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{1} - Transitioning to LEADER
18:25:09.931 [raft-server-1] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{1} - Found leader 1
18:25:09.943 [raft-server-3] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{3} - Found leader 1
18:25:09.947 [raft-server-2] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{2} - Found leader 1
18:25:11.113 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5004
18:25:11.114 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:11.217 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5005
18:25:11.217 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:11.264 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5006
18:25:11.264 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:11.307 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5007
18:25:11.307 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
18:25:11.369 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - TCP server listening for connections on 0.0.0.0:5008
18:25:11.370 [netty-messaging-event-nio-server-0] INFO  i.a.c.m.impl.NettyMessagingService - Started
readCount: 500/1000, writeCount: 500/1000, runTime: 5894ms
Completed 1 iterations
averageRunTime: 5894ms
18:25:17.297 [raft-server-1] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{1} - Transitioning to INACTIVE
18:25:17.304 [raft-server-2] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{2} - Transitioning to INACTIVE
18:25:17.307 [raft-server-3] INFO  i.a.protocols.raft.impl.RaftContext - RaftServer{3} - Transitioning to INACTIVE
18:25:19.520 [ForkJoinPool.commonPool-worker-1] INFO  i.a.c.m.impl.NettyMessagingService - Stopped
18:25:19.533 [ForkJoinPool.commonPool-worker-3] INFO  i.a.c.m.impl.NettyMessagingService - Stopped
18:25:19.535 [ForkJoinPool.commonPool-worker-2] INFO  i.a.c.m.impl.NettyMessagingService - Stopped

Process finished with exit code -1

至此Raft算法源码已能初步执行了，要想进一步理解Raft算法就要继续啃源码了。虽然Raft号称容易理解与掌握，但总感觉这种容易是相对Paxos而言的吧~

参考文档

作者：侯嘉逊