应用场合:比如http请求,有先后次序,需要实现:先请求(request)先发送,并且读取(response)的时候也是遵循这个规则,这个读写构成一个pair(有请求并有返回)
过来,直接上代码吧:
func (cc *ClientConn) Do(req *http.Request) (resp *http.Response, err error) { err = cc.Write(req) //client向http服务器发送请求 if err != nil { return } return cc.Read(req) // client读取http服务器返回回来数据 }
再仔细看看Write的时候干了什么:
func (cc *ClientConn) Write(req *http.Request) (err error) { // Ensure ordered execution of Writes // 生成序号ID,保证 ID= i 比 ID>i 先执行 id := cc.pipe.Next() cc.pipe.StartRequest(id) //这个是重点 defer func() { cc.pipe.EndRequest(id) //当前id执行完,去触发下一个id+1的请求执行 if err != nil { cc.pipe.StartResponse(id) cc.pipe.EndResponse(id) } else { // Remember the pipeline id of this request cc.lk.Lock() cc.pipereq[req] = id //结束的时候保存这个,req的id,方便后面read时候继续按序 cc.lk.Unlock() } }() cc.lk.Lock() //读写锁,防止执行冲突 //判断read/write的错误信息,和net.Conn是否关闭,后面介绍cc的具体struct结构 if cc.re != nil { // no point sending if read-side closed or broken defer cc.lk.Unlock() return cc.re } if cc.we != nil { defer cc.lk.Unlock() return cc.we } if cc.c == nil { // connection closed by user in the meantime defer cc.lk.Unlock() return errClosed } c := cc.c if req.Close { // We write the EOF to the write-side error, because there // still might be some pipelined reads cc.we = ErrPersistEOF } cc.lk.Unlock() //到这里才是具体执行写请求,所以前面都是保证按序请求的步骤 err = cc.writeReq(req, c) cc.lk.Lock() defer cc.lk.Unlock() if err != nil { cc.we = err return err } cc.nwritten++ //次序++ return nil }
再来看看cc的结构:
type ClientConn struct { lk sync.Mutex // 读写锁 c net.Conn // golang连接interface r *bufio.Reader //bufReader re, we error // read/write errors lastbody io.ReadCloser //上一次ioReader nread, nwritten int //读和写的个数 pipereq map[*http.Request]uint //保存pair request和id pipe textproto.Pipeline writeReq func(*http.Request, io.Writer) error //写数据匿名函数 } type Pipeline struct { mu sync.Mutex id uint request sequencer response sequencer } type sequencer struct { mu sync.Mutex id uint wait map[uint]chan uint // 就是用这个管道来阻塞没有到次序的操作 }
按序具体是怎么实现的:
//生成id的代码,批发次序 func (p *Pipeline) Next() uint { p.mu.Lock() id := p.id p.id++ p.mu.Unlock() return id } //执行StartResponse,实际是执行sequencer的Start方法 func (p *Pipeline) StartResponse(id uint) { p.response.Start(id) } func (s *sequencer) Start(id uint) { s.mu.Lock() if s.id == id { //到达当前id咯,可以执行咯,不需要阻塞 s.mu.Unlock() return } c := make(chan uint) if s.wait == nil { s.wait = make(map[uint]chan uint) } s.wait[id] = c //在map里面记录chan s.mu.Unlock() <-c //读取阻塞,等待c的写入 } //当然是前一个id执行结束的时候,后一个id触发阻塞解开的 func (s *sequencer) End(id uint) { s.mu.Lock() if s.id != id { panic("out of sync") } id++ //这里指向后一个id s.id = id if s.wait == nil { s.wait = make(map[uint]chan uint) } c, ok := s.wait[id] if ok { delete(s.wait, id) //删除这个map里面的chan } s.mu.Unlock() if ok { c <- 1 //往这个chan里面写数据,把阻塞解开 } }
同理,read的过程和write过程相似。可见,golang各种锁和管道保证并发环境下的顺序执行