從本篇開始，將會有一個系列對Etcd原始碼進行分析。

我之前閱讀過很多開源軟體，對閱讀開源軟體，有如下基本思路：

1、瞭解該軟體相關背景知識，例如相關部落格、官網，要相信自己不是第一個分析該軟體的人

2、對該軟體進行使用，例如：編譯、執行或者基於介面進行開發

3、找到該軟體的合適切入點進行原始碼分析，例如網路相關的軟體（ovs、etcd）找到socket監聽服務、接收訊息、傳送訊息

對於etcd這種分散式儲存系統，我們的切入點就是socket服務。

一、配置檔案

1.1、配置檔案簡介

    研究一個軟體最好的方式就是先把配置檔案搞清楚，這個是根基。我們看一下etcd的配置檔案，預設儲存位置為/etc/etcd/etcd.conf，該配置檔案一共有5個section分別是：

名稱	作用
member	本節點的配置，包括監聽服務埠、心跳時間等
cluster	叢集配置，包括叢集狀態、叢集名稱以及本節點廣播地址
proxy	用於網路自動發現服務
security	安全配置
logging	日誌功能元件

其中配置檔案中比較重要的是member和cluster配置項。

1.2、部分引數詳細說明

1.2.1、member

    在Etcd中一共監聽了兩個服務埠分別2379和2380，對應在member中體現的如下兩個配置項：

名稱	舉例	作用
ETCD_LISTEN_CLIENT_URLS	ETCD_LISTEN_CLIENT_URLS="http://0.0.0.0:2379"	2379埠用於客戶端訪問etcd資料庫，例如向etcd中插入資料。
ETCD_LISTEN_PEER_URLS	ETCD_LISTEN_PEER_URLS="http://0.0.0.0:2380"	2380埠用於叢集成員間進行通訊，例如叢集資料同步、心跳。

1.2.2、cluster

初次看到配置檔案，都會有一個疑問，為什麼在members已經設定了監聽服務地址，為什麼在cluster還要再次設定一次廣播地址呢？原因：etcd主要的通訊協議主要是http協議，對於http協議中所周知它是B/S結構，而非C/S結構，只能一端主動給另一端發訊息而反過來則不可。所以對於叢集來說，雙方必須都要知道對方具體監聽地址。

名稱	舉例	作用
ETCD_ADVERTISE_CLIENT_URLS	ETCD_LISTEN_CLIENT_URLS="http://10.10.10.128:2379"	同上表
ETCD_INITIAL_ADVERTISE_PEER_URLS	ETCD_INITIAL_ADVERTISE_PEER_URLS="http://10.10.10.128:2380"	同上表

叢集相關配置，在後面介紹Raft叢集時再進行詳細說明。

二、服務監聽

從本小節開始介紹詳細程式碼。我們都知道，建立socket服務端一共有5個基本步驟（C語言）：建立socket套接字、bind地址及埠、listen監聽服務、accept接收客戶端連線、啟動新執行緒為客戶端服務。正所謂萬變不離其宗，到了etcd中（etcd使用預設golang http模組）也是這些步驟，只不過是被封裝了一下（語法糖）。

2.1、總體流程圖

從這裡開始介紹具體流程以及關鍵程式碼，對於資料結構會有專門一篇介紹，如下是從main方法入口函式：

流程圖簡要說明：

    1、方法startEtcdOrProxyV2()中，會根據配置檔案啟動兩種不同模式：預設模式和代理模式。預設模式進入方法startEtcd，代理模式進入方法startProxy。這裡介紹預設模式。

    2、方法newConfig，從名字上來看就知道，此方法是用於讀取配置檔案或者生成預設配置，在1.2章節中介紹的配置項就是在此方法中讀取。

    3、執行完方法serve後，會退回到startEtcd中，然後就阻塞在startEtcd方法中，這樣整個etcd啟動完畢。

2.2、核心方法embed/etcd.go StartEtcd

