NATS设计与实现 #

https://github.com/nats-io/nats-sevrer

NATS 就是一个消息中间件，提供了 Pub/Sub 核心数据流，并基于此构建了 Request/Reply API 和 JetStream 用来提供可靠的分布式存储能力，和更高的 QoS（至少一次 + ACK）。

1. 核心概念 #

序号	名词（ENG）	名词（zh-CN）	解释
1	Publish	发布	发布动作，往 subject 中投递一条消息
2	Subscribe	订阅	订阅动作，表示想要接受发布相应 subject 的消息
3	Subject	主题	唯一标识，用于标识一种或者一类事件的概念。订阅时，可以使用 nats 约定的通匹配符来接收一类 subjects，如 orders.> 。
4	Core Nats	NATS 核心	CORE NATS 提供了以下能力： - PUB / SUB https://docs.nats.io/nats-concepts/core-nats/pubsub - Request / Reply https://docs.nats.io/nats-concepts/core-nats/reqreply - Queue Groups https://docs.nats.io/nats-concepts/core-nats/queue 提供 最多一次 的消息传递保证
5	Request / Reply	请求 / 回复	NATS 基于 PUB / SUB 实现的对 request 异步回复的功能，依赖于消息中的 reply 字段。reply 是内置实现，随机生成一个 “inbox” subject。 订阅者中间也是存在 Queue Group 概念的。 注意：发起 request 如果没有reply，那么服务端会返回 No Responders 消息。
6	Queue groups	队列组	订阅者使用同一个队列名称，它们就会成为一个队列组。每次队列组接收到消息时，只有队列组中_随机选择_的一个订阅者会消费一条消息，内置的负载均衡。
7	Message	消息 or 事件	一条消息包含了以下内容： - Subject. - A payload in the form of a byte array. - Any number of header fields. - An optional ‘reply’ address field.
8	Subject wildcards https://docs.nats.io/nats-concepts/subjects#wildcards	主题通配符	订阅者可以使用这些通配符_通过单个订阅收听多个主题。_ 注意：但发布者将始终使用完全指定的主题，而不使用通配符。
9	JetStream https://docs.nats.io/nats-concepts/jetstream	无 / 流	NATS 中一个功能特性，它是一个内置的分布式存储，在 CORE nats 的基础上扩展了更多的功能和更高的 QoS。功能上： - Stream：将 NATS 消息存储在流中，提供多种保留策略、限制、丢弃策略和主题映射功能。 - Consumer: 让客户端应用订阅或拉取流中的消息，支持多种重放策略、确认机制和流控功能 - Persistence: 将流的数据复制到多个 NATS 服务器，提供容错能力和加密存储功能 - KV Store: 可以将消息与键关联，提供存储、检索、删除和监听键值变化的功能。 - Object Store: JetStream 可以存储任意大小的对象（如文件），提供分块、校验和、元数据等功能。 其中最核心（开发常用）的概念就是：stream 和 consumer。
10	Stream https://docs.nats.io/nats-concepts/jetstream/streams	流	流即是消息存储，它定义了消息的存储方式以及保留的限制。 更具体的内容参见 #5.1 Stream 配置解析
11	Consumer	消费者	消费者作为客户端的接口，使用存储在流中的消息，跟踪客户端传递和确认的消息。Nats 同时支持 pull 和 push 两种消费模式；consumer 还提供了 durable 配置，用于持久化 consumer 消费信息（除非设置了 InactiveThreshold） 更具体的内容参见 #5.2 Consumer 配置解析
12	Replay / Redelivery
13	Raft Group https://docs.nats.io/running-a-nats-service/configuration/clustering/jetstream_clustering	RAFT 组	对于特定内容达成一致的分组，nats 中有 meta, stream, consumer 几种组。 Meta: 全部节点都是组成员。负责：JetStream API + 集群管理。 Stream: (根据 replicas 配置选择组内的服务器成员)。负责：stream 数据 Consumer: （根据 stream group的成员来确定消费者组内的成员）。负责：消费者状态
14	KV Store	键值存储
15	Object Store	对象存储

2. 软件架构（集群模式） #

Nats 自身是提供了多种运行方式:

• 单机 只运行了一个 nats 实例。

