Redis源码分析(三十)--- pubsub发布订阅模式

2024-04-28 00:32

本文主要是介绍Redis源码分析(三十)--- pubsub发布订阅模式,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Redis源码分析(三十)--- pubsub发布订阅模式


        今天学习了Redis中比较高大上的名词,“发布订阅模式”,发布订阅模式这个词在我最开始接触听说的时候是在JMS(Java Message Service)java消息服务中听说的。这个名次用通俗的一点话说,就是我订阅了这类消息,当只有这类的消息进行广播发送的时候,我才会,其他的消息直接过滤,保证了一个高效的传输效率。下面切入正题,学习一下Redis是如何实现这个发布订阅模式的。先看看里面的简单的API构造;

/*-----------------------------------------------------------------------------* Pubsub low level API*----------------------------------------------------------------------------*/
void freePubsubPattern(void *p) /* 释放发布订阅的模式 */
int listMatchPubsubPattern(void *a, void *b) /* 发布订阅模式是否匹配 */
int clientSubscriptionsCount(redisClient *c) /* 返回客户端的所订阅的数量,包括channels + patterns管道和模式 */
int pubsubSubscribeChannel(redisClient *c, robj *channel) /* Client订阅一个Channel管道 */
int pubsubUnsubscribeChannel(redisClient *c, robj *channel, int notify) /* 取消订阅Client中的Channel */
int pubsubSubscribePattern(redisClient *c, robj *pattern) /* Client客户端订阅一种模式 */
int pubsubUnsubscribePattern(redisClient *c, robj *pattern, int notify) /* Client客户端取消订阅pattern模式 */
int pubsubUnsubscribeAllChannels(redisClient *c, int notify) /* 客户端取消自身订阅的所有Channel */
int pubsubUnsubscribeAllPatterns(redisClient *c, int notify) /* 客户端取消订阅所有的pattern模式 */
int pubsubPublishMessage(robj *channel, robj *message) /* 为所有订阅了Channel的Client发送消息message *//* ------------PUB/SUB API ---------------- */
void subscribeCommand(redisClient *c) /* 订阅Channel的命令 */
void unsubscribeCommand(redisClient *c) /* 取消订阅Channel的命令 */
void psubscribeCommand(redisClient *c) /* 订阅模式命令 */
void punsubscribeCommand(redisClient *c) /* 取消订阅模式命令 */
void publishCommand(redisClient *c) /* 发布消息命令 */
void pubsubCommand(redisClient *c) /* 发布订阅命令 */
在这里面出现了高频的词Pattern(模式)和Channel(频道,叫管道比较别扭),也就是说,后续所有的关于发布订阅的东东都是基于这2者展开进行的。现在大致讲解一下在Redis中是如何实现此中模式的:

1.在RedisClient 内部维护了一个pubsub_channels的Channel列表,记录了此客户端所订阅的频道

2.在Server服务端,同样维护着一个类似的变量叫做,pubsub_channels,这是一个dict字典变量,每一个Channel对应着一批订阅了此频道的Client,也就是Channel-->list of Clients

3.当一个Client publish一个message的时候,会先去服务端的pubsub_channels找相应的Channel,遍历里面的Client,然后发送通知,即完成了整个发布订阅模式。

    我们可以简单的看一下Redis订阅一个Channel的方法实现;

