stream作為網路傳輸的通道之一，通過建立HTTP長連接，主要傳輸資料量小、發送頻率較高的資料,例如追加日志、心跳資料，
作業原理
結合上一節的pipeline通道實作的作業原理，我們來看下stream的作業原理：
我們知道stream是HTTP長連接，所以stream需要專門負責寫入資料的streamWriter和專門讀取資料的streamReader，
streamReader作業原理：
在rafthttp.peer.startPeer()中通過初始化并呼叫streamReader.start()啟動streamReader實體，其中還啟動了一個后臺goroutine來執行streamReader.run()方法，在streamReader.run()方法中,主要完成下面兩件事情：
- 通過呼叫streamReader.dial()主動與對端建立連接
- 接收對端發送的資料,寫入streamReader.recvc或streamReader.propc,這兩個通道其實對應的是上層peer.recvc和peer.propc通道,最后訊息在rafthttp.peer.startPeer()
中被接收并提交到etcd-raft模塊被處理，
streamWriter作業原理：
在rafthttp.peer.startPeer()中通過呼叫startStreamWriter()初始化并啟動streamWriter實體，其中還啟動了一個后臺goroutine來執行streamWriter.run()方法，在streamWriter.run()方法中,主要完成下面三件事情:
- 在peer章節中提到過當其他節點主動與當前節點創建連接(streamReader.dial())時,該連接實體會寫入對應streamWriter.connc通道,在streamWriter.run方法中通過該通道獲取該連接實體并進行系結，之后才能開始后續的訊息發送和心跳檢測，
- 獲取連接之后，定時將心跳訊息寫入緩沖區，通過HTTP處理器將緩沖區資料刷到對端,用于定期檢測網路狀況
- 獲取連接之后，從streamWriter.msgc通道接收到要發送的訊息,將訊息寫入緩沖區,通過HTTP處理器將緩沖區資料刷到對端，
streamWriter
結構體
type streamWriter struct {
localID types.ID//本地節點ID
peerID types.ID//對端節點ID
status *peerStatus//網路連接狀態
r Raft //底層的Raft狀態
mu sync.Mutex // guard field working and closer
closer io.Closer//負責關閉底層的長連接
working bool //負責標識當前的streamWriter是否可用
msgc chan raftpb.Message//通過前面對Peer分析可知,Peer會將待發送的訊息寫入該通道,streamWriter則從該通道中讀取訊息并發送出去
connc chan *outgoingConn//通過該通道獲取當前streamWriter實體關聯的底層網路連接， outgoingConn其實是對網路連接的一層封裝，其中記錄了當前連接使用的協議版本，以及用于關閉連接的Flusher和Closer等資訊，
stopc chan struct{}
done chan struct{}
}
streamWriter啟動
func (cw *streamWriter) run() {
var (
msgc chan raftpb.Message //指向當前streamWriter.msgc欄位
/**
定時器會定時向該通道發送信號,觸發心跳訊息的發送,該心跳訊息與后面介紹的Raft的心跳訊息有所不同
該心跳訊息的主要目的是為了防止連接長時間不用斷開的
**/
heartbeatc <-chan time.Time
t streamType//區分不同版本的stream訊息通道
enc encoder//編碼器,負責將訊息序列化并寫入連接的緩沖區
flusher http.Flusher//負責重繪底層連接,將資料真正發送出去
batched int //記錄當前未Flush的訊息個數
)
tickc := time.NewTicker(ConnReadTimeout / 3)//發送心跳訊息的定時器
defer tickc.Stop()
unflushed := 0//未Flush的位元組數
for {
select {
case <-heartbeatc:
//將心跳訊息編碼到緩沖區
err := enc.encode(&linkHeartbeatMessage)
unflushed += linkHeartbeatMessage.Size()//統計未發送的位元組數
if err == nil {
//若沒有例外,則使用flusher將緩沖區的訊息全部發送出去,并重置batched和unflushed
flusher.Flush()
batched = 0
sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
unflushed = 0
continue //不阻塞,直接continue
}
//到這里說明有錯誤，將網路連接狀態置為不活躍
cw.status.deactivate(failureType{source: t.String(), action: "heartbeat"}, err.Error())
sentFailures.WithLabelValues(cw.peerID.String()).Inc()
cw.close()//關閉streamWriter,會導致底層連接的關閉
if cw.lg != nil {
cw.lg.Warn(
"lost TCP streaming connection with remote peer",
zap.String("stream-writer-type", t.String()),
zap.String("local-member-id", cw.localID.String()),
zap.String("remote-peer-id", cw.peerID.String()),
)
}
heartbeatc, msgc = nil, nil
case m := <-msgc://peer向streamWriter.msgc寫入待發送的訊息
err := enc.encode(&m)//將訊息編碼到緩沖區
if err == nil {
unflushed += m.Size()
//msgc通道中的訊息全部發送完成或是未Flush得訊息過多,則觸發Flush
//否則只是遞增batched變數
if len(msgc) == 0 || batched > streamBufSize/2 {
flusher.Flush()
sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
unflushed = 0
batched = 0
} else {
batched++
}
continue
}
//到這里說明有錯誤，將網路連接狀態置為不活躍(略)
/**
當其他節點(對端)主動與當前節點創建Stream訊息通道時，會先通過StreamHandler的處理，
StreamHandler會通過attach()方法將連接寫入對應的peer.writer.connc通道，
而當前的goroutine會通過該通道獲取連接，然后開始發送訊息
**/
case conn := <-cw.connc://獲取當前streamWriter實體系結的底層連接
cw.mu.Lock()
closed := cw.closeUnlocked()
t = conn.t //獲取該連接底層發送得訊息版本,并創建相應的encoder實體
switch conn.t {//這里只關注最新版本,忽略其他版本
case streamTypeMsgAppV2:
enc = newMsgAppV2Encoder(conn.Writer, cw.fs)
case streamTypeMessage:
//將http.ResponseWriter封裝成messageEncoder,上層呼叫通過messageEncoder實體完成訊息發送
enc = &messageEncoder{w: conn.Writer}
default:
}
flusher = conn.Flusher //記錄底層連接對應的Flusher
unflushed = 0 //重置未Flush得位元組數
cw.status.activate() //節點網路連接置為活躍狀態
cw.closer = conn.Closer//
cw.working = true //標識當前streamWriter正在運行
cw.mu.Unlock()
//更新heartbeatc和msg兩個通道,獲取連接后才能發送訊息
heartbeatc, msgc = tickc.C, cw.msgc
case <-cw.stopc:
close(cw.done)
return
}
}
}
streamWriter系結連接
在rafthttp.peer.attachOutgoingConn()提到會呼叫streamWriter.attach()方法streamWriter與connc進行系結
func (cw *streamWriter) attach(conn *outgoingConn) bool {
select {
case cw.connc <- conn://將conn實體寫入streamWriter.connc通道
return true
case <-cw.done:
return false
}
}
主要用于接收對端得連接并進行系結，
streamReader
結構體
type streamReader struct {
peerID types.ID //對應節點得ID
typ streamType//區分不同版本的stream訊息通道
tr *Transport//關聯rafthttp.Transport實體
picker *urlPicker//用于獲取對端可用的URL
status *peerStatus//節點連接狀態
recvc chan<- raftpb.Message創建streamReader時會啟動一個后臺goroutine從peer.recvc通道中讀取訊息，從對端節點發送來的非MsgProp型別訊息會寫入到該通道中，然后由新建的goroutine讀取出來，交給底層的etcd-raft模塊進行處理
propc chan<- raftpb.Message//功能基本同上，只不過用于接收MsgProp型別的訊息，
rl *rate.Limiter //撥號得頻率限制
errorc chan<- error
mu sync.Mutex
paused bool //是否暫停讀取資料
closer io.Closer
ctx context.Context
cancel context.CancelFunc
done chan struct{}
}
streamReader啟動
func (cr *streamReader) run() {
t := cr.typ //獲取使用的訊息版本
for {
//向對端節點發送一個GET請求，然后獲取并回傳回應的ReadCloser，主要用于和對端建立連接
rc, err := cr.dial(t)
if err != nil {
//發生錯誤，將連接狀態置為不活躍
if err != errUnsupportedStreamType {
cr.status.deactivate(failureType{source: t.String(), action: "dial"}, err.Error())
}
} else {
//成功建立連接,將連接狀態置為活躍
cr.status.activate()
//讀取對端回傳的訊息，并將讀取到的訊息寫入streamReader.recvc通道中
err = cr.decodeLoop(rc, t)
switch {
// all data is read out
case err == io.EOF:
// connection is closed by the remote
case transport.IsClosedConnError(err):
default:
cr.status.deactivate(failureType{source: t.String(), action: "read"}, err.Error())
}
}
//來到這里說明連接發送錯誤,等待一會,發起新的dial連接嘗試,呼叫rate.Limiter可限制撥號頻率
err = cr.rl.Wait(cr.ctx)
}
}
streamReader.dial()方法主要負責與對端節點建立連接，
func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
u := cr.picker.pick()//選用對端可用的URL
uu := u
//根據使用協議得版本和節點ID創建最終得URL地址
uu.Path = path.Join(t.endpoint(cr.lg), cr.tr.ID.String())
//創建一個GET請求
req, err := http.NewRequest("GET", uu.String(), nil)
//設定頭部資訊
req.Header.Set("X-Server-From", cr.tr.ID.String())
req.Header.Set("X-Server-Version", version.Version)
req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
req.Header.Set("X-Etcd-Cluster-ID", cr.tr.ClusterID.String())
req.Header.Set("X-Raft-To", cr.peerID.String())
//將當前節點暴露得URL也一起發送給對端節點
setPeerURLsHeader(req, cr.tr.URLs)
//檢測當前streamReader是否被關閉,若已被關閉，則回傳例外
req = req.WithContext(cr.ctx)
cr.mu.Lock()
select {
case <-cr.ctx.Done():
cr.mu.Unlock()
return nil, fmt.Errorf("stream reader is stopped")
default:
}
cr.mu.Unlock()
//通過前面介紹的rafthttp.Transport.streamRt與對端建立連接
resp, err := cr.tr.streamRt.RoundTrip(req)
if err != nil {
cr.picker.unreachable(u)
return nil, err
}
//決議HTTP回應,檢測版本資訊
}
streamReader.decodeLoop()方法是streamReader得核心方法，它從底層得網路連接讀取資料并進行反序列化，之后將得到得訊息實體寫入recvc或propc通道中,等待Peer進行處理，
func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
var dec decoder
cr.mu.Lock()
//根據使用的協議版本創建對應的decoder實體,這里只關注協議的最新版本
switch t {
case streamTypeMsgAppV2:
dec = newMsgAppV2Decoder(rc, cr.tr.ID, cr.peerID)
case streamTypeMessage:
dec = &messageDecoder{r: rc}//初始化messageDecoder實體并從連接中得到資料
default:
}
//檢測streamReader是否關閉,關閉則回傳例外
select {
case <-cr.ctx.Done():
cr.mu.Unlock()
if err := rc.Close(); err != nil {
return err
}
return io.EOF
default:
cr.closer = rc
}
cr.mu.Unlock()
for {
//從底層連接中讀取資料,并反序列化成raftpb.message實體
m, err := dec.decode()
cr.mu.Lock()
paused := cr.paused
cr.mu.Unlock()
if paused {//檢測是否已經暫停從該連接上讀取資料
continue
}
//忽略連接層的心跳訊息,注意與后面介紹的raft得心跳訊息進行區分
if isLinkHeartbeatMessage(&m) {
// raft is not interested in link layer
// heartbeat message, so we should ignore
// it.
continue
}
//根據讀取到得訊息型別,選擇對應的通道進行寫入
recvc := cr.recvc
if m.Type == raftpb.MsgProp {
recvc = cr.propc
}
select {
case recvc <- m://將訊息寫入對應的通道中,之后會交給底層Raft狀態機進行處理
default:
}
}
}
更多歡迎關注go成神之路
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/205233.html
標籤：其他