• 普通集群： 运行了 3 / 5 个实例，其中一个为集群 leader，此时集群如下：

  

• 超级集群： 存在多个集群，集群之间可以通过 网关 来传播消息_。_网关提供了三种传播机制：这里 A / B 分别代表两个集群。

  • Optimistic Mode 乐观模式

     当 A 中的发布者发布“foo”时，A 网关将检查集群 B 是否已注册对“foo”没有兴趣。如果没有，则将“foo”转发给B。如果B收到“foo”后，在“foo”上没有订阅者，则B将向A发送一条网关协议消息，表示它对“foo”没有兴趣，从而阻止将来的消息关于“foo”被转发。

      当后续 B 中有订阅者关注 “foo”, 那么 B 会发送一条网关协议，用来取消对 “foo” 没兴趣的设置。

  • Interest-only Mode 兴趣（关注）模式

      当 A 上的网关发送许多关于 B 不感兴趣的各种主题的消息时。 B 发送一条网关协议消息，要求 A 乐观地停止发送，如果已知对该主题感兴趣，则发送。随着 B 上的订阅不断增多，B 将向 A 更新其主题兴趣。

  • Queue Subscriptions 队列订阅模式

      服务器将始终首先尝试为本地队列订阅者提供服务，并且仅在未找到本地队列订阅者时进行故障转移。服务器将选择 RTT 最低的集群。

• 叶子结点/集群（代理模式）：

  透明地从本地客户端路由消息到一个或多个远程 NATS 系统。叶子结点采用本地的认证进行认证，连接远程 NATS 结点时，采用远程 NATS 系统的认证。通常用来降低 local 服务的延迟和流量。 注意：如果集群中一个节点配置为叶子结点，那么其余结点也要配置为叶子结点。

一般场景下最常用的就是集群模式。

Qs:

运维时怎么保证集群各节点的负载均衡？怎么保证集群的高可用？

3. Nats 客户端协议 #

https://docs.nats.io/reference/reference-protocols/nats-protocol

Nats 协议是一个_文本协议_ 这意味着，我们通过 telnet 就可以与之交互，通过抓包工具也可以很轻松的分析客户端和服务器之间的交互行为

从表里没有看到 Request 对不对？那是因为 Request 是基于 Pub / Sub 实现的 API，因此不在通信协议中，属于应用层的功能。

另外也没有看到 JetStream 相关的 Op 对不对？翻下代码就可以发现，它是基于 Request 机制实现的 pub，但通过前面的介绍，我们已经知道 CORE Nats 的 Reply 其实是 Subscriber 回复的，而 JetStream 是提供可靠存储的 ACK 并不能依赖于 Consumer ，所以这里服务器一定是有特殊处理。我们在 #6.5 JetStream 消息的投递和消费 一节中再详细展开。

4. NATS-Server 配置解析 #

https://docs.nats.io/running-a-nats-service/configuration

Nats 端口列表

5. JetStream 配置解析 #

Nats 的 jetstream 是通过 raft 来解决分布式一致性问题，以此实现 stream 相关的功能。

上图中展示了 jetstream 的一些关键点：

• Stream 可以存储多个 subject的消息。
• 消费者有多种消费模式（pull/push），还可以过滤stream 中的 subject 进行消费。
• 消费有多种确认模式（ack）

5.1 Stream 配置清单 #

https://docs.nats.io/nats-concepts/jetstream/streams#configuration

我们当前使用的 nats 版本对应 2.9.21 下表配置中，Metadata，compression, FirstSeq 和 SubjectTransform 暂不可用。

5.2 Consumer 配置清单 #

https://docs.nats.io/nats-concepts/jetstream/consumers

5.2.1 通用配置 #

5.2.2   拉模式 - 专属配置 #

5.2.3   推模式 - 专属配置 #

6. 关键流程 #

这里梳理了一些 Nats 的关键流程，用来理解 nats 中的设计和实现原理。

6.1  服务启动 / 集群 #

 集群 gossip 流程示意图：

nats-server -D -p 4222 -cluster nats://localhost:6222
nats-server -D -p 4333 -cluster nats://localhost:6333 -routes nats://localhost:6222
nats-server -D -p 4444 -cluster nats://localhost:6444 -routes nats://localhost:6333

  流程解释说明：

