通俗易懂的 redis 发布订阅原理实现！

psubscribe指令： psubscribe pattern [pattern ...] 订阅一个或多个符合给定模式的频道；时间复杂度O(n)，n是订阅的模式的数量。
注意点：每个模式以 * 作为匹配符；例如 mumu*匹配所有以 mumu 开头的频道：mumu.juejin、mumu.zhihu、mumu.csdn
publist指令：publish channel message 把信息message发送到指定的频道channel；时间复杂度O(n+m)，n是频道channel的订阅者数量，m则是使用模式订阅(subscribed patterns)的客户端的数量。
注意点：结果集返回是接收到message的订阅者数量，没有订阅者返回0。
pubsub指令：pubsub channels [argument [argument ...]] 查看订阅与发布系统状态；时间复杂度O(n)，n为活跃频道的数量(对于长度较短的频道和模式来说，将进行模式匹配的复杂度视为常数)。
注意：列出当前的活跃频道(指的是那些至少有一个订阅者的频道， 订阅模式的客户端不计算在内)，返回一个活跃频道组成的列表。
punsubscribe指令：punsubscribe [pattern [pattern ...]] 退订所有给定模式的频道；时间复杂度O(n+m)，其中n是客户端已订阅的模式的数量， m则是系统中所有客户端订阅的模式的数量。
注意：pattern未指定那么订阅的所有模式都会被退订；否则只会退订指定的订阅的模式
subscribe指令：subscribe channel [channel ...] 订阅给定的一个或多个频道的信息；时间复杂度O(n)，其中n是订阅的频道的数量。
unsubscribe指令：unsubscribe channel [channel ...] 指退订给定的频道；时间复杂度O(n)，其中n是订阅的频道的数量。
注意：若没有指定退订channel，则默认退订所有频道；否则退订指定频道。BSCRIBE 命令订阅的所有频道都会被退订。在这种情况下，命令会返回一个信息，告知客户端所有被退订的频道。

127.0.0.1:6379> SUBSCRIBE mumu_1 mumu_2Reading messages... (press Ctrl-C to quit)1) "subscribe"    -- 返回值的类型：显示订阅成功2) "mumu_1"       -- 订阅的频道名字3) (integer) 1    -- 目前已订阅的频道数量
1) "subscribe"2) "mumu_2"3) (integer) 2
1) "message"      -- 返回值的类型：信息2) "mumu_1"       -- 来源(从那个频道发送过来)3) "\xe6\x88\x91\xe6\x98\xaf\xe9\x98\xbf\xe6\xb2\x90\xe5\x95\x8a" -- 消息内容

//获取指定频道的订阅的客户端数量127.0.0.1:6379> PUBSUB numsub mumu_1 mumu_21) "mumu_1"    -- 频道名称2) (integer) 1 -- 订阅该频道的客户端数量3) "mumu_2"4) (integer) 1127.0.0.1:6379> pubsub channels1) "mumu_2"  -- 频道名称2) "mumu_1"  -- 频道名称

127.0.0.1:6379> UNSUBSCRIBE mumu_11) "unsubscribe"  -- 返回值的类型：显示取消订阅成功2) "mumu_1"       -- 取消订阅的频道名字3) (integer) 0

struct redisServer {    /* General */    pid_t pid;          //省略百十行        // 百度翻译😜 😜 😜之后意思是： 将频道映射到已订阅客户端的列表(就是保存客户端和订阅的频道信息)    dict *pubsub_channels;  /* Map channels to list of subscribed clients */}

127.0.0.1:6379> PSUBSCRIBE mumu.*Reading messages... (press Ctrl-C to quit)1) "psubscribe"       -- 返回值的类型：显示订阅成功2) "mumu.*"           -- 订阅的模式3) (integer) 1        -- 目前已订阅的模式的数量
1) "pmessage"         -- 返回值的类型：信息2) "mumu.*"           -- 信息匹配的模式3) "mumu.list"        -- 信息本身的目标频道4) "i am a mumu"      -- 信息的内容

127.0.0.1:6379> PUNSUBSCRIBE mumu.*1) "punsubscribe"  -- 返回值的类型：显示退订成功2) "mumu.*"        -- 退订的模式3) (integer) 1     -- 目前已退订的模式的数量