此方法內容比較長，分為三部分說明：

       /*
	* 構建Etcd結構 包含一個server、listener
	*/
	serving := false
	e = &Etcd{cfg: *inCfg, stopc: make(chan struct{})} //etcd結構體
	cfg := &e.cfg
	defer func() {//解構函式
		if e == nil || err == nil {
			return
		}
		if !serving {
			// errored before starting gRPC server for serveCtx.grpcServerC
			for _, sctx := range e.sctxs {
				close(sctx.grpcServerC)
			}
		}
		e.Close()
		e = nil
	}()

	if e.Peers, err = startPeerListeners(cfg); err != nil {//為peer建立listener，socket三部曲只到了第二個步驟
		return
	}
	if e.sctxs, err = startClientListeners(cfg); err != nil {//為client建立listener，socket三部曲只到了第二個步驟
		return
	}
	for _, sctx := range e.sctxs {
		e.Clients = append(e.Clients, sctx.l)
	}

上面完成etcd結構初始化以及listener建立。

// 建立EtcdServer

srvcfg := &etcdserver.ServerConfig{
		Name:                cfg.Name,
		ClientURLs:           cfg.ACUrls, //客戶端url監聽地址，2379埠
		PeerURLs:            cfg.APUrls, //peer url監聽地址，2380埠
		DataDir:              cfg.Dir,
		DedicatedWALDir:     cfg.WalDir,
		SnapCount:           cfg.SnapCount,
		MaxSnapFiles:        cfg.MaxSnapFiles,
		MaxWALFiles:        cfg.MaxWalFiles,
		InitialPeerURLsMap:   urlsmap,
		InitialClusterToken:     token,
		DiscoveryURL:            cfg.Durl,
		DiscoveryProxy:          cfg.Dproxy,
		NewCluster:              cfg.IsNewCluster(),  /* 是否新的叢集 */
		ForceNewCluster:         cfg.ForceNewCluster,
		PeerTLSInfo:             cfg.PeerTLSInfo,
		TickMs:                  cfg.TickMs,
		ElectionTicks:           cfg.ElectionTicks(),
		AutoCompactionRetention: cfg.AutoCompactionRetention,
		QuotaBackendBytes:       cfg.QuotaBackendBytes,
		StrictReconfigCheck:     cfg.StrictReconfigCheck,
		ClientCertAuthEnabled:   cfg.ClientTLSInfo.ClientCertAuth,
		AuthToken:               cfg.AuthToken,
	}
	//建立EtcdServer並且建立raftNode並執行raftNode
	if e.Server, err = etcdserver.NewServer(srvcfg); err != nil {
		return
	}

	// configure peer handlers after rafthttp.Transport started
	// 生成http.hander 用於處理peer請求
	ph := etcdhttp.NewPeerHandler(e.Server) 
	for _, p := range e.Peers {
		srv := &http.Server{
			Handler:     ph,
			ReadTimeout: 5 * time.Minute,
			ErrorLog:    defaultLog.New(ioutil.Discard, "", 0), // do not log user error
		}

		l := p.Listener //上一段程式碼建立的listener
		p.serve = func() error { return srv.Serve(l) } //回撥函式，啟用服務，主要是Accept方法
		p.close = func(ctx context.Context) error {關閉服務，回撥掉函式。即socket關閉時呼叫此方法
			// gracefully shutdown http.Server
			// close open listeners, idle connections
			// until context cancel or time-out
			return srv.Shutdown(ctx)
		}
	}
上面handler用於處理peer發過來的請求以及設定回撥函式。
	// buffer channel so goroutines on closed connections won't wait forever
	e.errc = make(chan error, len(e.Peers)+len(e.Clients)+2*len(e.sctxs))

	//執行EtcdSever 監聽服務
	e.Server.Start()
	if err = e.serve(); err != nil {//啟用服務，主要呼叫第二段程式碼的回撥函式serve
		return
	}
	serving = true

下面重點分析一下listerner以及serve回撥函式。

2.3、Listener分析

Listener有兩個分別為：peer listener和client listener，兩者大同小異，這裡拿peer listener做為分析物件。

方法startPeerListeners，中主要核心程式碼，如下：