• nats-1 启动
• nats-2 连接到 nats-1, nats-1 回复 INFO 消息，并将 nats-2 添加到自己的路由，同时建立到 nats-2 的连接。
• nats-3 连接到 nats-2, nats-2 回复 INFO 消息，并将 nats-3 添加到自己的路由，同时建立到 nats-3 的连接。这里 nats-2 会将新路由（ nats-3 ）传播（INFO 消息）到自己已知的服务节点，也就是 nats-1, 而 nats-1 收到 INFO 消息后，会向 nats-3 建立连接，与前面的流程类似。

6.2 PUB / SUB #

  整体流程概览如下：

6.2.1  SUB #

  当客户端新增订阅时，会向服务器发送一条 SUB 消息，服务器则会更新客户端（及对应账户）的订阅关系，同时向其余集群中的 route (服务节点) 发送 RS+ （集群通讯协议）消息以更新订阅。

 6.2.2 PUB #

  客户端发布一条消息，服务器内部匹配（有缓存设计以提高匹配效率）相关订阅客户端。将消息加入到客户端的发送缓冲区（注意：订阅中区分 普通订阅和 队列，队列模式则需要随机选中一个客户端），当相应的 route 节点收到消息后，会再从本地的订阅列表中匹配当前节点中的客户端推送消息。

6.3  Request / Reply #

从 Nats 设计的协议已经知道，Request 和 Reply 两个 API 是基于 PUB / SUB 机制。PUB 参数中可以携带一个 reply 参数，该 reply 参数是发布时自动创建的一个 _INBOX subject。服务端在推送消息时，会将 reply 放到 MSG 消息头传递给订阅者。

// This processes the sublist results for a given message.  
// Returns if the message was delivered to at least target and queue filters.  
func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver, subject, reply []byte, flags int) (bool, [][]byte) {
	// ...
	
	var creply = reply
	
	// ...
	
	// Loop over all normal subscriptions that match.  
	for _, sub := range r.psubs {  
		...
		
	    // Normal delivery  
	    mh := c.msgHeader(dsubj, creply, sub)  
	    c.deliverMsg(prodIsMQTT, sub, acc, dsubj, creply, mh, msg, rplyHasGWPrefix)
	}
}

6.4  新建 jetstream #

在开始之前，我们需要对 multi-raft 有个大致的概念，示意图如下：

stream raft group 就对应上图中的一个 raft group。与此同时，集群中所有开启 jetstream 功能的节点还会组成一个 meta raft group, 用来管理 jestream API 的执行和 jetstream 集群拓扑（节点上线离线）。

从 nats-cli 的源码入手，我们可以找到：

// NewStreamFromDefault creates a new stream based on a supplied template and optionsfunc (m *Manager) NewStreamFromDefault(name string, dflt api.StreamConfig, opts ...StreamOption) (stream *Stream, err error) {  
	// ...
	
    var resp api.JSApiStreamCreateResponse  
    err = m.jsonRequest(fmt.Sprintf(api.JSApiStreamCreateT, name), &cfg, &resp)  
    if err != nil {  
       return nil, err  
    }  
  
    return m.streamFromConfig(&resp.Config, resp.StreamInfo), nil  
} 

Manager 的含义是 JetStreamManager，代表与服务端 JS API 交互，其中 api. JSApiStreamCreateT string = "$JS.API.STREAM.CREATE.%s" 可以发现底层还是通过 PUB / SUB 机制实现的交互。

在 nats-server 中检索可以发现，在 nats-server 内部定义了 JetStream 相关API 的 msgHandler。