struct redisServer {    /* General */    pid_t pid;          //省略百十行        // 百度翻译😄 😄 😄之后意思是：pubsub订阅的列表信息(大致就是存储订阅模式的信息)    list *pubsub_patterns;  /* A list of pubsub_patterns */}
// 1303行订阅模式列表结构：typedef struct pubsubPattern {    client *client;  -- 订阅模式客户端    robj *pattern;   -- 被订阅的模式} pubsubPattern;

<?php/** * Created by 我是阿沐. * Date: 2021/05/04 * Time: 下午16:02 * QQ: 2511221051@qq.com */
// 设置php脚本执行时间set_time_limit(0);
// 申明测试的平道名称$channel_names = ['mumu_test1', 'mumu_test2', 'mumu_test3'];//当前执行时间$cur_time = time();try {    // 实例化redis    $redis = new Redis();
    // 创建redis链接    $redis->pconnect('127.0.0.1', 6379);        //echo "Server is running: " . $redis->ping();    //阻塞获取消息    while (true) {        // 阻塞获取消息 $redis redis的实例  $channel_name 频道名称  $msg 生产者生成的消息体        $redis->subscribe($channel_names, function ($redis, $channel_name, $msg) {              switch ($channel_name) {                case 'mumu_test1':                    echo "channel:".$channel_name.",message:".$msg."\n";                    break;                case 'mumu_test2':
                    break;                case 'mumu_test3':
                    break;            }            if (!$msg) { //当没有收到消息时 就休眠1s钟                echo "channel:".$channel_name.",message: not appoint channel name"."\n";                sleep(1);            }        });        // 本地测试 运行超过10分钟 则自动结束 并关闭redis链接        if (time() - $cur_time > 10*60){            $redis -> close();            break;        }    }} catch (Exception $e) {    echo $e->getMessage();}

<?php/** * Created by 我是阿沐. * Date: 2021/05/04 * Time: 下午16:04 * QQ: 2511221051@qq.com */
// 申明测试的平道名称$channel_names = ['mumu_test1', 'mumu_test2', 'mumu_test3', 'mumu_test4'];
$channel_name = $channel_names[rand(0,3)];
try {    // 实例化redis类    $redis = new Redis();    // 建立redis链接    $redis->connect('127.0.0.1', 6379);
    for ($i = 0; $i < 10; $i++) {
        $data = array('key' => 'key' . ($i+1), 'msg' => 'I am li a mu !');
        $ret = $redis->publish($channel_name, json_encode($data));
        print_r($ret);    }} catch (Exception $e) {    echo $e->getMessage();}

终端执行消费者订阅，开始阻塞获取消息/usr/local/opt/php@7.2/bin/php Subscribe.php；结果集：
➜  publish-subscribe git:(master) ✗ /usr/local/opt/php@7.2/bin/php Subscribe.phpchannel:mumu_test1,message:{"key":"key1","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key2","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key3","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key4","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key5","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key6","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key7","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key8","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key9","msg":"I am li a mu !"}channel:mumu_test1,message:{"key":"key10","msg":"I am li a mu !"}
上面就是生产者生成的消息内容：msg字符串
终端执行生产者生产数据，开始发送消息/usr/local/opt/php@7.2/bin/php Publish.php；结果集：➜  publish-subscribe git:(master) ✗ /usr/local/opt/php@7.2/bin/php Publish.php1111111111
// 这种场景是 我模拟发送了没有创建的频道 mumu_test4 导致返回的结果集都是0 说明没有被订阅的channel，消息会被丢弃➜  publish-subscribe git:(master) ✗ /usr/local/opt/php@7.2/bin/php Publish.php0000000000

$redis->pubsub('channels'); // 获取所有频道$redis->pubsub('channels', '*pattern*'); // 仅仅获取指定频道$redis->pubsub('numsub', ['channel1', 'channel2']); // 查看指定频道的订阅数$redis->pubsub('numpat'); // 返回订阅模式的数量$redis->unsubscribe(['channel1', 'channel2']); // 客户端退订指定的频道$redis->punsubscribe(['pattern1', 'pattern2']); // 客户端退订所有指定定模式