7 Etcd服务端实现
7.1 Etcd启动
Etcd有多种启动方式，我们从最简单的方式入手，也就是从embed的etcd.go开始启动，最后会启动EtcdServer。
先看看etcd.go中的启动代码：
func StartEtcd(inCfg *Config) (e *Etcd, err error)从StartEtcd方法启动etcd服务，参数是初始配置信息config，启动集群间监听进程和客户端监听进程，最后启动EtcdServer。
主要代码：
e = &Etcd{cfg: *inCfg, stopc: make(chan struct{})}
cfg := &e.cfg
if e.Peers, err = startPeerListeners(cfg); err != nil {
return
}
if e.sctxs, err = startClientListeners(cfg); err != nil {
return
}
if e.Server, err = etcdserver.NewServer(srvcfg); err != nil {
return
}
e.Server.Start()startPeerListeners启动Peer监听，等待集群中其他机器连接自己。startClientListeners启动客户端监听Socket，等待客户端请求并响应。最后调用Start方法启动EtcdServer。
7.2 EtcdServer
EtcdServer位于etcdserver/server.go，定义了Server接口和EtcdServer对象。EtcdServer从逻辑上讲代表了一个完整的Etcd服务。
图7.1 Etcd服务端的功能示意图
Etcd服务端主要提供两大类客户端接口：
（1）集群配置
由memberHandler负责，提供添加集群成员，删除成员，更新成员信息三种接口服务。
（2）KV键值：由keysHandler负责。
KeysHandler接收到客户端请求后，调用EtcdServer的Do方法处理请求，Watcher类的客户端请求信息同样包含在keysHandler中了。KV键值响应主要在v2_server.go中定义，etcd新版本同时还提供了v3操作命令集，本文不讨论v3的源码实现。
7.2.1 接口定义
Etcdserver/server.go中定义了Server接口，是服务端的主接口，其中Do方法处理客户端请求。
Server.go中定义了EtcdServer对象，它是Server接口的实现类。Server中的Do接口是专门用来响应客户端请求的。
Server接口定义：
-
start
读取配置文件，启动本Server。
-
stop
停止本Server
-
ID
获取本节点server的ID，集群中所有的机器都有唯一ID，用于标识自己。
-
Leader
获取leader的ID
-
Do
处理客户群请求，返回处理结果。
定义：
func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error)在server.go中并没有看到Go接口的实现，其实它是在v2_server.go文件中定义的。
-
Process
Process(ctx context.Context, m raftpb.Message) error处理Raft消息。
-
AddMember
向Etcd集群中增加一台服务器，新增服务器的ID必须唯一标识。
-
RemoveMember
从集群删除一台服务器，删除服务器的ID必须已经存在于集群中。
-
UpdateMember
修改集群成员属性，如果成员ID不存在则返回ErrIDNotFound错误。
7.2.2 实体定义
EtcdServer表示一个独立运行的Etcd节点。
type EtcdServer struct {
inflightSnapshots int64
appliedIndex uint64
committedIndex uint64.
consistIndex consistentIndex
Cfg *ServerConfig
readych chan struct{}
r raftNode
snapCount uint64
w wait.Wait
readMu sync.RWMutex
readwaitc chan struct{}
readNotifier *notifier
stop chan struct{}
stopping chan struct{}
done chan struct{}
errorc chan error
id types.ID
attributes membership.Attributes
cluster *membership.RaftCluster
store store.Store
applyV2 ApplierV2
applyV3 applierV3
applyV3Base applierV3
applyWait wait.WaitTime
kv mvcc.ConsistentWatchableKV
lessor lease.Lessor
bemu sync.Mutex
be backend.Backend
authStore auth.AuthStore
alarmStore *alarm.AlarmStore
stats *stats.ServerStats
lstats *stats.LeaderStats
SyncTicker *time.Ticker
compactor *compactor.Periodic
peerRt http.RoundTripper
reqIDGen *idutil.Generator
forceVersionC chan struct{}
wgMu sync.RWMutex
wg sync.WaitGroup
ctx context.Context
cancel context.CancelFunc
leadTimeMu sync.RWMutex
leadElectedTime time.Time
}7.2.3 Do
Do定义在v2_server.go中，处理客户群请求包，调用raftNode的Propose方法。在上一章已经介绍过。
对于KV键值请求，Do方法是在etcdServer/v2_server.go中定义的，它的相关代码逻辑如下：
func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error) {
r.ID = s.reqIDGen.Next()
if r.Method == "GET" && r.Quorum {
r.Method = "QGET"
}
v2api := (v2API)(&v2apiStore{s})
switch r.Method {
case "POST":
return v2api.Post(ctx, &r)
case "PUT":
return v2api.Put(ctx, &r)
case "DELETE":
return v2api.Delete(ctx, &r)
case "QGET":
return v2api.QGet(ctx, &r)
case "GET":
return v2api.Get(ctx, &r)
case "HEAD":
return v2api.Head(ctx, &r)
}
return Response{}, ErrUnknownMethod
}可以看到对客户端的KV键值请求，最终是通过v2apiStore的相关方法来实现。客户端的命令前缀为"/v2/keys"。支持的命令有以下这些：
-
GET/QGET：读取键值
-
POST:创建一个新的KV键值
-
PUT：重新设置键值的值
-
DELETE：删除已有键值
v2apiStore包含了EtcdServer引用。
type v2apiStore struct{ s *EtcdServer }除了GET命令，其余Post，Put和Delete每个写操作请求最后都是通过processRaftRequest方法来处理的。
我们先看看GET命令的处理：
func (a *v2apiStore) Get(ctx context.Context, r *pb.Request) (Response, error) {
if r.Wait {
wc, err := a.s.store.Watch(r.Path, r.Recursive, r.Stream, r.Since)
if err != nil {
return Response{}, err
}
return Response{Watcher: wc}, nil
}
ev, err := a.s.store.Get(r.Path, r.Recursive, r.Sorted)
if err != nil {
return Response{}, err
}
return Response{Event: ev}, nil
}看到对于普通的GET操作，直接调用store.Get方法获取KV值返回给客户端，如果是Watcher操作，则返回Watcher给客户端，客户端后续通过Watcher接口读取变化值。
对于POST，PUT，DELETE命令，走下述Propose流程处理。
图7.2 Propose流程示意图
比如"DELETE"命令。
func (a *v2apiStore) Delete(ctx context.Context, r *pb.Request) (Response, error) {
return a.processRaftRequest(ctx, r)
}processRaftRequest方法的源码如下：
func (a *v2apiStore) processRaftRequest(ctx context.Context, r *pb.Request) (Response, error) {
data, err := r.Marshal()
if err != nil {
return Response{}, err
}
ch := a.s.w.Register(r.ID)
start := time.Now()
a.s.r.Propose(ctx, data)
proposalsPending.Inc()
defer proposalsPending.Dec()
select {
case x := <-ch:
resp := x.(Response)
return resp, resp.err
case <-ctx.Done():
proposalsFailed.Inc()
a.s.w.Trigger(r.ID, nil) // GC wait
return Response{}, a.s.parseProposeCtxErr(ctx.Err(), start)
case <-a.s.stopping:
}
return Response{}, ErrStopped
}-
data, err := r.Marshal()语句：
这条语句从pb.request得到请求数据data
-
ch := a.s.w.Register(r.ID)语句：
注册chain，一直等待直到ch有响应数据。
Register方法是wait的Register方法。该方法直到调用wait的Trigger方法后才会有数据从而触发select在该Register Id上线程被唤醒。Wait在pkg/wait中定义。
-
a.s.r.Propose(ctx, data)
Propose方法在node中定义，raftNode在etcdserver/node.go文件中。Propose将写事务请求发给Leader，等待集群间同步。Propose集群间同步消息完成后会唤醒a.s.w.Register语句。
调用raft/node的Propose方法处理写事务请求，进一步调用step方法将写事务封装成MsgProp消息并传递给集群中其他机器。
func (n *node) Propose(ctx context.Context, data []byte) error { return n.step(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}}) }step会调用StepFunc函数来处理MsgProp消息，根据leader，follower，candidate等运行状态分别调用不同的实现函数。
-
select语句
select … case …语句类似于Socket通信中的select语句，它的含义是只要任意一个case语句有数据返回就往下执行，否则就阻塞在这里让出CPU给其他线程执行。
case x := <-ch:当ch有值时，将ch赋值给x变量，同时唤醒case语句被执行，这里将执行以下代码：
resp := x.(Response) return resp, resp.err此时将ch中的返回结果Response回复给调用者（即客户端）。
case <-ctx.Done()：说明上下文被中断，Context的Done()被触发，此时写事务执行失败，返回空Response。
7.2.4 初始化
Etcd服务端主要由5大组件构成，他们的分工如下：
-
etcdServer：主进程，相当于整个Etcd的容器，包含了raftNode，WAL，snapshotter等多个关键组件。
-
raftNode：执行raft协议，保证写事务的集群一致性维护。
-
Store：管理维护Etcd数据库
-
Wal：管理事务日志
-
Snapshotter：负责数据快照，管理store数据库在内存中和磁盘上的相互转换。
raftNode除了负责集群间raft消息交互，还负责事务和快照的存储，保持数据一致性。
Etcd定义了一个storage数据结构，一起负责事务和快照。
type storage struct {
*wal.WAL
*snap.Snapshotter
}storage中没有指定WAL和Snapshotter的变量名称，这两个类的方法都可直接通过storage来调用，比如WAL的Save方法，可以通过storage.Save来调用，也可以通过storage.WAL.Save来调用，这两者是等价的，在阅读源码的时候要注意这一点，否则对Go语法不太了解的读者会感到迷惑。
func NewServer(cfg *ServerConfig) (srv *EtcdServer, err error) {
st := store.New(StoreClusterPrefix, StoreKeysPrefix)
var (
w *wal.WAL
n raft.Node
s *raft.MemoryStorage
id types.ID
cl *membership.RaftCluster
)
haveWAL := wal.Exist(cfg.WALDir())
ss := snap.New(cfg.SnapDir())
bepath := filepath.Join(cfg.SnapDir(), databaseFilename)
beExist := fileutil.Exist(bepath)
switch {
case haveWAL:
snapshot, err = ss.Load()
if snapshot != nil {
if err = st.Recovery(snapshot.Data); err != nil {
plog.Panicf("recovered store from snapshot error: %v", err)
}
}
cfg.Print()
if !cfg.ForceNewCluster {
id, cl, n, s, w = restartNode(cfg, snapshot)
} else {
id, cl, n, s, w = restartAsStandaloneNode(cfg, snapshot)
}
cl.SetStore(st)
cl.SetBackend(be)
cl.Recover(api.UpdateCapability)
}
if terr := fileutil.TouchDirAll(cfg.MemberDir()); terr != nil {
return