func (s *Server) setJetStreamExportSubs() error {  
    // Start the go routine that will process API requests received by the  
    // subscription below when they are coming from routes, etc..    s.jsAPIRoutedReqs = newIPQueue[*jsAPIRoutedReq](s, "Routed JS API Requests")  
    s.startGoRoutine(s.processJSAPIRoutedRequests)  
  
    // This is the catch all now for all JetStream API calls.  
    if _, err := s.sysSubscribe(jsAllAPI, js.apiDispatch); err != nil {  
       return err  
    }  
  
	// ...
  
    // API handles themselves.  
    pairs := []struct {  
       subject string  
       handler msgHandler  
    }{  
       {JSApiAccountInfo, s.jsAccountInfoRequest},  
       {JSApiTemplateCreate, s.jsTemplateCreateRequest},  
       {JSApiTemplates, s.jsTemplateNamesRequest},  
       {JSApiTemplateInfo, s.jsTemplateInfoRequest},  
       {JSApiTemplateDelete, s.jsTemplateDeleteRequest},  
       {JSApiStreamCreate, s.jsStreamCreateRequest},  
       {JSApiStreamUpdate, s.jsStreamUpdateRequest},  
       {JSApiStreams, s.jsStreamNamesRequest},  
       {JSApiStreamList, s.jsStreamListRequest},  
       {JSApiStreamInfo, s.jsStreamInfoRequest},  
       {JSApiStreamDelete, s.jsStreamDeleteRequest},  
       {JSApiStreamPurge, s.jsStreamPurgeRequest},  
       {JSApiStreamSnapshot, s.jsStreamSnapshotRequest},  
       {JSApiStreamRestore, s.jsStreamRestoreRequest},  
       {JSApiStreamRemovePeer, s.jsStreamRemovePeerRequest},  
       {JSApiStreamLeaderStepDown, s.jsStreamLeaderStepDownRequest},  
       {JSApiConsumerLeaderStepDown, s.jsConsumerLeaderStepDownRequest},  
       {JSApiMsgDelete, s.jsMsgDeleteRequest},  
       {JSApiMsgGet, s.jsMsgGetRequest},  
       {JSApiConsumerCreateEx, s.jsConsumerCreateRequest},  
       {JSApiConsumerCreate, s.jsConsumerCreateRequest},  
       {JSApiDurableCreate, s.jsConsumerCreateRequest},  
       {JSApiConsumers, s.jsConsumerNamesRequest},  
       {JSApiConsumerList, s.jsConsumerListRequest},  
       {JSApiConsumerInfo, s.jsConsumerInfoRequest},  
       {JSApiConsumerDelete, s.jsConsumerDeleteRequest},  
    }  
  
    js.mu.Lock()  
    defer js.mu.Unlock()  
  
    for _, p := range pairs {  
       sub := &subscription{subject: []byte(p.subject), icb: p.handler}  
       if err := js.apiSubs.Insert(sub); err != nil {  
          return err  
       }  
    }  
  
    return nil  
}

在 client 消息分发(deliverMsg) 的逻辑中会对匹配的 subscription 执行 icb (msgHandler)。

在集群模式下，这些 JetStream API 的订阅也是会通过 route 客户端传导到集群中的每个节点。相当于每个节点都订阅/处理，JS API 的消息，但是只有 leader 节点可以处理。

jsStreamCreateRequest 后续代码逻辑（集群模式下, 单机只需要在本地操作即可）:

jsStreamCluster(jetstream meta group) 的启动路径： Reload / Start -> enableJetStream -> enableJetStreamClustering -> startRaftNode/setupMetaGroup -> monitorCluster

• 通过 meta raft group 提交一条新增 stream 的日志消息（其中已经设置 raftgroup 的基本信息）参见 jsClusteredStreamRequest 函数
• jetstream 集群中的节点接收到该日志后（monitorCluster），会执行（applyMetaEntries）其中的操作 （assignStreamOp）添加 stream 执行逻辑参见 processStreamAssignment -> processClusterCreateStream
  • 新建一个 stream raft group 启动 raft 节点
  • 内部创建一个 stream 数据结构（mset= stream）
  • 启动监视协程 monitorStream，处理 raft group 中的事件（快照，领导者变更，日志应用）

6.5  JetStream 消息的投递和消费 #

最简单的开始使用 JetStream 消费者的代码如下：

// connect to nats server
nc, _ := nats.Connect(nats.DefaultURL)

// create jetstream context from nats connection
js, _ := jetstream.New(nc)

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

// get existing stream handle
stream, _ := js.Stream(ctx, "foo")

// retrieve consumer handle from a stream
cons, _ := stream.Consumer(ctx, "cons")

// consume messages from the consumer in callback
cc, _ := cons.Consume(func(msg jetstream.Msg) {
    fmt.Println("Received jetstream message: ", string(msg.Data()))
    msg.Ack()
})
defer cc.Stop()