/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or* 0 if the client was already subscribed to that channel. */
/* Client订阅一个Channel管道 */
int pubsubSubscribeChannel(redisClient *c, robj *channel) {struct dictEntry *de;list *clients = NULL;int retval = 0;/* Add the channel to the client -> channels hash table *///在Client的字典pubsub_channels中添加Channelif (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {retval = 1;incrRefCount(channel);/* Add the client to the channel -> list of clients hash table *///添加Clietn到server中的pubsub_channels,对应的列表中de = dictFind(server.pubsub_channels,channel);if (de == NULL) {//如果此频道的Client列表为空,则创建新列表并添加clients = listCreate();dictAdd(server.pubsub_channels,channel,clients);incrRefCount(channel);} else {//否则,获取这个频道的客户端列表,在尾部添加新的客户端clients = dictGetVal(de);}listAddNodeTail(clients,c);}/* Notify the client *///添加给回复客户端addReply(c,shared.mbulkhdr[3]);addReply(c,shared.subscribebulk);addReplyBulk(c,channel);addReplyLongLong(c,clientSubscriptionsCount(c));return retval;
}
添加操作主要分2部,Client自身的内部维护的pubsub_channels的添加,是一个dict字典对象,然后,是server端维护的pubsub_channels中的client列表的添加。在进行Channel频道的删除的时候,也是执行的这2步骤操作:

/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or* 0 if the client was not subscribed to the specified channel. */
/* 取消订阅Client中的Channel */
int pubsubUnsubscribeChannel(redisClient *c, robj *channel, int notify) {struct dictEntry *de;list *clients;listNode *ln;int retval = 0;/* Remove the channel from the client -> channels hash table */incrRefCount(channel); /* channel may be just a pointer to the same objectwe have in the hash tables. Protect it... *///字典删除Client中pubsub_channels中的Channelif (dictDelete(c->pubsub_channels,channel) == DICT_OK) {retval = 1;/* Remove the client from the channel -> clients list hash table *///再移除Channel对应的Client列表de = dictFind(server.pubsub_channels,channel);redisAssertWithInfo(c,NULL,de != NULL);clients = dictGetVal(de);ln = listSearchKey(clients,c);redisAssertWithInfo(c,NULL,ln != NULL);listDelNode(clients,ln);if (listLength(clients) == 0) {/* Free the list and associated hash entry at all if this was* the latest client, so that it will be possible to abuse* Redis PUBSUB creating millions of channels. */dictDelete(server.pubsub_channels,channel);}}/* Notify the client */if (notify) {addReply(c,shared.mbulkhdr[3]);addReply(c,shared.unsubscribebulk);addReplyBulk(c,channel);addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(c->pubsub_patterns));}decrRefCount(channel); /* it is finally safe to release it */return retval;
}
里面还有对应的模式的订阅和取消订阅的操作,原理和channel完全一致,二者的区别在于,pattern是用来匹配的Channel的,这个是什么意思呢。在后面会做出答案,接着看。最后看一个最最核心的方法,客户端发步消息方法:

/* Publish a message */
/* 为所有订阅了Channel的Client发送消息message */
int pubsubPublishMessage(robj *channel, robj *message) {int receivers = 0;struct dictEntry *de;listNode *ln;listIter li;/* Send to clients listening for that channel *///找到Channel所对应的dictEntryde = dictFind(server.pubsub_channels,channel);if (de) {//获取此Channel对应的客户单列表list *list = dictGetVal(de);listNode *ln;listIter li;listRewind(list,&li);while ((ln = listNext(&li)) != NULL) {//依次取出List中的客户单,添加消息回复redisClient *c = ln->value;addReply(c,shared.mbulkhdr[3]);addReply(c,shared.messagebulk);addReplyBulk(c,channel);//添加消息回复addReplyBulk(c,message);receivers++;}}/* Send to clients listening to matching channels *//* 发送给尝试匹配该Channel的客户端消息 */if (listLength(server.pubsub_patterns)) {listRewind(server.pubsub_patterns,&li);channel = getDecodedObject(channel);while ((ln = listNext(&li)) != NULL) {pubsubPattern *pat = ln->value;//客户端的模式如果匹配了Channel,也会发送消息if (stringmatchlen((char*)pat->pattern->ptr,sdslen(pat->pattern->ptr),(char*)channel->ptr,sdslen(channel->ptr),0)) {addReply(pat->client,shared.mbulkhdr[4]);addReply(pat->client,shared.pmessagebulk);addReplyBulk(pat->client,pat->pattern);addReplyBulk(pat->client,channel);addReplyBulk(pat->client,message);receivers++;}}decrRefCount(channel);}return receivers;
}
pattern的作用就在上面体现了,如果某种pattern匹配了Channel频道,则模式的客户端也会接收消息。在server->pubsub_patterns中, pubsub_patterns是一个list列表,里面的每一个pattern只对应一个Client,就是上面的pat->client,这一点和Channel还是有本质的区别的。讲完发布订阅模式的基本操作后,顺便把与此相关的notify通知类也稍稍讲讲,通知只有3个方法,
/* ----------------- API ------------------- */
int keyspaceEventsStringToFlags(char *classes) /* 键值字符类型转为对应的Class类型 */
sds keyspaceEventsFlagsToString(int flags) /* 通过输入的flag值类,转为字符类型*/
void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) /* 发布通知方法,分为2类,keySpace的通知,keyEvent的通知 */
涉及到string To flag 和flag To String 的转换,也不知道这个会在哪里用到;
/* Turn a string representing notification classes into an integer* representing notification classes flags xored.** The function returns -1 if the input contains characters not mapping to* any class. */
/* 键值字符类型转为对应的Class类型 */
int keyspaceEventsStringToFlags(char *classes) {char *p = classes;int c, flags = 0;while((c = *p++) != '\0') {switch(c) {case 'A': flags |= REDIS_NOTIFY_ALL; break;case 'g': flags |= REDIS_NOTIFY_GENERIC; break;case '$': flags |= REDIS_NOTIFY_STRING; break;case 'l': flags |= REDIS_NOTIFY_LIST; break;case 's': flags |= REDIS_NOTIFY_SET; break;case 'h': flags |= REDIS_NOTIFY_HASH; break;case 'z': flags |= REDIS_NOTIFY_ZSET; break;case 'x': flags |= REDIS_NOTIFY_EXPIRED; break;case 'e': flags |= REDIS_NOTIFY_EVICTED; break;case 'K': flags |= REDIS_NOTIFY_KEYSPACE; break;case 'E': flags |= REDIS_NOTIFY_KEYEVENT; break;default: return -1;}}return flags;
}
应该是响应键盘输入的类型和Redis类型之间的转换。在notify的方法还有一个event事件的通知方法:

/* The API provided to the rest of the Redis core is a simple function:** notifyKeyspaceEvent(char *event, robj *key, int dbid);** 'event' is a C string representing the event name.* 'key' is a Redis object representing the key name.* 'dbid' is the database ID where the key lives.  */
/* 发布通知方法,分为2类,keySpace的通知,keyEvent的通知 */ 
void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) {sds chan;robj *chanobj, *eventobj;int len = -1;char buf[24];/* If notifications for this class of events are off, return ASAP. */if (!(server.notify_keyspace_events & type)) return;eventobj = createStringObject(event,strlen(event));//2种的通知形式,略有差别/* __keyspace@<db>__:<key> <event> notifications. */if (server.notify_keyspace_events & REDIS_NOTIFY_KEYSPACE) {chan = sdsnewlen("__keyspace@",11);len = ll2string(buf,sizeof(buf),dbid);chan = sdscatlen(chan, buf, len);chan = sdscatlen(chan, "__:", 3);chan = sdscatsds(chan, key->ptr);chanobj = createObject(REDIS_STRING, chan);//上述几步操作,组件格式字符串,最后发布消息,下面keyEvent的通知同理pubsubPublishMessage(chanobj, eventobj);decrRefCount(chanobj);}/* __keyevente@<db>__:<event> <key> notifications. */if (server.notify_keyspace_events & REDIS_NOTIFY_KEYEVENT) {chan = sdsnewlen("__keyevent@",11);if (len == -1) len = ll2string(buf,sizeof(buf),dbid);chan = sdscatlen(chan, buf, len);chan = sdscatlen(chan, "__:", 3);chan = sdscatsds(chan, eventobj->ptr);chanobj = createObject(REDIS_STRING, chan);pubsubPublishMessage(chanobj, key);decrRefCount(chanobj);}decrRefCount(eventobj);
}
有keySpace和keyEvent的2种事件通知。具体怎么用,等后面碰到的时候在看看。

