Redis进阶:管道、事务与发布订阅摘要: 本篇深入Redis高级操作讲解Pipeline管道批量执行原理与使用、MULTI/EXEC事务配合WATCH实现乐观锁、Pub/Sub发布订阅模式的消息收发实现分享Pipeline中混入耗时操作导致后续命令延迟的踩坑经历对比Pipeline、事务与Lua脚本的适用场景。开篇故事上个月我们做用户积分批量更新1000个用户的积分要同时刷新。同事写了个for循环每个用户一条命令发到Redis。上线后Redis的QPS翻了十倍网络延迟也上去了。我改成Pipeline后1000条命令一次发完执行时间从800ms降到50ms。Redis单线程处理命令但网络往返的延迟可以优化。Pipeline、事务、发布订阅是go-redis的三个进阶能力用好了性能能提升一个量级。这篇我把三者的原理和使用场景讲清楚。一、Pipeline管道批量操作Pipeline把多条命令打包一次网络往返发到Redis结果一次性拿回来。100条命令从100次往返变成1次。packagemainimport(contextfmttimegithub.com/redis/go-redis/v9)// Pipeline基本用法funcpipelineBasic(ctx context.Context,rdb*redis.Client){pipe:rdb.Pipeline()// 注册命令不立即执行返回Cmd用于取结果cmd1:pipe.Set(ctx,key1,value1,0)cmd2:pipe.Get(ctx,key1)// Exec一次性发送所有命令pipe.Exec(ctx)fmt.Println(key1设置结果:,cmd1.Val())fmt.Println(key1的值:,cmd2.Val())}// Pipeline vs 普通操作性能对比funcpipelineBenchmark(ctx context.Context,rdb*redis.Client){constcount1000// 普通方式:逐条执行start:time.Now()fori:0;icount;i{rdb.Set(ctx,fmt.Sprintf(normal:%d,i),x,0)}fmt.Printf(普通方式 %d条: %v\n,count,time.Since(start))// Pipeline方式:批量执行starttime.Now()pipe:rdb.Pipeline()fori:0;icount;i{pipe.Set(ctx,fmt.Sprintf(pipe:%d,i),x,0)}pipe.Exec(ctx)fmt.Printf(Pipeline %d条: %v\n,count,time.Since(start))// Pipeline通常快10-20倍}TxPipeline是事务型Pipeline命令包在MULTI/EXEC里执行保证原子性。// TxPipeline:事务型Pipeline命令要么全成功要么全失败functxPipelineExample(ctx context.Context,rdb*redis.Client){pipe:rdb.TxPipeline()pipe.Incr(ctx,counter)pipe.Set(ctx,flag,done,0)pipe.Expire(ctx,counter,10*time.Minute)pipe.Exec(ctx)}二、事务(MULTI/EXEC/WATCH)Redis事务是一组命令的顺序执行中间不会被其他客户端打断。但Redis事务不支持回滚某条命令出错后面的照样执行。WATCH实现乐观锁的场景很典型。扣库存时先WATCH库存key读取当前值如果大于0就扣减。如果在WATCH和EXEC之间库存被别人改了事务自动失败需要重试。// WATCH实现乐观锁扣库存funcdeductStock(ctx context.Context,rdb*redis.Client,productIDstring)error{stockKey:fmt.Sprintf(stock:%s,productID)maxRetry:3fori:0;imaxRetry;i{// Watch监视stockKey被修改则事务失败err:rdb.Watch(ctx,func(tx*redis.Tx)error{stock,err:tx.Get(ctx,stockKey).Int()iferrredis.Nil{returnfmt.Errorf(商品不存在)}iferr!nil{returnerr}ifstock0{returnfmt.Errorf(库存不足)}// 事务内扣减库存pipe:tx.TxPipeline()pipe.Decr(ctx,stockKey)pipe.HIncrBy(ctx,sales,productID,1)_,errpipe.Exec(ctx)returnerr},stockKey)iferrnil{returnnil// 成功}// 事务冲突则重试iferrredis.TxFailedErr{fmt.Printf(第%d次重试\n,i1)continue}returnerr}returnfmt.Errorf(重试次数用完)}三、发布订阅(Pub/Sub)Pub/Sub是Redis内置的消息广播机制。发布者往channel发消息所有订阅了该channel的客户端都能收到。适合实时通知和聊天室。packagemainimport(contextfmttimegithub.com/redis/go-redis/v9)// 订阅者:监听channelfuncsubscriber(ctx context.Context,rdb*redis.Client,namestring){pubsub:rdb.Subscribe(ctx,chat_room)deferpubsub.Close()// 循环接收消息formsg:rangepubsub.Channel(){fmt.Printf([%s] %s: %s\n,name,msg.Channel,msg.Payload)}}funcmain(){rdb:redis.NewClient(redis.Options{Addr:localhost:6379})deferrdb.Close()ctx:context.Background()// 启动两个订阅者gosubscriber(ctx,rdb,客户端A)gosubscriber(ctx,rdb,客户端B)time.Sleep(time.Second)// 等订阅者就绪// 发布者发消息fori:0;i5;i{rdb.Publish(ctx,chat_room,fmt.Sprintf(消息%d,i1))time.Sleep(time.Second)}time.Sleep(2*time.Second)}Pub/Sub有个特点消息发出去没人订阅就丢了Redis不保存历史消息。需要消息可靠性用Stream或外部消息队列。四、独家踩坑:Pipeline中混入耗时操作这个坑比较隐蔽。我们在一个Pipeline里塞了50条命令其中有一条是KEYS *。结果整个Pipeline的执行时间从20ms飙到3秒后面所有命令都被阻塞了。// 问题代码:Pipeline中混入O(N)耗时命令funcbadPipeline(ctx context.Context,rdb*redis.Client){pipe:rdb.Pipeline()pipe.Set(ctx,k1,v1,0)pipe.Set(ctx,k2,v2,0)// KEYS *是O(N)操作阻塞Redis主线程// Redis单线程执行期间所有后续命令都等着pipe.Keys(ctx,*)pipe.Get(ctx,k1)// 要等KEYS *执行完才能返回pipe.Get(ctx,k2)pipe.Exec(ctx)}排查过程比较曲折。看网络延迟0.3ms正常然后看Redis的SLOWLOG发现KEYS *执行时间有2-3秒。Redis是单线程Pipeline里命令逐条执行一个慢命令阻塞整个Pipeline。修复方案很简单。把KEYS *从Pipeline拿出来单独执行更好的做法是用SCAN替代游标式遍历不阻塞主线程。// 修复后:慢命令独立执行用SCAN替代KEYSfuncfixedPipeline(ctx context.Context,rdb*redis.Client){// Pipeline只放轻量级命令pipe:rdb.Pipeline()pipe.Set(ctx,k1,v1,0)pipe.Set(ctx,k2,v2,0)pipe.Get(ctx,k1)pipe.Get(ctx,k2)pipe.Exec(ctx)// SCAN游标式遍历不阻塞varcursoruint64for{result,newCursor,_:rdb.Scan(ctx,cursor,*,100).Result()fmt.Println(扫描到:,result)cursornewCursorifcursor0{break// 遍历完成}}}经验就是Pipeline里只放O(1)或O(log N)的轻量命令。KEYS、FLUSHALL、大范围SORT这些重操作要独立执行或用SCAN替代。五、对比分析特性Pipeline事务(MULTI/EXEC)Lua脚本原子性无命令间可插入其他客户端命令有顺序执行不被打断有整个脚本原子执行网络往返1次1次1次条件逻辑不支持不支持WATCH只做冲突检测支持脚本内可写if/for错误回滚不涉及不回滚出错继续执行脚本报错不回滚已执行部分适用场景批量读写无依赖需要原子性的简单操作复杂条件判断的原子操作调试难度低中高选择思路很直接。批量读写无依赖用Pipeline。需要原子性但逻辑简单用事务。逻辑复杂且必须原子执行用Lua脚本。日常开发Pipeline用得最多事务次之Lua脚本用在扣库存这类需要条件判断的场景。总结与预告Pipeline是Redis性能优化第一手段把N次网络往返压成1次。事务配合WATCH能实现乐观锁但Redis事务不支持回滚。Pub/Sub适合实时广播不保证消息到达需要可靠消息用Stream。下一篇讲Redis缓存策略深入缓存穿透、击穿、雪崩三种经典问题的解决方案。