Redis 發(fā)布訂閱Pub/Sub一、Pub/Sub 是什么想象一下訂單服務(wù)剛創(chuàng)建了一筆訂單庫存系統(tǒng)要減庫存、通知服務(wù)要發(fā)短信、分析平臺(tái)要記日志——如果每個(gè)服務(wù)之間都直接調(diào) HTTP當(dāng)?shù)谒膫€(gè)消費(fèi)者加入時(shí)所有上游都得改代碼。這種網(wǎng)狀調(diào)用會(huì)隨著系統(tǒng)變大而崩潰。發(fā)布訂閱Pub/Sub翻轉(zhuǎn)了這種模型Publisher ──(msg)── Redis ──(fan-out)── Subscriber A ── Subscriber B ── Subscriber C核心思想發(fā)布者只管往頻道扔消息不關(guān)心誰在聽。訂閱者只管從頻道收消息不關(guān)心誰發(fā)的。Redis 作為中間人負(fù)責(zé)路由。二、Redis Pub/Sub 的三個(gè)核心命令命令作用PUBLISH channel message向頻道發(fā)送消息返回收到消息的訂閱者數(shù)量SUBSCRIBE channel [channel ...]訂閱一個(gè)或多個(gè)頻道精確匹配PSUBSCRIBE pattern訂閱匹配模式的所有頻道如order:*一個(gè)關(guān)鍵性質(zhì)消息不落地。Redis 不存儲(chǔ)已發(fā)送的消息——如果訂閱者掉線了錯(cuò)過就是錯(cuò)過了。這是 Pub/Sub 和消息隊(duì)列RabbitMQ/Kafka的根本區(qū)別。三、Go 中的訂閱與發(fā)布訂閱者// 訂閱者開啟一個(gè)專用長連接持續(xù)接收消息pubsub:rdb.Subscribe(ctx,orders:new,payments:done)deferpubsub.Close()// Channel() 返回一個(gè) Go channel把 Redis 的網(wǎng)絡(luò)流轉(zhuǎn)換為本地 channelch:pubsub.Channel()// 循環(huán)讀消息阻塞直到來消息或 ctx 取消formsg:rangech{fmt.Printf([%s] %s\n,msg.Channel,msg.Payload)}關(guān)鍵細(xì)節(jié)Subscribe()后 Redis 客戶端會(huì)分配一條專用 TCP 連接這條連接被鎖定在 Pub/Sub 模式不能再執(zhí)行普通命令GET/SET 之類的。這也是為什么生產(chǎn)環(huán)境通常給 Pub/Sub 單獨(dú)配一個(gè)連接池。發(fā)布者// 發(fā)布一個(gè) JSON 消息typeOrderEventstruct{OrderIDstringjson:order_idAmountfloat64json:amountStatusstringjson:status}event,_:json.Marshal(OrderEvent{OrderID:O-1024,Amount:99.9,Status:paid})n,_:rdb.Publish(ctx,orders:new,event).Result()fmt.Printf(消息觸達(dá) %d 個(gè)訂閱者\(yùn)n,n)PUBLISH的返回值是收到消息的訂閱者數(shù)量——如果返回 0說明沒人監(jiān)聽消息丟了。這是 Pub/Sub 的fire-and-forget天性。模式訂閱用通配符監(jiān)聽一批頻道// 訂閱所有 order 相關(guān)事件orders:created, orders:cancelled, orders:paid...pubsub:rdb.PSubscribe(ctx,orders:*)ch:pubsub.Channel()formsg:rangech{fmt.Printf([%s → %s] %s\n,msg.Pattern,msg.Channel,msg.Payload)}*匹配一個(gè)層級(jí)如order:*匹配order:new但不匹配order:item:stock。如果需要多級(jí)匹配可以用order:**取決于 Redis 版本。四、Pub/Sub 的消息可靠性問題Pub/Sub 設(shè)計(jì)上是一條廣播水管——水流過就沒了。什么時(shí)候會(huì)丟消息場景后果訂閱者未上線時(shí)發(fā)布消息消息丟失訂閱者的 Channel 緩沖區(qū)滿處理慢消息積壓Redis 客戶端可能主動(dòng)斷連網(wǎng)絡(luò)閃斷期間的消息丟失Redis 宕機(jī)重啟所有訂閱關(guān)系 未發(fā)消息全部消失解決方案Pub/Sub Stream 雙通道Pub/Sub 負(fù)責(zé)實(shí)時(shí)推送Stream 負(fù)責(zé)消息持久化Publisher ──PUBLISH── Redis Pub/Sub ──實(shí)時(shí)推送── Subscriber ──XADD──── Redis Stream ──補(bǔ)償讀取── Subscriber掉線重連后在線時(shí)吃 Pub/Sub 的實(shí)時(shí)推送低延遲掉線后從 Stream 里按消費(fèi)者組的 LastDeliveredID 往回拉取遺漏的消息這不是 Redis 內(nèi)建功能而是應(yīng)用層組合拳。五、Pub/Sub vs 其他消息模式維度Pub/SubRedis StreamsRabbitMQKafka消息持久化? 不持久化???消費(fèi)確認(rèn)? 無? XACK? ACK? offset commit回溯重復(fù)消費(fèi)????延遲 1ms 1ms~ ms 級(jí)~ ms 級(jí)運(yùn)維復(fù)雜度極低低中高適用場景實(shí)時(shí)廣播、緩存失效通知、WebSocket 跨節(jié)點(diǎn)同步事件日志、消息隊(duì)列、消費(fèi)者組企業(yè)消息中間件大數(shù)據(jù)流處理一句話選型要實(shí)時(shí) 丟幾條沒事 → Pub/Sub要可靠 不能丟 → Streams。六、常見應(yīng)用場景1. 跨節(jié)點(diǎn)緩存失效多個(gè) Web 節(jié)點(diǎn)各自有本地緩存數(shù)據(jù)更新后通過 Pub/Sub 廣播一個(gè)cache:invalidate:user:42所有節(jié)點(diǎn)同時(shí)清掉本地緩存。這是 Redis Pub/Sub 最經(jīng)典的生產(chǎn)場景。2. WebSocket 跨節(jié)點(diǎn)消息推送用戶連在節(jié)點(diǎn) A 的 WebSocket 上但觸發(fā)消息的服務(wù)跑在節(jié)點(diǎn) B 上。B 發(fā) Pub/Sub 到 RedisA 訂閱并推給 WebSocket 客戶端。Socket.IO 的 Redis Adapter 就是這個(gè)原理。3. 配置熱更新配置中心更新配置后PUBLISH 一條通知。所有服務(wù)訂閱了config:reload收到后立刻拉取最新配置。七、本章要點(diǎn)要點(diǎn)一句話Pub/Sub 是廣播不是隊(duì)列消息發(fā)完就消失沒有重放、沒有回執(zhí)一條長連接一個(gè) PubSubSubscribe 后連接被專用不要混用PSubscribe 用*通配可以監(jiān)聽整批頻道省去逐個(gè) SUBSCRIBE可靠性靠組合拳Pub/Sub 做實(shí)時(shí)通道Stream 做補(bǔ)償兜底延遲極低亞毫秒級(jí)適合實(shí)時(shí)場景核心認(rèn)知把 Pub/Sub 當(dāng)成一個(gè)廣播喇叭而非消息倉庫來用就對(duì)了。