PHP使用Redis的Pub/Sub(發佈訂閱)命令

1.概念

名稱	含義
channel	頻道：生產者和消費者直接操做的對象
publish	生產者：向channel發送消息
subscribe	消費者：訂閱一個或多個channel
psubscribe	消費者：匹配訂閱一個或多個channel
punsubscribe	退訂：匹配退訂，無參數則退訂所有channel
unsubscribe	退訂：退訂指定的channel，無參數則退訂所有channel
pubsub	列出當前活動channel（至少有一個訂閱）

2.注意事項

1.生產者publish消息時打開一個鏈接，publish後鏈接能夠當即關閉
2.channel只接收publish發送的消息，自身不存儲消息，若是channel沒有被訂閱，則消息丟棄
3.訂閱的消費者須要一直在線，阻塞獲取消息，鏈接斷開表示當即退訂

3.使用rawCommand命令實現發佈訂閱

rawCommand是php-redis擴展中提供的命令，能夠向redis發送任何原生的命令
1.消費者訂閱Subscribe.php
消費者須要建立redis長鏈接，而且設置set_time_limit和default_socket_timeout，以確保阻塞獲取消息過程php不超時，socket鏈接不超時

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/23
 * Time: 11:29
 */
//設置php腳本執行時間
set_time_limit(0);
//設置socket鏈接超時時間
ini_set('default_socket_timeout', -1);
//聲明測試頻道名稱
$channelName = "testPubSub";
$channelName2 = "testPubSub2";
try {
    $redis = new Redis();
    //創建一個長連接
    $redis->pconnect('192.168.75.132', 6379);
    //阻塞獲取消息
    while (true){
        //構建命令參數
        $param = array('subscribe', $channelName, $channelName2);
        //使用call_user_func_array回調執行命令
        $ret = call_user_func_array(array($redis, 'rawCommand'), $param);
        //若是結果是消息結構
        if (isset($ret[0]) && $ret[0] == 'message'){
            //輸出消息頻道和消息內容
            echo "channel:".$ret[1].",message:".$ret[2]."\n";
        } else {
            //沒有消息休眠1秒
            sleep(1);
        }
    }
} catch (Exception $e){
    echo $e->getMessage();
}

2.生產者發送消息Publish.php

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/23
 * Time: 11:59
 */
$channelName = "testPubSub";
$channelName2 = "testPubSub2";
//向指定頻道發送消息
try {
    $redis = new Redis();
    $redis->connect('192.168.75.132', 6379);
    for ($i=0;$i<5;$i++){
        $data = array('key' => 'key'.$i, 'data' => 'testdata');
        $param = array('publish', $channelName, json_encode($data));
        $ret = call_user_func_array(array($redis, 'rawCommand'), $param);
        print_r($ret);
    }
} catch (Exception $e){
    echo $e->getMessage();
}

3.執行消費者訂閱，開始阻塞獲取消息php Subscribe.php
4.執行生產者，開始發送消息php Publish.php

php .\Publish.php
22222
#返回執行結果：頻道的訂閱數量

查看消費者終端

php .\Subscribe.php
channel:testPubSub,message:{"key":"key0","data":"testdata"}
channel:testPubSub,message:{"key":"key1","data":"testdata"}
channel:testPubSub,message:{"key":"key2","data":"testdata"}
channel:testPubSub,message:{"key":"key3","data":"testdata"}
channel:testPubSub,message:{"key":"key4","data":"testdata"}

消費者獲取到了生產者發送的消息。

4.直接使用php-redis擴展提供的方法實現發佈訂閱

1.消費者訂閱Subscribe.php

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/23
 * Time: 11:29
 */
//設置php腳本執行時間
set_time_limit(0);
//設置socket鏈接超時時間
ini_set('default_socket_timeout', -1);
//聲明測試頻道名稱
$channelName = "testPubSub";
$channelName2 = "testPubSub2";
try {
    $redis = new Redis();
    //創建一個長連接
    $redis->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

<?php
/**
 * Created by PhpStorm.
 * User: jmsite.cn
 * Date: 2019/1/23
 * Time: 11:59
 */
$channelName = "testPubSub";
$channelName2 = "testPubSub2";
//向指定頻道發送消息
try {
    $redis = new Redis();
    $redis->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

php .\Publish.php
22222
#返回執行結果：頻道的訂閱數量

查看消費者終端

php .\Subscribe.php
channel:testPubSub,message:{"key":"key0","data":"testdata"}
channel:testPubSub,message:{"key":"key1","data":"testdata"}
channel:testPubSub,message:{"key":"key2","data":"testdata"}
channel:testPubSub,message:{"key":"key3","data":"testdata"}
channel:testPubSub,message:{"key":"key4","data":"testdata"}

消費者一樣獲取到了生產者發送的消息。
退訂和查看活動channel命令與生產者和消費者相似，只是參數不一樣而已。

 

 

 

 

 

原文地址： https://www.jmsite.cn/blog-586.html