for i, u := range cfg.LPUrls {//迴圈遍歷多個peer url
		if u.Scheme == "http" {
			if !cfg.PeerTLSInfo.Empty() {
				plog.Warningf("The scheme of peer url %s is HTTP while peer key/cert files are presented. Ignored peer key/cert files.", u.String())
			}
			if cfg.PeerTLSInfo.ClientCertAuth {
				plog.Warningf("The scheme of peer url %s is HTTP while client cert auth (--peer-client-cert-auth) is enabled. Ignored client cert auth for this url.", u.String())
			}
		}
		/* 構造peerListener物件 監聽2380 作為服務端模式 */
		peers[i] = &peerListener{close: func(context.Context) error { return nil }} 
		//呼叫介面，建立listener物件，返回來之後，
//socket套接字已經完成listener監聽流程
peers[i].Listener, err = rafthttp.NewListener(u, &cfg.PeerTLSInfo) 
		if err != nil {
			return nil, err
		}
		// once serve, overwrite with 'http.Server.Shutdown'
        // close回撥方法，用於關閉socket套接字
		peers[i].close = func(context.Context) error {
			return peers[i].Listener.Close()
		}
		plog.Info("listening for peers on ", u.String())
	}
func newListener(addr string, scheme string) (net.Listener, error) {
	if scheme == "unix" || scheme == "unixs" {
		// unix sockets via unix://laddr
		return NewUnixListener(addr)
	}
	return net.Listen("tcp", addr) //呼叫golang內建方法，返回listener物件
}

    從startPeerListeners到net.Listen整個流程中，有摻雜tls以及unix socket相關邏輯，新增這些只為了保證各種需求，大體流程並沒有變化，這裡不在對齊進行詳細說明。至此，兩個服務均已完成監聽步驟，下面就是接收對端請求即Accept過程。

三、服務啟用

    在上面已經介紹了，服務端socket需要呼叫Accept方法，我們來看一下serve方法。方法serve大致內容為：將每個服務放到gorouting中，也就是啟動一個協程來監聽服務。

func (e *Etcd) serve() (err error) {
	var ctlscfg *tls.Config
	if !e.cfg.ClientTLSInfo.Empty() {
		plog.Infof("ClientTLS: %s", e.cfg.ClientTLSInfo)
		if ctlscfg, err = e.cfg.ClientTLSInfo.ServerConfig(); err != nil {
			return err
		}
	}

	if e.cfg.CorsInfo.String() != "" {
		plog.Infof("cors = %s", e.cfg.CorsInfo)
	}

	// Start the peer server in a goroutine
    // 為Peer啟動協程
	for _, pl := range e.Peers {
		go func(l *peerListener) {
			// 叢集peer 前期已經建立listener，此處將會呼叫accept，
// 那麼serve()是什麼地方定義的？
e.errHandler(l.serve())  
		}(pl)
	}

	// Start a client server goroutine for each listen address
// 為Client啟動協程
	var h http.Handler
	if e.Config().EnableV2 {
		h = v2http.NewClientHandler(e.Server, e.Server.Cfg.ReqTimeout())
	} else {
		mux := http.NewServeMux()
		etcdhttp.HandleBasic(mux, e.Server)
		h = mux
	}
	h = http.Handler(&cors.CORSHandler{Handler: h, Info: e.cfg.CorsInfo})

	for _, sctx := range e.sctxs {
		go func(s *serveCtx) {
			// client 前期已經建立listener，此處將呼叫accept 
e.errHandler(s.serve(e.Server, ctlscfg, h, e.errHandler)) 
		}(sctx)
	}
	return nil
}

    那麼在l.serve()或者s.serve()在什麼地方定義的？在方法StartEtcd中生成http.PeerHandler附近，在第二章節已經有介紹。我們來看一下具體內容：

srv := &http.Server{
	Handler:     ph, //http handler 用於處理業務邏輯 後面呼叫handler中ServeHTTP方法
	ReadTimeout: 5 * time.Minute,
	ErrorLog:    defaultLog.New(ioutil.Discard, "", 0), // do not log user error
}
l := p.Listener
p.serve = func() error { return srv.Serve(l) } //啟用服務，這地方呼叫的srv.Serve(l)，此方法呼叫golang內建方法http.Server.

