pipeline是什麼？——lonng/nano

以下是源碼中涉及到使用pipeline的地方

localProcess是nano中處理用戶端請求消息的方法

func (h *LocalHandler) localProcess(... ...) {
    // NOTE: msg是message.Message類型
    if pipe := h.pipeline; pipe != nil {
        err := pipe.Inbound().Process(session, msg)
        if err != nil {
            log.Println("Pipeline process failed: " + err.Error())
            return
        }
    }
    
    // ... 執行請求消息對應handler
}
           

write是nano中處理服務端響應消息的方法，說白了就是讀寫分離的異步寫消息的地方

func (a *agent) write(){
    for {
        select{
            case: ... ...
            
            case data := <-a.chSend:
                // 業務消息序列化 ...
                // 包裝成Message
                if pipe := a.pipeline; pipe != nil {
                err := pipe.Outbound().Process(a.session, m)
                if err != nil {
                    log.Println("broken pipeline", err.Error())
                    break
                }
                
                // 發送 Message
            }
        }
        
    }
}
           

Pipeline含有兩個pipelineChannel(以下簡稱Channel), 一個稱為outbound,一個稱為inbound。每個Channel可以注冊一系列Pipeline.Func, 在調用Process順序執行。

源碼進入=> lonng/nano/blob/master/pipeline/pipeline.go

也就是說，Pipeline相當于Message與Handler(業務邏輯)的中間人，如果我們需要在message和Handler的中間對message做一些處理可以在Pipeline中完成。比如，流量統計、加密、校驗等行為。