Redis Pub/Sub 发布/订阅模式

一、

概念：
redis 是一个快速、稳定的发布/订阅的信息系统;发布者不是计划发送消息给特定的接受者，而是发布的消息分到不同的频道，不需要知道什么样的订阅者订阅；订阅者对一个或者多个频道感兴趣，只需接受感兴趣的消息，不需要知道什么样的发布者发布的；

发布者和订阅者的解耦合可以带来更大的扩展性和更加动态的网络拓扑

Redis的Pub/Sub功能，只能实时获取订阅的频道消息，当客户端离线后，离线后的频道消息不会被保存起来。

特点
1、客户端执行订阅以后，除了可以继续订阅，取消订阅，PING命令和结束连接外（但是我发现好像只有ctrl+c才可以直接退出redis服务，其他的命令都不好使），不能执行任何其他操作，客户端将阻塞直到订阅通道上发布消息的到来。
2、发布的消息在Redis系统中不存储，因此，必须先执行订阅，再等待消息发布；相反的顺序不支持；
3、订阅的通道名称支持glob模式匹配，如果客户端同时订阅了glob模式的通道和非glob模式的通道，并且名称存在交集，则对于一个发布的消息，该执行订阅的客户端接受到两个消息；

二、操作
redis是先执行订阅，在发布消息；

所以，第一步就是订阅频道；

表示我们订阅了cctv和cctv1频道

127.0.0.1:6379> subscribe CCTV CCTV1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "CCTV"
3) (integer) 1  
1) "subscribe"
2) "CCTV1"
3) (integer) 2
//此时可以吧这个订阅分为两组，

下面就是发布消息：

// 发布端发布消息
127.0.0.1:6379> publish CCTV "cctv is good"
(integer) 1


//而此时订阅消息则变成：
127.0.0.1:6379> subscribe CCTV CCTV1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "CCTV"
3) (integer) 1
1) "subscribe"
2) "CCTV1"
3) (integer) 2
1) "message"
2) "CCTV"
3) "cctv is good"
//此时多出来最后一组消息

通配符的Pub/Sub发布/订阅消息；
客户端可以订阅满足一个或多个规则的channel频道;

127.0.0.1:6379> psubscribe cctv*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "cctv*"
3) (integer) 1//该客户端目前订阅的所有规则个数

127.0.0.1:6379> psubscribe CCTV*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "CCTV*"
3) (integer) 1
1) "pmessage"
2) "CCTV*"
3) "CCTV1"
4) "goods"

可以看出 subscribe和psubscribe 最后得到的结果类似，但是psubscribe多出一个匹配哪个频道的结果

三、实现原理
路径：redis/src/pubsub.c 看出来是C语言基础的

1、Redis将所有订阅关系保持在服务器状态的pubsub_channels 字典项中；

struct redisServer {
    ...
    dict *pubsub_channels;  
    ...
}

该字典项中的key是频道channel名称，value是一个链表，而链表中保存了所有订阅这个channel的客户端；

2、订阅频道
当客户端执行subscribe没命令。订阅某个或某些频道时，服务器会将客户端与被订阅的频道在pubsub_channels字典中进行关联。

void subscribeCommand(client *c) {
    int j;

    for (j = 1; j < c->argc; j++)
        pubsubSubscribeChannel(c,c->argv[j]);
    c->flags |= CLIENT_PUBSUB;
}


int pubsubSubscribeChannel(client *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);
        
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            clients = dictGetVal(de);//订阅该频道所有客户端列表
        }
        listAddNodeTail(clients,c);
    }
    
    addReply(c,shared.mbulkhdr[3]);
    addReply(c,shared.subscribebulk);
    addReplyBulk(c,channel);
    addReplyLongLong(c,clientSubscriptionsCount(c));
    return retval;
}

判断频道是否存在于pubsub_channels字典中分开操作：

如果频道不存在，表示该频道没有任何订阅者，程序则在pubsub_channels中创建该频道，键名为频道名，value为空链表；
如果频道存在，表示该频道已有其他订阅者，那么它在pubsub_channels字典中必然有响应的订阅者链表，程序要将客户端添加到订阅者链表的末尾；

3、取消订阅

void unsubscribeCommand(client *c) {
    if (c->argc == 1) {
        pubsubUnsubscribeAllChannels(c,1);
    } else {
        int j;

        for (j = 1; j < c->argc; j++)
            pubsubUnsubscribeChannel(c,c->argv[j],1);
    }
    if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
}

int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
    dictEntry *de;
    list *clients;
    listNode *ln;
    int retval = 0;

    
    incrRefCount(channel); 
    if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {
        retval = 1;
        
        de = dictFind(server.pubsub_channels,channel);//找到要取消订阅的频道
        serverAssertWithInfo(c,NULL,de != NULL);
        clients = dictGetVal(de);//订阅该频道的所有客户端列表
        ln = listSearchKey(clients,c);//找到要取消订阅的客户端
        serverAssertWithInfo(c,NULL,ln != NULL);
        listDelNode(clients,ln);//将从客户端列表移除
        if (listLength(clients) == 0) {
            //如果订阅该频道的客户端列表为空，表示没有人订阅该频道，将该频道从pubsub_channels字典中移除
            
            dictDelete(server.pubsub_channels,channel);
        }
    }
    
    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); 
    return retval;
}

4、发布

void publishCommand(client *c) {
    int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
    if (server.cluster_enabled)
        clusterPropagatePublish(c->argv[1],c->argv[2]);
    else
        forceCommandPropagation(c,PROPAGATE_REPL);
    addReplyLongLong(c,receivers);
}

四、php+redis代码实现

1、消费者订阅 subscribe.php

pconnect('192.168.75.132', 6379);
    //阻塞获取消息
    $redis->subscribe(array($channelName, $channelName2), function ($redis, $chan, $msg){
        echo "channel:".$chan.",message:".$msg."n";
    });
} catch (Exception $e){
    echo $e->getMessage();
}

2、生产者发布 publish.php

connect('192.168.75.132', 6379);
    for ($i=0;$i<5;$i++){
        $data = array('key' => 'key'.$i, 'data' => 'testdata');
        $ret = $redis->publish($channelName, json_encode($data));
        print_r($ret);
    }
} catch (Exception $e){
    echo $e->getMessage();
}

3、执行消费者订阅，开始阻塞获取消息php subscribe.php

4、执行生产者，开始发送消息php publish.php

Redis Pub/Sub 发布/订阅模式

Java相关栏目本月热门文章