发布订阅

redis发布订阅(pub/sub)是一种消息通信模式；发送者(pub)发送消息，订阅者(sub)接收消息。redis客户端可以订阅任意数量的频道。

基础命令：

语法

redis publish命令基本语法如下：

redis 127.0.0.1:6379> PUBLISH channel massage

redis subscribe 命令基本语法如下:

redis 127.0.0.1:6379 > subscribe channel [channel ...]

返回值

接收到信息的订阅者数量。

接收到的消息。

redis-cli
​
subscribe xx
​
​
​
redis-cli
​
publish xx "hello"
​
publish xx "reids publish subscribe"

案例

touch publish.php
​
touch subscribe.php
curl http://192.168.80.21/publish.php?message=你好
**报错需要检查的地方
systemctl stop firewalld
service nginx start
/www/server/php/sbin/php-fpm
​
curl http://192.168.80.21/publish.php?message=你好

public.php

<?php
$redis = new Redis();
$redis->connect("127.0.0.1",6379);
//链接了redis之后，应该需要一个权限验证，密码验证，这里没有，跳过
//$redis->auth("root");
$message = $_GET["message"];
var_dump($message);
$restful = $redis->publish("new",$message);//向频道new发送一条消息
if ($restful){echo "消息发送成功";exit;
}
echo "发送失败";

subscribe.php

<?php
$redis = new Redis();
$redis->connect('127.0.0.1',6379);
//$redis->auth('root');while (true){echo time();ini_set('default_socket_timeout',-1);//不超时$redis->subscribe(['new'],function ($redis,$chan,$message){echo ' chan'.$chan.'  message'.$message;var_dump($redis);sleep(1);});
}

  

广播案例

node-v16是开箱即用的，解压之后放在/www/server/node 中

vim /etc/profile
​
export NODE_HOME=/www/server/node
​
export PATH=$PATH:$NODE_HOME/bin
​
export NODE_PATH=$NODE_HOME/lib/node_modules
​
source /etc/profile
node -v
npm -v
cd /www/wwwroot
ll
cd php/
ll
cd blog/
ll
​
npm install http socket.io ioredis socket.io-client
​
npm run dev **更新js代码
node websocket **更新websocket代码
​
php artisan redis:publish

版本问题：

const {io} = require('socket.io-client');**4以上的要加{}
const io = require("socket.io-client");**4以下的不用加

Redis的Stream类型

1.削峰，减少响应时间

2.降低系统耦合性

1.Redis List类型实现

使用redis的lpush/rpop(rpush/lpop)命令简单实现左进右出或右进左出的list

列表。然后需要开启一个线程任务或者定时任务或者轮询方式，不停的调用rpop方法查看list中是否待处理消息。

2.发布/订阅

生产者和消费者通过相同的一个信道进行交互。信道其实也就是队列。通常会有多个消费者。多个消费者订阅同一个信道，当生产者向信道发布消息时，该信道会立即将消息逐一发布给每个消费者。所以该信道对于消费者是发散的信道，每个消费者都可以得到相同的消息。典型的一对多的关系。

Redis Stream主要用于消息队列（MQ，Message Queue），Redis本身是有一个Redis发布订阅(pub/sub)来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis宕机等，消息就会被丢弃。

XADD:插入消息

xread：消费消息

ack：消费之后用于验证消息是否消费成功

Stream的使用

touch Stream.php
​
redis-cli

xadd stream * massage sky 
​
xadd stream * massage york
​
xadd stream * massage cara
​
xInfo GROUPS  sky
xGroup create sky a1 0
xInfo GROUPS sky
xInfo GROUPS jifen
XReadGroup GROUP a1 aa block 0 count 1 stream sky >
xInfo GROUPS sky
xack sky a1 1642165324727-0 

<?php
$redis = new Redis();
$redis->connect("127.0.0.1",6379);
//$redis->auth("root");
echo "开始下单".PHP_EOL;
$order = ["id" => 1,"price" => 12000,"product_id" => 1
];
//存储订单数据，表示下单成功
$restful = $redis ->hMSet('order:'.$order["id"],$order);
//通过消息队列异步增加用户积分
if ($restful){$redis->xAdd("jifen","*",$order);
}
echo "下单成功";

<?php
$redis = new Redis();
$redis->connect("127.0.0.1",6379);const xKey = "jifen";
const groupName = 'sendMail';function initGroup(){global $redis;$groups = $redis->xInfo('GROUPS',xKey);//这里我们会得到一个数组，数组中的数据就是消息队列的详细信息//因为此处需要判断队列是否有创建消费组，如果有就创建，默认为没有，下面在根据上面xinfo查询的消息来判断队列是否有消费组$exisGroups = false;if ($groups){foreach ($groups as $group){if ($group[1] === groupName){$exisGroups = true;break;}}}if ($exisGroups){//判断消费组是否已经创建，如果没有那么就使用xGroup创建消费组，并从0开始进行消费，消费组名称为groupName$redis->xGroup('create',xKey,groupName,"0");//表示从0开始消费echo "init Group:".groupName.PHP_EOL;}
}
initGroup();
while (true){$getRestful = $redis -> xReadGroup(groupName,'c1',[xKey=>'>'],1,2000);if ($getRestful && isset($getRestful[xKey])){$sendMail = $getRestful[xKey];//取出消费的消息信息foreach ($sendMail as $id =>$fieldList){$redis->xAck(xKey,groupName,[$id]);echo "sendMail with id = ".$fieldList["id"].PHP_EOL;}}usleep(5);
}