6.5.1 消费者创建 #

消费者可以显式或者隐式的创建，如果隐式创建往往都代表临时，也就是如果消费者不活跃则会被服务端删除。之后，客户端开始订阅消费者下发消息的 subject，push 方式等服务端推送，而 pull 模式则需要定时往服务端发送一条消息，触发消息下推。

从 nats.go 的源码出发，StreamConsumerManager 接口定义了一个 JetStream 中 Consumer 的相关操作。我们首先看下 consumer 的创建，同样的客户端是通过 JS API 来完成调用：

服务端统一使用了 jsConsumerCreateRequest 进行处理，其中逻辑与添加 stream 类似：jetstream集群模式下会创建一个raft group 设置好 leader, 并通过 meta group 来提交新增 consumer 的日志，每个节点会处理该消息（processConsumerAssignment->processClusterCreateConsumer）：

• 获取到流
• 通过 addConsumerWithAssignment 添加一个 consumer
• （leader）发送响应

这个操作会在 stream raft group 的 部分/全部（取决于 consumer 配置的 Replicas）节点中创建一个 raft group (consumer) 用来保存 consumer 的相关信息（消费点位）。同时在 stream 对应的节点中添加 consumer 实例。

再回到客户端，consumer 创建（查询）后，获取到 consume 的信息，从中获取到 delivery subject（代表 consumer 消费的消息会从这个 subject 推送出来），后续客户端会订阅这个这个 subject。

这里可以说明服务端 consumer 和 客户端 consumer 没有必然的的联系，后续另外的客户端使用同一个 consumer 信息时，也只是共享 consumer 的配置，订阅相同的 subject，回到了 CORE NATS 提供的数据模型。

6.5.2 消息消费 #

consumer 提供了两种模式 pull (PullSubscribe) 和 push(Subscribe) 模式，虽然叫做 pull 和 push，但是这两种机制的底层实现还是基于 nats 的 pub / sub 机制。push 等价于 subscribe，服务端会将消息直接推送到客户端；pull 则是客户端主动调用 JS API ”请求“，服务端 “响应” 数据给客户端消费。参见下图抓包数据分析截图（提前创建了 stream 和 consumer，指定了拉模式）

nats.go 的 jetstream 包针对 pull consumer 提供了丰富的API（尤其是 Consume）让 pull consumer 可以和 push consumer 用相似的方式来持续的处理消息。

6.5.2.1 pull 模式 #

pullConsumer Consume 方法实现如下（已精简），与 PullSubscribe.Fetch 的原理类似（请求 JS API），区别在于 API 的形式表现不同：

func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) {
	// 解析消费选项
	consumeOpts, err := parseConsumeOpts(false, opts...)
	if err != nil {
		return nil, fmt.Errorf("%w: %s", ErrInvalidOption, err)
	}
	// pull subject 是 JS.API 下的一个 subject 
	// apiRequestNextT string = "CONSUMER.MSG.NEXT.%s.%s"
	subject := apiSubj(p.jetStream.apiPrefix, fmt.Sprintf(apiRequestNextT, p.stream, p.name))
	sub := &pullSubscription{
		consumer:    p,
		errs:        make(chan error, 1),
		done:        make(chan struct{}, 1),
		fetchNext:   make(chan *pullRequest, 1),
		consumeOpts: consumeOpts,
	}
	
	// 创建一个 inbox subject 作为 pull 请求的 reply，internalHandler 就是 MsgHandler
	inbox := p.jetStream.conn.NewInbox()
	sub.subscription, err = p.jetStream.conn.Subscribe(inbox, internalHandler)
	if err != nil {
		return nil, err
	}
	go func() {
		for {
			select {
				case status, ok := <-sub.connStatusChanged:
				case err := <-sub.errs:
					if errors.Is(err, ErrNoHeartbeat) {
						  sub.fetchNext <- &pullRequest{  
						    Expires:   sub.consumeOpts.Expires,  
						    Batch:     batchSize,  
						    MaxBytes:  sub.consumeOpts.MaxBytes,  
						    Heartbeat: sub.consumeOpts.Heartbeat,  
						  }
					}
			}
		}
	}()
	// 根据 fetchNext 通知向服务端发送 pull 请求
	// fetchNext 是由一个 consumeOpts 的心跳设置的一个 timer 触发 ErrNoHeartbeat 来激活
	go sub.pullMessages(subject)
	
	return sub, nil
}