下面簡要分析一下http.Server方法：

func (srv *Server) Serve(l net.Listener) error {
	defer l.Close()
	if fn := testHookServerServe; fn != nil {
		fn(srv, l)
	}
	var tempDelay time.Duration // how long to sleep on accept failure

	if err := srv.setupHTTP2_Serve(); err != nil {
		return err
	}

	srv.trackListener(l, true)
	defer srv.trackListener(l, false)

	baseCtx := context.Background() // base is always background, per Issue 16220
	ctx := context.WithValue(baseCtx, ServerContextKey, srv)
	for {//死迴圈
		rw, e := l.Accept()  //阻塞等待客戶端程式連線
    ....
		tempDelay = 0
		c := srv.newConn(rw)
		c.setState(c.rwc, StateNew) // before Serve can return
		go c.serve(ctx) //表示啟動協程，服務新的連線。
	}
}

此處的Accept定義在什麼地方呢？Accept為golang內建方法，定義在Go/src/net/tcpsock.go，對於Accept內容不再展開，只要知道它會一直阻塞，直到有新的連線才會返回。

這裡著重介紹一下http/server.go中serve方法。在serve方法內部主要內容是一個for迴圈：

for {
		w, err := c.readRequest(ctx) //接收到對端傳送的http請求
		if c.r.remain != c.server.initialReadLimitSize() {
			// If we read any bytes off the wire, we're active.
			c.setState(c.rwc, StateActive)
		}
    ...
		// Expect 100 Continue support
		req := w.req
		if req.expectsContinue() {
			if req.ProtoAtLeast(1, 1) && req.ContentLength != 0 {
				// Wrap the Body reader with one that replies on the connection
				req.Body = &expectContinueReader{readCloser: req.Body, resp: w}
			}
		} else if req.Header.get("Expect") != "" {
			w.sendExpectationFailed()
			return
		}

		c.curReq.Store(w)

		if requestBodyRemains(req.Body) {
			registerOnHitEOF(req.Body, w.conn.r.startBackgroundRead)
		} else {
			if w.conn.bufr.Buffered() > 0 {
				w.conn.r.closeNotifyFromPipelinedRequest()
			}
			w.conn.r.startBackgroundRead()
		}

		// HTTP cannot have multiple simultaneous active requests.[*]
		// Until the server replies to this request, it can't read another,
		// so we might as well run the handler in this goroutine.
		// [*] Not strictly true: HTTP pipelining. We could let them all process
		// in parallel even if their responses need to be serialized.
		// But we're not going to implement HTTP pipelining because it
		// was never deployed in the wild and the answer is HTTP/2.
		serverHandler{c.server}.ServeHTTP(w, w.req) //處理http請求
		w.cancelCtx()
		if c.hijacked() {
			return
		}
		w.finishRequest() //將緩衝區資料傳送到對端，完成http此次請求
		...
	}

呼叫c.readRequest(ctx)方法等待對端發來的請求，呼叫serverHandler{c.server}.ServeHTTP(w,w.req)處理客戶端請求並且傳送到對端。來看一下ServeHTTP實現：

func (sh serverHandler) ServeHTTP(rw ResponseWriter, req *Request) {
	handler := sh.srv.Handler
	if handler == nil {
		handler = DefaultServeMux
	}
	if req.RequestURI == "*"&& req.Method == "OPTIONS" {
		handler = globalOptionsHandler{}
	}
	handler.ServeHTTP(rw, req)
}

會發現在內部會呼叫handler.ServeHTTP方法，那麼此處的handler在什麼地方定義的？ServeHTTP又在哪裡定義的？繼續看下面章節

