解鎖 Redis 發(fā)布訂閱模式：通過(guò)實(shí)踐演示挖掘消息通信潛能
Redis 的發(fā)布訂閱模式，正是這樣一把強(qiáng)大的“通信鑰匙”，它為開(kāi)發(fā)者們開(kāi)啟了一扇實(shí)現(xiàn)高效消息交互的大門。通過(guò)發(fā)布訂閱模式，一個(gè)消息生產(chǎn)者可以將消息發(fā)布到指定的頻道中，而多個(gè)對(duì)該頻道感興趣的消息消費(fèi)者能夠同時(shí)接收到這些消息，實(shí)現(xiàn)了消息的一對(duì)多廣播。
在本文中，我們將通過(guò)詳細(xì)的實(shí)踐演示，深入探索 Redis 發(fā)布訂閱模式的奧秘。從基礎(chǔ)概念的簡(jiǎn)要回顧，到搭建環(huán)境、編寫代碼示例，再到對(duì)運(yùn)行結(jié)果的詳細(xì)分析，一步步帶你領(lǐng)略這一強(qiáng)大模式在實(shí)際項(xiàng)目中的應(yīng)用魅力，讓你在面對(duì)類似的消息通信需求時(shí)能夠游刃有余。
一、詳解redis發(fā)布訂閱模式
1. 什么是發(fā)布與訂閱
redis發(fā)布訂閱是一種解耦生產(chǎn)者和消費(fèi)者一種消息通信模式，訂閱者通過(guò)訂閱channel等待最新的消息，發(fā)布者按需將消息發(fā)送到指定channel上供訂閱者消費(fèi)：
2. redis發(fā)布訂閱模式的使用
3. redis發(fā)布訂閱的兩種模式
redis發(fā)布訂閱有兩種方式：
- 基于精確頻道的訂閱模式，即訂閱者訂閱名為channel-1的頻道，那么只有channel-1有消息時(shí)才會(huì)通知這些訂閱。
- 基于匹配模型的訂閱模式，即訂閱者可以通過(guò)表達(dá)式訂閱頻道，例如訂閱者訂閱了channel*的頻道，那么所有前綴為channel的頻道都會(huì)向這個(gè)訂閱者發(fā)布消息。
4. 基于頻道的訂閱模式
基于頻道訂閱模式的指令格式如下，可以看到訂閱者可以訂閱多個(gè)頻道
subscribe channel [channel ...]所以我們開(kāi)啟一個(gè)redis客戶端，訂閱一個(gè)channel:sport的頻道，對(duì)應(yīng)的指令和輸出結(jié)果如下，可以看到發(fā)送指令后redis服務(wù)端返回1，通知訂閱者成功訂閱該頻道：
# 客戶端1 訂閱 channel:sport
127.0.0.1:6379> SUBSCRIBE channel:sport
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel:sport"
3) (integer) 1對(duì)應(yīng)的我們也給出發(fā)布消息的指令格式：
publish channel message此時(shí)，我們?cè)匍_(kāi)啟一個(gè)redis客戶端，發(fā)布一條消息到channel:sport的頻道：
# 另一個(gè)客戶端發(fā)送消息
127.0.0.1:6379> PUBLISH channel:sport "this is why we play"
(integer) 1
127.0.0.1:6379>此時(shí)剛剛訂閱消息的redis客戶端就會(huì)實(shí)時(shí)的收到這條消息:
5. 基于匹配模式的發(fā)布與訂閱示例
有時(shí)候我們會(huì)訂閱多個(gè)頻道，我們不可能每次都去手動(dòng)增加訂閱的頻道，例如我們當(dāng)前訂閱頻道有：c1、c2、c3、c4、c5、c6、c7、c8。將來(lái)還可能出現(xiàn)c9等情況。我們不可能實(shí)時(shí)去添加訂閱的頻道。 觀察上面的頻道我們發(fā)現(xiàn)頻道都是以c開(kāi)頭，后續(xù)的數(shù)字不斷變化，所以我們完全可以使用模式匹配來(lái)實(shí)現(xiàn)頻道訂閱。
模式訂閱和取消的命令為：
psubscribe pattern [pattern...]
punsubscribe [pattern [pattern ...]]關(guān)于parttern常見(jiàn)的匹配符有：
- ：表示任意占位符，例如c,可以匹配c、c1、c111
- ?*：匹配一個(gè)及以上個(gè)占位符
- ?：表示匹配一個(gè)占位符
我們希望訂閱c1-c9的頻道基于模式匹配我們就能夠做到這一點(diǎn)。
我們首先開(kāi)啟一個(gè)客戶端，使用模式匹配發(fā)起訂閱：
# 訂閱匹配cxx相關(guān)的模式
PSUBSCRIBE c?*然后我們?cè)陂_(kāi)啟另一個(gè)客戶端，發(fā)送消息到c1頻道：
127.0.0.1:6379> PUBLISH c1 "this is c1 message"
(integer) 1
127.0.0.1:6379>剛剛訂閱的客戶端就會(huì)收到消息：
127.0.0.1:6379> PSUBSCRIBE c?*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "c?*"
3) (integer) 1
1) "pmessage"
2) "c?*"
3) "c1"
4) "this is c1 message"注意:當(dāng)你訂閱PSUBSCRIBE c?* c1訂閱時(shí)，若另一個(gè)客戶端發(fā)送消息到c1你會(huì)收到兩條消息(原因會(huì)在后文源碼解析時(shí)補(bǔ)充)。
如下便是PSUBSCRIBE c?* c1的收到PUBLISH c1 "this is c1 message"的消息內(nèi)容：
# 訂閱c?* 和精確的c1兩個(gè)頻道
127.0.0.1:6379> PSUBSCRIBE c?* c1
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "c?*"
3) (integer) 1
1) "psubscribe"
2) "c1"
3) (integer) 2
1) "pmessage"
2) "c1"
3) "c1"
4) "this is c1 message"
1) "pmessage"
2) "c?*"
3) "c1"
4) "this is c1 message"6. 基于spring boot集成redis落地發(fā)布與訂閱模式
我們將使用一個(gè)用戶訂閱channel:sport，一個(gè)用戶訂閱channel:stock,而另一個(gè)用戶則通過(guò)channel*訂閱兩個(gè)都訂閱：
了解了大體思路，我們就開(kāi)始落地這個(gè)需求，首先第一步還是引入redis的依賴：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>然后在配置文件中給出redis文件的配置信息：
spring.redis.host=127.0.0.1
spring.redis.port=6379隨后就是定義3個(gè)訂閱者的處理器的通用接口定義，后續(xù)我們將通過(guò)函數(shù)時(shí)編程來(lái)分別落地3個(gè)訂閱者的處理器：
public interface RedisMsgHandler {
/**
* 處理發(fā)布者發(fā)送的消息
* @param message
*/
void handleMessage(String message);
}然后我們就可以針對(duì)redis不同的訂閱者進(jìn)行配置進(jìn)行聲明：
@Configuration
@EnableCaching
@Slf4j
public class RedisConfig {
/**
* Redis消息監(jiān)聽(tīng)器容器
*
* @param connectionFactory
* @return
*/
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
//訂閱體育消息
container.addMessageListener(listenerAdapter(msg -> log.info("收到體育新聞，消息內(nèi)容：{}", msg)), new PatternTopic("channel:sport"));
//訂閱股票消息
container.addMessageListener(listenerAdapter(msg -> log.info("收到庫(kù)存消息，消息內(nèi)容：{}", msg)), new PatternTopic("channel:stock"));
//channel前綴的消息都訂閱
container.addMessageListener(listenerAdapter(msg -> log.info("收到channel前綴的消息：{}", msg)), new PatternTopic("channel:*"));
return container;
}
/**
* 配置消息接收處理類
*
* @return
*/
@Bean
@Scope("prototype")
MessageListenerAdapter listenerAdapter(RedisMsgHandler handler) {
//這個(gè)地方 是給messageListenerAdapter 傳入一個(gè)消息接受的處理器，利用反射的方法調(diào)用“receiveMessage”
//也有好幾個(gè)重載方法，這邊默認(rèn)調(diào)用處理器的方法 叫 handleMessage 可以自己到源碼里面看
return new MessageListenerAdapter(handler, "handleMessage");//注意2個(gè)通道調(diào)用的方法都要為 receiveMessage
}
}最后我們啟動(dòng)一個(gè)定時(shí)器定時(shí)向頻道發(fā)送消息：
@EnableScheduling
@Component
public class SenderTask {
@Autowired
private StringRedisTemplate stringRedisTemplate;
//向redis消息隊(duì)列index通道發(fā)布消息
@Scheduled(fixedRate = 2000)
public void sendMessage() {
stringRedisTemplate.convertAndSend("channel:stock", "股票漲了"+Math.random()+"個(gè)百分點(diǎn)");
stringRedisTemplate.convertAndSend("channel:sport", "籃球新聞 No."+((int)(Math.random()*100)+1)+"選手得分");
}
}最后運(yùn)行結(jié)果如下所示：
2025-01-11 22:39:03.129 INFO 16952 --- [ container-3] com.sharkChili.RedisConfig : 收到channel前綴的消息：股票漲了0.5562952653668436個(gè)百分點(diǎn)
2025-01-11 22:39:03.129 INFO 16952 --- [ container-4] com.sharkChili.RedisConfig : 收到體育新聞，消息內(nèi)容：籃球新聞 No.20選手得分
2025-01-11 22:39:03.129 INFO 16952 --- [ container-5] com.sharkChili.RedisConfig : 收到channel前綴的消息：籃球新聞 No.20選手得分
2025-01-11 22:39:05.121 INFO 16952 --- [ container-6] com.sharkChili.RedisConfig : 收到庫(kù)存消息，消息內(nèi)容：股票漲了0.47411355078897777個(gè)百分點(diǎn)
2025-01-11 22:39:05.122 INFO 16952 --- [ container-7] com.sharkChili.RedisConfig : 收到channel前綴的消息：股票漲了0.47411355078897777個(gè)百分點(diǎn)
2025-01-11 22:39:05.135 INFO 16952 --- [ container-9] com.sharkChili.RedisConfig : 收到channel前綴的消息：籃球新聞 No.15選手得分
2025-01-11 22:39:05.135 INFO 16952 --- [ container-8] com.sharkChili.RedisConfig : 收到體育新聞，消息內(nèi)容：籃球新聞 No.15選手得分
2025-01-11 22:39:07.135 INFO 16952 --- [ container-11] com.sharkChili.RedisConfig : 收到channel前綴的消息：股票漲了0.886860372468581個(gè)百分點(diǎn)
2025-01-11 22:39:07.135 INFO 16952 --- [ container-10] com.sharkChili.RedisConfig : 收到庫(kù)存消息，消息內(nèi)容：股票漲了0.886860372468581個(gè)百分點(diǎn)
2025-01-11 22:39:07.136 INFO 16952 --- [ container-12] com.sharkChili.RedisConfig : 收到體育新聞，消息內(nèi)容：籃球新聞 No.89選手得分
2025-01-11 22:39:07.136 INFO 16952 --- [ container-13] com.sharkChili.RedisConfig : 收到channel前綴的消息：籃球新聞 No.89選手得分7. 發(fā)布訂閱的常見(jiàn)的使用場(chǎng)景和優(yōu)缺點(diǎn)
使用場(chǎng)景如:聊天室、公告牌、異步處理電商訂單非核心操作等需要實(shí)現(xiàn)消息解耦的場(chǎng)景都可以使用消息訂閱發(fā)布，如下所示，這就是電商下單業(yè)務(wù)對(duì)于redis發(fā)布訂閱模式的使用模型圖：
當(dāng)然redis發(fā)布訂閱模式優(yōu)缺點(diǎn)也很明顯，我們先來(lái)說(shuō)說(shuō)優(yōu)點(diǎn)，redis發(fā)布訂閱模式 實(shí)現(xiàn)簡(jiǎn)單，對(duì)于簡(jiǎn)單的解耦生產(chǎn)者和消費(fèi)者關(guān)系的應(yīng)用場(chǎng)景綽綽有余，而缺點(diǎn)也是一樣，因?yàn)閞edis發(fā)布訂閱模式實(shí)現(xiàn)比較簡(jiǎn)單，并沒(méi)有持久化機(jī)制無(wú)法保證可靠消費(fèi)和故障恢復(fù)，同時(shí)，相較于Kafka、RocketMQ，Redis的實(shí)現(xiàn)略顯粗糙，無(wú)法實(shí)現(xiàn)消息堆積與回溯。
二、redis如何實(shí)現(xiàn)發(fā)布訂閱模型
關(guān)于redis發(fā)布訂閱模型，對(duì)于底層實(shí)現(xiàn)感興趣的讀者，可以參考筆者下面這篇關(guān)于pub/sub的源碼解析：《聊聊 Redis 的發(fā)布訂閱設(shè)計(jì)與實(shí)現(xiàn)》
三、小結(jié)
本文從redis發(fā)布訂閱模型基本介紹和幾個(gè)實(shí)踐案例角度帶讀者快速入門了一下redis發(fā)布訂閱模型的使用場(chǎng)景以及一些注意事項(xiàng)，希望對(duì)你有幫助。