6.5.2.2 push 模式 #

使用 DeliverSubject (没有设置会自动生成一个) 来设置消费者订阅的主题，同时创建一个对该主题的订阅，用于接收处理消息。push 模式同样会在服务器创建消费者。

小结：对于客户端来说消费 stream 中的消息，其实就是产生一个订阅主题，客户端会如同普通的 SUB 客户端一样消费这里的消息。但在此之前，需要根据场景和选项在服务器创建一个 consumer，告诉服务端自己的消费场景（是否持久化，关注stream中的哪些 subject, 消息的分发策略 等等）

6.5.3 消息投递推送 #

通过前面的消息消费我们可以知道，pullConsumer 拉取消息时会往一个形如 CONSUMER.MSG.NEXT 的主题中发起请求，通过在服务端代码检索可以发现，consumer 结构内保存了这么一个 subject, 并关联了对应的 icb（processNextMsgReq），而这个动作是在 consumer 创建时，在 leader 节点上设置的（consumer.setLeader）。

这里要注意，这个请求只能被 consumer group 中的 leader 处理。

与此同时 consumer leader 节点还设置了（与消费相关的操作）:

• ack订阅、请求订阅 和 流控订阅等
• 如果消费者是推模式，那么会注册“兴趣通知”
• 同时会启动定时器来清除不活跃的消费者
• 启动消息推送的 loopAndGatherMsgs 逻辑，往 consumer.outq -> stream.outq 发送消息, stream.outq 在 stream.internalLoop 中执行往客户端推送的逻辑（通过 stream 的内部客户端往 subject 推送一条消息，与 pub 流程类似再由 CORE NATS 完成消息的分发）。

到这里总结下这个环节的问题：

• jetstream publish 的消息经过了什么样的流程？
  • 消息到服务端后会检查 subject 相关的订阅，而 stream(leader)设置时已经配置了一个内部订阅。订阅会传播到集群中。 processInboundJetStreamMsg 的目的就是将将消息提交到 raft group, 最终通过 processJetStreamMsg 执行。
  • stream 组提交后，各个副本保存，同时通知相关的 consumer
  • server consumer（leader） 通过 loopAndGatherMsgs 往 subject 中推送消息
  • server consumer 订阅了客户端 ack 消息，用于更新消费者位点等数据
• 是先保存还是先推送？ 在stream 组内部提交保存后推送
• consumer 的消息是从 leader 来还是可以从 follower 直接读取？ leader
• 服务端中 stream 和 consumer 这两个结构在其中承担什么样的职责？
  • stream 代表 stream （包括 store / raft group / stream config)
    • 负责 jetstream 相关API的请求处理
    • stream 消息 复制/存储
    • 通知/唤醒等待的 consumer
  • consumer 代表 消费者（包括消费配置 / raft group ） 注意它不等于客户端实例
    • consumer 组内的数据保存（raft group）
    • API 处理（拉消费的 API）
    • 消息推送
    • 消费确认处理

JetStream 消息投递API 也是基于 PUB / SUB 机制实现的，但是存在的区别是： JetStream 是存在确认机制的，而 Core Nats 的 Publish API 并没有。因此要严格保证消息发布到 stream 中，需要使用 JetStream 的 API 来发送。

7. 扩展 #

7.1 RAFT Consensus Protocol #

Consensus is a fundamental problem in fault-tolerant distributed systems. Consensus involves multiple servers agreeing on values. Once they reach a decision on a value, that decision is final.

共识问题是分布式系统容错中的基础性问题，它描述的是：在多个服务节点间对某个值达成一致，一旦达成一致那么这个值就是最终的结果。RAFT 就是解决共识问题的一种算法，它以简单著名。

常见的一种实现是，通过 复制/多副本 的方式来解决分布式问题中的 可靠性 问题。常见的，我们的系统提供了3个副本，当其中一个节点宕机时，也不影响系统对外提供服务。而 RAFT 则提供相应的机制（通过当选的领导者达成共识）来解决 复制/多副本 过程中的一致性问题。

