codis proxy處理流程

cmd/proxy/main.go檔案

解析配置檔案之後重點是proxy.new(config)函式

該函式中，首先會建立乙個proxy結構體，如下:

type proxy struct

然後起兩個協程,分別處理11080和19000埠的請求

go s.serveadmin()
go s.serveproxy()

我們重點看s.serveproxy()的處理流程，即redis client連線19000埠後proxy如何分發到codis server並且將結果返回到客戶端

s.serverproxy也啟動了兩個協程，乙個協程對router中連線池中的連線進行連線可用性檢測，另乙個協程是乙個死迴圈，accept lproxy埠的連線，並且啟動乙個新的session進行處理，**流程如下:

go func(l net.listener) (err error) ()
for 
newsession(c, s.config).start(s.router)//啟動乙個新的session進行處理
}}(s.lproxy)//s為proxy,s.lproxy即19000埠的監聽

首先介紹一下request結構體，該結構體會貫穿整個流程

type request struct

start函式處理流程如下:

tasks := newrequestchanbuffer(1024)//tasks是乙個指向requestchan的指標,requestchan結構體中有乙個data欄位,data欄位是個陣列，儲存1024個指向request的指標
go func() ()
go func() ()

可以看到,s.loopwriter只是從requestchan的data欄位中取出請求並且返回給客戶端，通過上文request結構體的介紹，可以看到，通過在request的batch執行wait操作，只有請求處理完成後loopwriter才會執行

下邊我們看loopreader的執行流程

r := &request{}   //新建乙個request結構體，該結構體會貫穿請求的始終，請求字段，響應欄位都放在request中
r.multi = multi
r.batch = &sync.waitgroup{}
r.database = s.database
r.unixnano = start.unixnano()
if err := s.handlerequest(r, d); err != nil 
} else

看handlerequest函式如何處理請求,重點是router的dispatch函式

func (s *router) dispatch(r *request) error

forward函式呼叫process函式，返回乙個backendconn結構,然後呼叫其pushback函式將請求放入bc.input中

func (d *forwardsync) forward(s *slot, r *request, hkey byte) error 
bc.pushback(r)
return nil
}bc.pushback(r)函式如下:
func (bc *backendconn) pushback(r *request) 
bc.input 
}

至此可以看到,proxy的處理流程

loopwriter->ruquestchan的data欄位中讀取請求並且返回。在batch處等待 loopreader->將請求放入requestchan的data欄位中，並且將請求放入bc.input channel中。在batch處加1

很明顯,proxy並沒有真正處理請求,肯定會有goroutine從bc.input中讀取請求並且處理完成後在batch處減1，這樣當請求執行完成後,loopwriter就可以返回給客戶端端響應了。

從上文得知,proxy結構體中有乙個router欄位，型別為router,結構體型別如下:

type router struct 
slots [maxslotnum]slot //slot
...}

router的pool中管理連線池,執行fillslot時會真正生成連線，放入slot結構體的backend欄位的bc欄位中，slot結構體如下:

type slot struct 
...method forwardmethod
}

我們看一下bc欄位的結構體sharedbackendconn:

type sharedbackendconn struct

每個backendconn中有乙個 input chan *request欄位,是乙個channel,channel中的內容為request指標。也就是第二章節loopreader選取乙個backendconn後，會將請求放入input中。

下邊我們看看處理backendconn input欄位中資料的協程是如何啟動並處理資料的。**路徑為pkg/proxy/backend.go的newbackendconn函式

func newbackendconn(addr string, database int, config *config) *backendconn 
//1024長度的管道,存放1024個*request
bc.input = make(chan *request, 1024)
bc.retry.delay = &delayexp2
go bc.run()
return bc
}

可以看到，在此處建立的backendconn結構，並且初始化bc.input欄位。連線池的建立是在proxy初始化啟動的時候就會建立好。繼續看bc.run()函式的處理流程

func (bc *backendconn) run() 
}log.warnf("backend conn [%p] to %s, db-%d stop and exit",
bc, bc.addr, bc.database)} 
func (bc *backendconn) loopwriter(round int) (err error) 
...for r := range bc.input 
if err := p.flush(len(bc.input) == 0); err != nil else 
}return nil
}

注意此處的loopwriter會從bc.input中取出資料傳送到codis server,bc.newbackendreader會起乙個loopreader,從codis server中讀取資料並且寫到request結構體中，此處的loopreader和loopwriter通過tasks這個channel通訊。

func (bc *backendconn) newbackendreader(round int, config *config) (*redis.conn, chan
func (bc *backendconn) loopreader(tasks 
...bc.setresponse(r, resp, nil)//設定響應資料到request結構體中
}return nil
}func (bc *backendconn) setresponse(r *request, resp *redis.resp, err error) error 
if r.batch != nil 
return err
}

總結一下,backendconn中的函式功能如下

loopwriter->從bc.input中取出請求並且發給codis server,並且將請求放到tasks channel中 loopreader->從tasks中取出請求，設定codis server的響應欄位到request的resp欄位中，並且將batch執行減1操作

小結

一圖勝千言，版權歸***，如下

codis proxy處理流程

codis proxy處理流程

Codis proxy的配置和啟動

IRP 處理流程

codis proxy處理流程

codis proxy處理流程

Codis proxy的配置和啟動

IRP 處理流程

相關推薦