Go語(yǔ)言io?pipe源碼分析詳情

更新時(shí)間：2022年02月09日 10:01:24 作者：zwf193071

這篇文章主要介紹了Go語(yǔ)言io?pipe源碼分析詳情，pipe是一個(gè)適配器,用于連接Reader和Writer，pipe的方法不多,新的寫(xiě)法卻不少，并且結(jié)構(gòu)體分兩塊，讀寫(xiě)信道和結(jié)束標(biāo)識(shí)，下面進(jìn)入文章了解具體的內(nèi)容吧

1.結(jié)構(gòu)分析

對(duì)外暴露的是一個(gè)構(gòu)造函數(shù)和構(gòu)造的兩個(gè)對(duì)象. 兩個(gè)對(duì)象分別暴露了方法,同時(shí)這兩個(gè)對(duì)象還有一個(gè)共同的底層對(duì)象. 實(shí)際上,這兩個(gè)對(duì)象暴露的方法是直接調(diào)用底層對(duì)象的, 那么核心還是在底層對(duì)象上,只是通過(guò)兩個(gè)對(duì)象和一個(gè)構(gòu)造方法將底層對(duì)象的細(xì)節(jié)隱藏了.

2.pipe sruct分析

pipe的方法不多,新的寫(xiě)法卻不少.

 type atomicError struct{ v atomic.Value }

? ? func (a *atomicError) Store(err error) {
? ? ? a.v.Store(struct{ error }{err})
? ? }
? ? func (a *atomicError) Load() error {
? ? ? err, _ := a.v.Load().(struct{ error })
? ? ? return err.error
? ? }

atomicError提供了error的原子讀寫(xiě).

? type pipe struct {
? ? ? wrMu sync.Mutex // Serializes Write operations
? ? ? wrCh chan []byte
? ? ? rdCh chan int

? ? ? once sync.Once // Protects closing done
? ? ? done chan struct{}
? ? ? rerr atomicError
? ? ? werr atomicError
? ? }

可以看到pipe結(jié)構(gòu)體中主要分兩塊:

讀寫(xiě)信道
- 兩個(gè)無(wú)緩沖信道
- 一個(gè)互斥量(保護(hù)暴露的寫(xiě)函數(shù))
結(jié)束標(biāo)識(shí)
- once保證done的關(guān)閉只執(zhí)行一次
- done標(biāo)志整個(gè)讀寫(xiě)的結(jié)束
- 剩下兩個(gè)用于存儲(chǔ)讀寫(xiě)錯(cuò)誤
- PipeReader/PipeWriter的分析

3.PipeReader對(duì)外暴露的是讀/關(guān)閉

? ? type PipeReader struct {
? ? ? p *pipe
? ? }

? ? func (r *PipeReader) Read(data []byte) (n int, err error) {
? ? ? return r.p.Read(data)
? ? }

? ? func (r *PipeReader) Close() error {
? ? ? return r.CloseWithError(nil)
? ? }

? ? func (r *PipeReader) CloseWithError(err error) error {
? ? ? return r.p.CloseRead(err)
? ? }

PipeWriter對(duì)外暴露的是寫(xiě)/關(guān)閉

 ? ?type PipeWriter struct {
? ? ? ?p *pipe
? ? ?}

? ? func (w *PipeWriter) Write(data []byte) (n int, err error) {
? ? ? return w.p.Write(data)
? ? }

? ? func (w *PipeWriter) Close() error {
? ? ? return w.CloseWithError(nil)
? ? }

? ? func (w *PipeWriter) CloseWithError(err error) error {
? ? ? return w.p.CloseWrite(err)
? ? }

他們的方法集都是指針接收者.具體方法的實(shí)現(xiàn)是通過(guò)pipe的方法完成的. pipe的方法更加明確:讀/獲取讀錯(cuò)誤/結(jié)束讀寫(xiě)并設(shè)置讀錯(cuò)誤; 寫(xiě)/獲取寫(xiě)錯(cuò)誤/結(jié)束讀寫(xiě)并設(shè)置寫(xiě)錯(cuò)誤.思路相當(dāng)明確.

下面主要分析pipe的讀寫(xiě)

 ? func (p *pipe) Read(b []byte) (n int, err error) {
? ? ? select {
? ? ? case <-p.done:
? ? ? ? return 0, p.readCloseError()
? ? ? default:
? ? ? }

? ? ? select {
? ? ? case bw := <-p.wrCh:
? ? ? ? nr := copy(b, bw)
? ? ? ? p.rdCh <- nr
? ? ? ? return nr, nil
? ? ? case <-p.done:
? ? ? ? return 0, p.readCloseError()
? ? ? }
? ? }

? ? func (p *pipe) Write(b []byte) (n int, err error) {
? ? ? select {
? ? ? case <-p.done:
? ? ? ? return 0, p.writeCloseError()
? ? ? default:
? ? ? ? p.wrMu.Lock()
? ? ? ? defer p.wrMu.Unlock()
? ? ? }

? ? ? for once := true; once || len(b) > 0; once = false {
? ? ? ? select {
? ? ? ? case p.wrCh <- b:
? ? ? ? ? nw := <-p.rdCh
? ? ? ? ? b = b[nw:]
? ? ? ? ? n += nw
? ? ? ? case <-p.done:
? ? ? ? ? return n, p.writeCloseError()
? ? ? ? }
? ? ? }
? ? ? return n, nil
? ? }

讀寫(xiě)都是利用兩個(gè)階段的select來(lái)完成,第一個(gè)階段的select是判斷讀寫(xiě)有沒(méi)有結(jié)束, 第二階段處理實(shí)際的讀寫(xiě).

Read

每次將讀的數(shù)量寫(xiě)到讀信道

Write

先將緩沖寫(xiě)到寫(xiě)信道,再?gòu)淖x信道中獲取讀字節(jié)數(shù),最后調(diào)整緩沖
如果緩沖太大,一次讀沒(méi)讀完,就將寫(xiě)的過(guò)程多來(lái)幾遍,知道緩沖全部寫(xiě)完

4.寫(xiě)法

PipeWriter/PipeReader對(duì)外暴露的關(guān)閉,其實(shí)只可以保留一個(gè)CloseWithError, 但是為了方便客戶(hù)(調(diào)用者),還是拆成兩個(gè),其實(shí)可以做測(cè)試比較一下. 性能測(cè)試發(fā)現(xiàn)拆成兩個(gè)或?qū)懗梢粋€(gè)可選參函數(shù),性能上差別不大, 那這種寫(xiě)法的主要作用是讓暴露的方法更加清晰易懂.

pipe.Write中,for循環(huán)帶有once參數(shù),可以保證循環(huán)至少來(lái)一次, 算是do while的一種實(shí)現(xiàn).