Raft 可视化介绍：https://thesecretlivesofdata.com/raft/

它其中有几个概念：

• Leader Election（选举）产生唯一的一个 leader 角色。
  • Leader
  • Candidate
  • Follower
  • Vote
  • Term
• Log Replication（日志复制）用于服务器之间保持一致要素。
• Replicated State Machine（复制状态机）_每个服务器存储一个包含一系列指令的日志，并且按顺序执行指令。由于日志都包含相同顺序的指令，状态机会按照相同的顺序执行指令，由于状态机是确定的（deterministic），因此状态机会产生相同的结果。

7.2   Gossip Protocol #

它基于流行病传播方式的节点或者进程之间信息交换的协议。以给定的频率，每台计算机随机选择另一台计算机，并共享任何消息。

它的定义如下：

• 如果有某一项信息需要在整个网络中所有节点中传播，那从信息源开始，选择一个固定的传播周期（譬如 1 秒），随机选择它相连接的 k 个节点（称为 Fan-Out）来传播消息。
• 每一个节点收到消息后，如果这个消息是它之前没有收到过的，将在下一个周期内，选择除了发送消息给它的那个节点外的其他相邻 k 个节点发送相同的消息，直到最终网络中所有节点都收到了消息，尽管这个过程需要一定时间，但是理论上最终网络的所有节点都会拥有相同的消息。

它对网络节点的 连通性和稳定性 几乎没有任何要求，它一开始就将网络某些节点只能与一部分节点_部分连通（Partially Connected Network）而不是以全连通网络_（Fully Connected Network）作为前提；没有任何中心化节点或者主节点的概念。

相应的它的缺点是：无法准确地预计到需要多长时间才能达成全网一致；也存在重复传播的概率，消息在网路中冗余。

由此，Gossip 设计了两种可能的消息传播模式：反熵（Anti-Entropy）和传谣（Rumor-Mongering）：

• 反熵：会同步节点的全部数据，以消除各节点之间的差异，目标是整个网络各节点完全的一致。
• 传谣：仅仅发送新到达节点的数据，即只对外发送变更信息。

在 Nats 中，gossip 被用于实现 集群服务发现 https://docs.nats.io/reference/reference-protocols/nats-server-protocol。routes 里指定了一个种子服务器，新启动的服务器连接上种子服务器后，就可以获取到全部的服务器列表。在服务器的配置里，可以不用知道所有的节点，而只用配置一个即可，但通常为了配置简单，会统一使用一个种子服务器。如下：

nats-server -D -p 4222 -cluster nats://localhost:6222
nats-server -D -p 4333 -cluster nats://localhost:6333 -routes nats://localhost:6222
nats-server -D -p 4444 -cluster nats://localhost:6444 -routes nats://localhost:6222
// 使用下面的命令启动也是可行的
nats-server -D -p 4444 -cluster nats://localhost:6444 -routes nats://localhost:6333 

可以参见服务启动 / 集群的流程。

7.3   Zero allocation byte parser #

https://www.youtube.com/watch?v=ylRKac5kSOk&t=646s 零分配 的含义是：nats 在解析协议时，避免不必要的内存分配：

• 使用局部变量，在栈上分配

  

• 使用 slice 复用底层数组

  

• 采用状态机来解析协议的各个部分，而不是构建临时对象来存储中间状态。如: 一般情况下解析如下协议时，最常规的思路就是，先把 Command(PUB) 读出来，然后根据这个command 决定要后面的操作，这里再读去两个参数 subject, payload length, 有了 length 之后再读去 payload，这样我们就创建了_4个临时变量_在表示下面的命令。

  PUB foo.bar 7
  goodbye
  

  在 nats 中则是通过状态机，一步一步的解析中间没有使用临时变量来存储命令中的数据。

  

• 避免使用字符串，网络IO场景中，操作的对象几乎都是字节，因此 nats parser 也避免使用字符串以减少内存分配和拷贝。

7.4  Nats CLI #

https://docs.nats.io/using-nats/nats-tools/nats_cli 能方便快捷的与 nats 交互 / 模拟 / 测试。