BYR Achieve · 镜像论坛

redis总报链接数的错误,于是写了个pipeline

2017/7/7镜像同步8 回复

本菜鸟的负责一个服务, 一次用户请求需要请求一次redis, 一直用redigo, 然后redis这块有瓶颈, 会报一些连接 reset by peer 写了一个简单的生产者消费者模型, 无锁, 起一个消费者routine,代理所有的redis读操作消费者routine凑够是个redis读, 就发一条pipeline出去 ``` type entry struct { rcmd rCmd valueChan chan rValue } const ( MaxCmds = 10 littleWhile = 5 MaxLen = 1000 ) var entryChan chan entry func init() { go consumer() } func consumer() { entryChan = make(chan entry,MaxLen) cmdArray := make([]entry,0) for { select { case rcmd := <- entryChan: cmdArray = append(cmdArray,rcmd) if len(cmdArray) > MaxCmds { err := pipeline(cmdArray) if err != nil { fmt.Println("pipeline err", err) } cmdArray = make([]entry,0) } default: if len(cmdArray) == 0 { time.Sleep(littleWhile * time.Millisecond) }else{ err := pipeline(cmdArray) if err != nil { fmt.Println("default pipeline err", err) } cmdArray = make([]entry,0) } } } } func producer(cmd string, keys ...string) (interface{}, error) { rc := rCmd{ cmd : cmd, key : keys, } ch := make(chan rValue) e := entry{ rcmd: rc, valueChan: ch, } entryChan <- e value := <- ch return value.val, value.err } func pipeline(entrys []entry)(err error) { defer func() { if e:=recover();e!=nil{ err = fmt.Errorf("%v",e) } }() c := pool.Get() defer c.Close() for _, ent := range entrys { c.Send(ent.rcmd.cmd, ent.rcmd.key) } c.Flush() for _, ent := range entrys { v, err := c.Receive() rv := rValue{ val : v, err : err, } ent.valueChan <- rv } return } ``` 经过这个routine代理, 整个服务的qps提升了1.3倍的样子吧...

订阅后，新回复会通过你的通知中心匿名送达。