这篇关于Redis源码分析(三十)--- pubsub发布订阅模式的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/941898

相关文章

SpringKafka消息发布之KafkaTemplate与事务支持功能

《SpringKafka消息发布之KafkaTemplate与事务支持功能》通过本文介绍的基本用法、序列化选项、事务支持、错误处理和性能优化技术,开发者可以构建高效可靠的Kafka消息发布系统,事务支... 目录引言一、KafkaTemplate基础二、消息序列化三、事务支持机制四、错误处理与重试五、性能优

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很

Spring事务中@Transactional注解不生效的原因分析与解决

《Spring事务中@Transactional注解不生效的原因分析与解决》在Spring框架中,@Transactional注解是管理数据库事务的核心方式,本文将深入分析事务自调用的底层原理,解释为... 目录1. 引言2. 事务自调用问题重现2.1 示例代码2.2 问题现象3. 为什么事务自调用会失效3

新特性抢先看! Ubuntu 25.04 Beta 发布:Linux 6.14 内核

《新特性抢先看!Ubuntu25.04Beta发布:Linux6.14内核》Canonical公司近日发布了Ubuntu25.04Beta版,这一版本被赋予了一个活泼的代号——“Plu... Canonical 昨日(3 月 27 日)放出了 Beta 版 Ubuntu 25.04 系统镜像,代号“Pluc

找不到Anaconda prompt终端的原因分析及解决方案

《找不到Anacondaprompt终端的原因分析及解决方案》因为anaconda还没有初始化,在安装anaconda的过程中,有一行是否要添加anaconda到菜单目录中,由于没有勾选,导致没有菜... 目录问题原因问http://www.chinasem.cn题解决安装了 Anaconda 却找不到 An

Spring定时任务只执行一次的原因分析与解决方案

《Spring定时任务只执行一次的原因分析与解决方案》在使用Spring的@Scheduled定时任务时,你是否遇到过任务只执行一次,后续不再触发的情况?这种情况可能由多种原因导致,如未启用调度、线程... 目录1. 问题背景2. Spring定时任务的基本用法3. 为什么定时任务只执行一次?3.1 未启用

Redis 中的热点键和数据倾斜示例详解

《Redis中的热点键和数据倾斜示例详解》热点键是指在Redis中被频繁访问的特定键,这些键由于其高访问频率,可能导致Redis服务器的性能问题,尤其是在高并发场景下,本文给大家介绍Redis中的热... 目录Redis 中的热点键和数据倾斜热点键(Hot Key)定义特点应对策略示例数据倾斜(Data S

SpringBoot如何通过Map实现策略模式

《SpringBoot如何通过Map实现策略模式》策略模式是一种行为设计模式,它允许在运行时选择算法的行为,在Spring框架中,我们可以利用@Resource注解和Map集合来优雅地实现策略模式,这... 目录前言底层机制解析Spring的集合类型自动装配@Resource注解的行为实现原理使用直接使用M

redis+lua实现分布式限流的示例

《redis+lua实现分布式限流的示例》本文主要介绍了redis+lua实现分布式限流的示例,可以实现复杂的限流逻辑,如滑动窗口限流,并且避免了多步操作导致的并发问题,具有一定的参考价值,感兴趣的可... 目录为什么使用Redis+Lua实现分布式限流使用ZSET也可以实现限流,为什么选择lua的方式实现

Redis中管道操作pipeline的实现

《Redis中管道操作pipeline的实现》RedisPipeline是一种优化客户端与服务器通信的技术,通过批量发送和接收命令减少网络往返次数,提高命令执行效率,本文就来介绍一下Redis中管道操... 目录什么是pipeline场景一:我要向Redis新增大批量的数据分批处理事务( MULTI/EXE