四、接收流程

    上一章節介紹呼叫handler.ServeHTTP，可想而知，這個是golang內建的http框架，框架不知道具體業務需求是什麼，所以一般場景下handler都是使用者自定義的，使用者根據不同業務需求來實現不同的handler。對Etcd中存在很多handler，後面會文章進行詳細分析。

    現在解答一下，handler在什麼地方定義以及ServeHTTP定義在什麼地方？這個其實在第三章就已經有提到，在建立完listener會建立一個handler，如etcdhttp.NewPeerHandler(e.Server)，處理叢集節點內訊息。

    先來看一下建立的handler裡面有什麼內容？用peerhandler作為例子說明，在構造方法NewPeerHandler中主要呼叫newPeerHandler方法：

func newPeerHandler(cluster api.Cluster, raftHandler http.Handler, leaseHandler http.Handler) http.Handler {
	mh := &peerMembersHandler{
		cluster: cluster,
	}
//將url和業務層handler註冊到servemux中，也就是每一個url請求都會有其對應的handler進行處理。
	mux := http.NewServeMux() //初始化一個Serve Multiplexer結構
	mux.HandleFunc("/", http.NotFound)
	mux.Handle(rafthttp.RaftPrefix, raftHandler)  /* rafthttp.RaftPrefix == /raft */
	mux.Handle(rafthttp.RaftPrefix+"/", raftHandler)
	mux.Handle(peerMembersPrefix, mh) //處理請求/members handler是mh，即peerMembersHandler
	if leaseHandler != nil {
		mux.Handle(leasehttp.LeasePrefix, leaseHandler)  /* /leases */
		mux.Handle(leasehttp.LeaseInternalPrefix, leaseHandler) /* /leases/internal */
	}
	mux.HandleFunc(versionPath, versionHandler(cluster, serveVersion))
	return mux
}

    通過上面的程式碼可知，應用層業務邏輯需要自己註冊url和handler，這樣才能保證每個http request都能夠被處理。而每個handler都必須要實現對應介面ServeHTTP，例如peerMembersHandler，實現的ServeHTTP介面是用於返回叢集成員列表。那麼此處只是完成註冊，那麼在什麼地方會呼叫此處handler？接下來看一下golang內建方法。

流程圖中所有方法均在Go/src/net/http/server.go中。通過流程圖可知：

// ServeHTTP dispatches the request to the handler whose
// pattern most closely matches the request URL.
func (mux *ServeMux) ServeHTTP(w ResponseWriter, r *Request) {
	if r.RequestURI == "*" {
		if r.ProtoAtLeast(1, 1) {
			w.Header().Set("Connection", "close")
		}
		w.WriteHeader(StatusBadRequest)
		return
	}
	h, _ := mux.Handler(r)  //登錄檔查詢，使用者自定義的handler，並且呼叫ServeHTTP介面，處理該http request請求。
	h.ServeHTTP(w, r)
}

五、傳送流程

    對於傳送流程，這裡打算簡單介紹一下，後續在介紹raft協議是會詳細說明。

    通過之前介紹，ServeHTTP介面是用於處理http request請求入口，每一個handler都必須實現此介面。此介面有兩個引數，分別為：ResponseWriter，Request。其中ResponseWriter就是用於響應，http請求，這裡用peerMembersHandler作為舉例（比較簡單），程式碼如下：

/* 獲取叢集成員列表 */
func (h *peerMembersHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if !allowMethod(w, r, "GET") {
		return
	}
	w.Header().Set("X-Etcd-Cluster-ID", h.cluster.ID().String())

	if r.URL.Path != peerMembersPrefix {
		http.Error(w, "bad path", http.StatusBadRequest)
		return
	}
	ms := h.cluster.Members() //獲取叢集成員列表
	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(ms); err != nil {//呼叫json介面，進行格式化並且將資料寫到緩衝區中
		plog.Warningf("failed to encode members response (%v)", err)
	}
}

    通過檢視Encode程式碼可知，最後會將會資料寫入到ResponseWriter的緩衝區中，即呼叫Wirter介面，並沒有真正傳送到對端。那麼在什麼地方才會真正發出去呢？在golang http原始碼w.finishRequest()，此處會進行重新整理操作，將緩衝區資料傳送到對端。在上面已經提交到。

六、總結

至此，介紹Etcd網路通訊流程就結束了，下一篇介紹網路模型進階篇。