Redis 发布订阅，小功能大用处，真没那么废材！
假設我們有這么一個業務場景，在網站下單支付以后，需要通知庫存服務進行發貨處理。
上面業務實現不難，我們只要讓庫存服務提供給相關的給口，下單支付之后只要調用庫存服務即可。
后面如果又有新的業務，比如說積分服務，他需要獲取下單支付的結果，然后增加用戶的積分。
這個實現也不難，讓積分服務同樣提供一個接口，下單支付之后只要調用庫存服務即可。
如果就兩個業務需要獲取下單支付的結果，那也還好，程序改造也快。可是隨著業務不斷的發展，越來越多的新業務說是要下單支付的結果。
這時我們會發現上面這樣的系統架構存在很多問題：
第一，下單支付業務與其他業務重度耦合，每當有個新業務需要支付結果，就需要改動下單支付的業務。
第二，如果調用業務過多，會導致下單支付接口響應時間變長。另外，如果有任一下游接口響應變慢，就會同步導致下單支付接口響應也變長。
第三，如果任一下游接口失敗，可能導致數據不一致的情況。比如說下圖，先調用 A，成功之后再調用 B，最后再調用 C。
如果在調用 B 接口的發生異常，此時可能就導致下單支付接口返回失敗，但是此時 A 接口其實已經調用成功，這就代表它內部已經處理下單支付成功的結果。
這樣就會導致 A，B，C 三個下游接口，A 獲取成功獲取支付結果，但是 B，C 沒有拿到，導致三者系統數據不一致的情況。
其實我們仔細想一下，對于下單支付業務來講，它其實不需要關心下游調用結果，只要有某種機制通知能通知到他們就可以了。
講到這里，這就需要引入今天需要介紹發布訂閱機制。
Redis 發布與訂閱
Redis 提供了基于「發布/訂閱」模式的消息機制，在這種模式下，消息發布者與訂閱者不需要進行直接通信。
如上圖所示，消息發布者只需要想指定的頻道發布消息，訂閱該頻道的每個客戶端都可以接受到到這個消息。
使用 Redis 發布訂閱這種機制，對于上面業務，下單支付業務只需要向「支付結果」這個頻道發送消息，其他下游業務訂閱「支付結果」這個頻道，就能收相應消息，然后做出業務處理即可。
這樣就可以解耦系統上下游之間調用關系。
接下來我們來看下，我們來看下如何使用 Redis 發布訂閱功能。
Redis 中提供了一組命令，可以用于發布消息，訂閱頻道，取消訂閱以及按照模式訂閱。
首先我們來看下如何發布一條消息，其實很簡單只要使用 「publish」 指令:
publish?channel?message上圖中，我們使用 「publish」 指令向 「pay_result」 這個頻道發送了一條消息。我們可以看到 redis 向我們返回 0 ，這其實代表當前訂閱者個數，由于此時沒有訂閱，所以返回結果為 0 。
接下來我們使用 「subscribe」 訂閱一個或多個頻道
subscribe?channel?[channel?...]如上圖所示，我們訂閱 「pay_result」 這個頻道，當有其他客戶端往這個頻道發送消息，
當前訂閱者就會收到消息。
我們子在使用訂閱命令，需要主要幾點：
第一，客戶端執行訂閱指令之后，就會進入訂閱狀態，之后就只能接收 「subscribe」、「psubscribe」、「unsubscribe」、「punsubscribe」 這四個命令。
第二，新訂閱的客戶端，是「無法收到這個頻道之前的消息」，這是因為 Redis 并不會對發布的消息持久化的。
?相比于很多專業 MQ，比如 kafka、rocketmq 來說， ?redis 發布訂閱功能就顯得有點簡陋了。不過 redis 發布訂閱功能勝在簡單，如果當前場景可以容忍這些缺點，還是可以選擇使用的。看到這里是不是感覺這個功能挺廢材的，不要急，往下看，適合的場景還是有大用處的~
除了上面的功能以外的，Redis 還支持模式匹配的訂閱方式。簡單來說，客戶端可以訂閱一個帶 * 號的模式，如果某些頻道的名字與這個模式匹配，那么當其他客戶端發送給消息給這些頻道時，訂閱這個模式的客戶端也將會到收到消息。
使用 Redis 訂閱模式，我們需要使用一個新的指令 「psubscribe」。
我們執行下面這個指令：
psubscribe?pay.*那么一旦有其他客戶端往 ?「pay」 開頭的頻道，比如 pay_result、pay_xxx，我們都可以收到消息。
如果需要取消訂閱模式，我們需要使用相應punsubscribe 指令，比如取消上面訂閱的模式：
punsubscribe?pay.*Redis 客戶端發布訂閱使用方式
基于 Jedis 開發發布/訂閱
聊完 Redis 發布訂閱指令，我們來看下 Java Redis 客戶端如何使用發布訂閱。
?下面的例子主要基于 Jedis，maven 版本為：
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>3.1.0</version> </dependency>其他 Redis 客戶端大同小異。
jedis 發布代碼比較簡單，只需要調用 Jedis 類的 publish 方法。
//?生產環境千萬不要這么使用哦，推薦使用?JedisPool?線程池的方式? Jedis?jedis?=?new?Jedis("localhost",?6379); jedis.auth("xxxxx"); jedis.publish("pay_result",?"hello?world");訂閱的代碼就相對復雜了，我們需要繼承 JedisPubSub實現里面的相關方法，一旦有其他客戶端往訂閱的頻道上發送消息，將會調用 JedisPubSub 相應的方法。
private?static?class?MyListener?extends?JedisPubSub?{@Overridepublic?void?onMessage(String?channel,?String?message)?{System.out.println("收到訂閱頻道："?+?channel?+?"?消息："?+?message);}@Overridepublic?void?onPMessage(String?pattern,?String?channel,?String?message)?{System.out.println("收到具體訂閱頻道："?+?channel?+?"訂閱模式："?+?pattern?+?"?消息："?+?message);}}其次我們需要調用 Jedis 類的 subscribe 方法：
Jedis?jedis?=?new?Jedis("localhost",?6379); jedis.auth("xxx"); jedis.subscribe(new?MyListener(),?"pay_result");當有其他客戶端往 pay_result頻道發送消息時，訂閱將會收到消息。
不過需要注意的是，jedis#subscribe 是一個阻塞方法，調用之后將會阻塞主線程的，所以如果需要在正式項目使用需要使用異步線程運行，這里就不演示具體的代碼了。
基于 Spring-Data-Redis 開發發布訂閱
原生 jedis 發布訂閱操作，相對來說還是有點復雜。現在我們很多應用已經基于 SpringBoot 開發，使用 spring-boot-starter-data-redis ，可以簡化發布訂閱開發。
首先我們需要引入相應的 startter 依賴：
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><exclusions><exclusion><artifactId>lettuce-core</artifactId><groupId>io.lettuce</groupId></exclusion></exclusions> </dependency> <dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId> </dependency>?這里我們使用 Jedis 當做底層連接客戶端，所以需要排除 lettuce，然后引入 Jedis 依賴。
然后我們需要創建一個消息接收類，里面需要有方法消費消息：
@Slf4j public?class?Receiver?{private?AtomicInteger?counter?=?new?AtomicInteger();public?void?receiveMessage(String?message)?{log.info("Received?<"?+?message?+?">");counter.incrementAndGet();}public?int?getCount()?{return?counter.get();} }接著我們只需要注入 Spring- Redis 相關 Bean，比如：
StringRedisTemplate，用來操作 Redis 命令
MessageListenerAdapter ，消息監聽器，可以在這個類注入我們上面創建消息接受類Receiver
RedisConnectionFactory, 創建 Redis 底層連接
最后我們使用 StringRedisTemplate#convertAndSend 發送消息，同時 Receiver 將會收到一條消息。
@SpringBootApplication public?class?MessagingRedisApplication?{public?static?void?main(String[]?args)?throws?InterruptedException?{ApplicationContext?ctx?=?SpringApplication.run(MessagingRedisApplication.class,?args);StringRedisTemplate?template?=?ctx.getBean(StringRedisTemplate.class);Receiver?receiver?=?ctx.getBean(Receiver.class);while?(receiver.getCount()?==?0)?{template.convertAndSend("pay_result",?"Hello?from?Redis!");Thread.sleep(500L);}System.exit(0);} }Redis 發布訂閱實際應用
Redis Sentinel 節點發現
「Redis Sentinel」 是 Redis 一套高可用方案，可以在主節點故障的時候，自動將從節點提升為主節點，從而轉移故障。
今天這里我們不詳細解釋 「Redis Sentinel」 詳細原理，主要來看下 「Redis Sentinel」 如何使用發布訂閱機制。
「Redis Sentinel」 ?節點主要使用發布訂閱機制，實現新節點的發現，以及交換主節點的之間的狀態。
如下所示，每一個 ?「Sentinel」 ?節點將會定時向 _sentinel_:hello 頻道發送消息，并且每個 「Sentinel」 都會訂閱這個節點。
這樣一旦有節點往這個頻道發送消息，其他節點就可以立刻收到消息。
這樣一旦有的新節點加入，它往這個頻道發送消息，其他節點收到之后，判斷本地列表并沒有這個節點，于是就可以當做新的節點加入本地節點列表。
除此之外，每次往這個頻道發送消息內容可以包含節點的狀態信息，這樣可以作為后面 「Sentinel」 領導者選舉的依據。
以上都是對于 Redis 服務端來講，對于客戶端來講，我們也可以用到發布訂閱機制。
當 ?「Redis Sentinel」 進行主節點故障轉移，這個過程各個階段會通過發布訂閱對外提供。
對于我們客戶端來講，比較關心切換之后的主節點，這樣我們及時切換主節點的連接（舊節點此時已故障，不能再接受操作指令），
客戶端可以訂閱 +switch-master頻道，一旦 ?「Redis Sentinel」 ?結束了對主節點的故障轉移就會發布主節點的的消息。
redission 分布式鎖
redission 開源框架提供一些便捷操作 Redis 的方法，其中比較出名的 redission ?基于 Redis 的實現分布式鎖。
今天我們來看下 Redis 的實現分布式鎖中如何使用 Redis 發布訂閱機制，提高加鎖的性能。
首先我們來看下 redission ? 加鎖的方法：
Redisson?redisson?=?.... RLock?redissonLock?=?redisson.getLock("xxxx"); redissonLock.lock();RLock 繼承自 Java 標準的 Lock 接口,調用 lock 方法,如果當前鎖已被其他客戶端獲取，那么當前加鎖的線程將會被阻塞，直到其他客戶端釋放這把鎖。
這里其實有個問題，當前阻塞的線程如何感知分布式鎖已被釋放呢？
這里其實有兩種實現方法：
第一鐘，定時查詢分布時鎖的狀態，一旦查到鎖已被釋放（Redis 中不存在這個鍵值），那么就去加鎖。
實現偽碼如下：
while?(true)?{boolean?result=lock();if?(!result)?{Thread.sleep(N);} }這種方式實現起來起來簡單，不過缺點也比較多。
如果定時任務時間過短，將會導致查詢次數過多，其實這些都是無效查詢。
如果定時任務休眠時間過長，那又會導致加鎖時間過長，導致加鎖性能不好。
那么第二種實現方案，就是采用服務通知的機制，當分布式鎖被釋放之后，客戶端可以收到鎖釋放的消息，然后第一時間再去加鎖。
這個服務通知的機制我們可以使用 Redis 發布訂閱模式。
當線程加鎖失敗之后，線程將會訂閱 redisson_lock__channel_xxx（xx 代表鎖的名稱） 頻道，使用異步線程監聽消息，然后利用 Java 中 Semaphore 使當前線程進入阻塞。
一旦其他客戶端進行解鎖，redission ?就會往這個redisson_lock__channel_xxx 發送解鎖消息。
等異步線程收到消息，將會調用 Semaphore 釋放信號量，從而讓當前被阻塞的線程喚醒去加鎖。
?ps：這里只是簡單描述了 redission ?加鎖部分原理，出于篇幅，這里就不再消息解析源碼。感興趣的小伙伴可以自己看下 redission ? 加鎖的源碼。
通過發布訂閱機制，被阻塞的線程可以及時被喚醒，減少無效的空轉的查詢，有效的提高的加鎖的效率。
?ps: 這種方式，性能確實提高，但是實現起來的復雜度也很高，這部分源碼有點東西，快看暈了。
?
總結
今天我們主要介紹 Redis 發布訂閱功能，主要對應的 Redis 命令為:
「subscribe channel [channel ...]」 ?訂閱一個或多個頻道
「unsubscribe channel」 ?退訂指定頻道
「publish channel message」 發送消息
「psubscribe pattern」 ?訂閱指定模式
「punsubscribe pattern」 ?退訂指定模式
我們可以利用 Redis ?發布訂閱功能，實現的簡單 MQ 功能，實現上下游的解耦。
不過需要注意了，由于 Redis 發布的消息不會被持久化，這就會導致新訂閱的客戶端將不會收到歷史消息。
所以，如果當前的業務場景不能容忍這些缺點，那還是用專業 MQ 吧。
最后介紹了兩個使用 Redis 發布訂閱功能使用場景供大家參考。
有道無術，術可成；有術無道，止于術
歡迎大家關注Java之道公眾號
好文章，我在看??
總結
以上是生活随笔為你收集整理的Redis 发布订阅，小功能大用处，真没那么废材！的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： System.Object 是 .NET
- 下一篇： 前端页面布局常见问题/已踩过的坑大杂烩