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中完成。比如，流量统计、加密、校验等行为。