下次自动登录
现在的位置:
& 综合 & 正文
Redis pub/sub(Publish,Subscribe)
Pub/Sub功能（means Publish, Subscribe）即发布及订阅功能。基于事件的系统中，Pub/Sub是目前广泛使用的通信模型，它采用事件作为基本的通信机制，提供大规模系统所要求的松散耦合的交互模式：订阅者(如客户端)以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件；发布者(如服务器)可将订阅者感兴趣的事件随时通知相关订阅者。
Publisher/subscribe 简易模型
Pub/Sub功能（means
Publish,Subscribe）即发布及订阅功能
1. 时间非耦合：发布者和订阅者不必同时在线，它们不必同时参与交互。
2. 空间非耦合：发布者和订阅者不必相互知道对方所在的位置。发布者通过事件服务发布事件，订阅者通过事件服务间接获得事件。发布者和订阅者不需要拥有直接到对方的引用，也不必知道有多少个订阅者或者是发布者参与交互。
3. 同步非耦合：发布者/订阅者是异步模式。发布者可不断地生产事件，而订阅者(通过一个回调)则可异步地得到产生事件的通知。
按照订阅方式分为基于主题（topic-based）、基于内容（content-based）、基于类型（type-based）的pub/sub方式。
Pub/Sub是可适用于可扩展要求高、松散耦合系统的分布式交互模型。
在抽象层中，它的时间非耦合、空间非耦合和同步非耦合性可允许参与者不依赖另一个而独立操作，具有一定的可扩展性；然而在实现层，可扩展性仍受其他原因的牵制。
例如：1、灵活的订阅要求复杂的过滤和路由；
2、高可用性开销（事件侦听、日志重传）；
3、消息认可带来的网络流量消耗；
4、庞大的订阅者数据带来的系统开销；
基于事件的Pub/Sub中间件的开发与利用在一定程度上可以提高系统的效率。
在redis当中的实现
Redis-cli连接测试命令行参考附件I
协议测试命令
l 订阅者（redis-cli）连接订阅
/usr/local/bin/redis-cli
-h 10.54.37.212 -p 6380
redis-cli充当订阅者订阅（first 与second）两个topic
SUBSCRIBE first second
l 发布者（redis-cli）连接发布
/usr/local/bin/redis-cli
-h 10.54.37.212 -p 6380
Pubish first wangbin
Publish second wangbin_second
l 订阅者（redis-cli）信息接收
l 取消订阅UNSUBSCRIBE测试
Redis-cli对取消订阅有bug可以使用telnet 进行测试
取消订阅测试成功；
l 匹配模式订阅PSUBSCRIBE测试
publish news.wangbin wangbin_patterncommand
l 匹配模式取消订阅PUNSUBSCRIBE测试
在telnet下测试如下：
PSUBSCRIBE news.*
punsubscribe wangbin*
pecl扩展包目前只提供了两个接口 publish
phpredis是c写的php模块
php这个是php是基于redis
protocol的fsocketopen链接后操作的类库，提供的接口比较全面；publish可以进入数据，但是subscrbie没有阻塞；
可以在原类包当中修改其加入对阻塞模形的支持；
发布功能：
$redis = new Redis();
$res = $redis-&connect($REDIS_HOSTS['CACHE']['host'], $REDIS_HOSTS['CACHE']['port'], 1 );
$res = $redis-&publish($key,$value);
定阅功能：
$redis = new Redis();
$res = $redis-&pconnect($REDIS_HOSTS['CACHE']['host'], $REDIS_HOSTS['CACHE']['port']);
$res = $redis-&subscribe(array($key),array('SinaRedis','subscribe_handler'));
第二个参数为回调方法；
public static function subscribe_handler($redis, $channel, $msg){
print_r($redis);
定阅 redis_subscriber.php
SinaRedis::subscribe('wangbin_test');
发布redis_publisher.php
测试收到订阅者收到发布的内容
Php-redis扩展bug
无论是connect 还是pcconnect 当超时断后，会报错segmentation
超时设置bug
仍会断开解决方法：ini_set('default_socket_timeout', -1);// it works fine
Web Server端测试
Web端订阅应用
Apache下ob_flush
flush后可以输出，但是会漏下最后一条；求解决方案
Nginx下 会报504错误；求解决方法
Nginx 模块与redis直接连接呢?
Php-Fpm有没有什么设置呢
客户端测试
Wget /downloads/andymccurdy/redis-py/redis-2.2.4.tar.gz
redis-2.2.4.tar.gz
cd redis-2.2.4
python setup.py intall
交互模式下测试：
Type "help", "copyright", "credits" or "license" for more information.
&&& import redis
&&& r = redis.Redis('10.54.37.212',6380);
&&& r.publish('wangbin_test','this is a information');
[root@hadoop-master1 python]# python
Python 2.4.3 (#1, Sep
[GCC 4.1.2
(Red Hat 4.1.2-46)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
&&& import redis
&&& r = redis.Redis('10.54.37.212',6380);
&&& r.subscribe('wangbin_test');
&&& for msg in r.listen(): \
{'pattern': None, 'type': 'message', 'channel': 'wangbin_test', 'data': 'this is a information'}
收到json字符串
Publisher.py
Subscribe.py
与node.js结合
Wget http://nodejs.org/dist/v0.6.6/node-v0.6.6.tar.gz
Tar –xvf node-v0.6.6.tar.gz
Cd node-v0.6.6
./configure
Make install
Node 安装在/usr/localbin
添加到环境变量里面：
PATH=/usr/local/bin/:$PATH
sh install.sh
Tar 版本过低，升级下
./configure --prefix=/usr --bindir=/bin --libexecdir=/usr/bin
Make install
Tar –version
tar (GNU tar) 1.26 升级成功
安装socket.io
hiredis redis
/usr/local/bin/npm
install socket.io
/usr/local/bin/npm
install express
/usr/local/bin/npm
install hiredis redis
事件驱动用例：
Publish.js
var redis = require("redis");
var client = redis.createClient(.37.212');
client.on(
function(err){
console.log("err"+err);
client.on('ready',
function(){
client.publish('wangbin_test',"test,i
am test");
client.publish('wangbin_test2',"test,
i am test2");
client.end();
console.log("err:"+e);
Subscribe.js
var redis = require("redis");
var client = redis.createClient(.37.212');
client.on(
function(err){
console.log("err"+err);
client.on('ready',
function(){
client.subscribe('wangbin_test');
client.subscribe('wangbin_test2');
//client.end();
client.on('subscribe',
function(channel,count){
console.log("channel:"
+ channel + ", count:"+count);
client.on('message',
function(channel,message){
console.log("channel:"
+ channel + ", msg:"+message);
client.on('unsubscribe',
function(channel,count){
console.log("channel:"
+ channel + ", count:"+count);
console.log("err:"+e);
测试结果：
另外还支持：pmessage, psubscribe, punsubscribe 事件监听；
---------------------------------------------------------------------------------------------------------
Socket.io redis pub/sub主动推送数据到浏览器前端
另外还支持：pmessage, psubscribe, punsubscribe 事件监听；
var app = require('http').createServer(handler)
, io = require('socket.io').listen(app)
, fs = require('fs');
var redis = require('redis');
var redis_client
= redis.createClient(.37.212');
app.listen(3000);
function handler (req, res) {
fs.readFile(__dirname + '/index.html',
function (err, data) {
if (err) {
res.writeHead(500);
return res.end('Error loading index.html');
res.writeHead(200);
res.end(data);
io.sockets.on('connection', function (socket) {
socket.emit('news', { hello: 'world' });
redis_client.subscribe('wangbin_test');
redis_client.on('message',
function(channel,message){
socket.emit('news',
console.log("channel:"
+ channel + ", msg:"+message);
socket.on('my other event', function (data) {
console.log(data);
Index.html
&script src="/socket.io/socket.io.js"&&/script&
var socket = io.connect('http://10.54.37.212');
socket.on('news', function (data) {
alert(data);
socket.emit('my
other event', { my: 'data' });
Redis-cli发送消息：
Publish wangbin_test test
Server.js订阅监控收到消息后通过socket.io主动push到brower
浏览器收到消息
redis-benchmark是redis的读写性能工具
生成测试数据源：
wangbin_test "'$i'"\r\n' && publish.
for i in $(seq 10);do (cat publish.sleep 100;)
| nc 10.54.37.212
订阅（模拟多个客户端同时订阅）
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_01.txt
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_02.txt
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_03.txt
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_04.txt
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_05.txt
(echo -ne 'subscribe wangbin_test\r\n';sleep 10000)
| nc 10.54.37.212 6380 | grep
& subscribe_06.txt
性能监控测试
watch -d -n 1 'wc -l
subscribe_*.txt '
for i in $(seq 15); do wc -l subscribe_*.txt & result_$i.sleep 1;
测试结果为：
每秒返回的数据为4-5w/s
服务应用场景分析
其他PECL 扩展[SAM]Simple
Asynchronous Messaging
Redis-cli命令信息：
-h &hostname&
端口号 (default:
-s &socket&
例如：/usr/local/bin/redis-cli
/tmp/redis.sock
-a &password&
-r &repeat&
-i &interval&
重复时间隔时间
数据库个数
最后的标准输入
-d &delimiter&
原始格式回复
采集延迟模式
&&&&推荐文章:
【上篇】【下篇】辛星浅析Redis中的pub/sub功能
pub/sub功能也就是publish/subscribe功能，也就是发布订阅功能。基于事件的中，pub/sub是目前广泛使用的通信模型，它采用事件作为基本的通信机制，提供大规模系统所要求的松散耦合的交互模式：订阅者比如客户端以事件订阅的方式表达出它有兴趣接收的一个事件或一类事件，发布者比如服务器可以将订阅者兴趣的事件随时通知相关订阅者。
pub/sub功能可以有三个非耦合：
(1)时间非耦合，也就是发布者和订阅者不必同时在线，它们不必同时参与交互。
(2)空间非耦合，也就是发布者和订阅者不必相互知道对方所在的位置。发布者通过事件服务发布时间，订阅者通过事件服务间接获得事件。发布者和订阅者不需要拥有直接到对方的引用，也不必知道有多少个订阅者或者是发布者参与交互。
(3)同步非耦合，发布/订阅是异步模式。发布者可以不断的产生事件，而订阅者(通过一个回调)则可以异步地得到产生事件的通知。
pub/sub功能按照订阅方式分为基于主题(topic-based)、基于内容(content-based)、基于类型(type-based)的方式。它是可适用于可扩展性高、松散耦合系统的分布式交互模型。在抽象层中，它的时间非耦合、空间非耦合和空间非耦合性可允许参与者不依赖于另一个而独立操作，具有一定的可扩展性。但是在实现的时候，可扩展性依然受到其他原因的牵制。
pub/sub通常适用的范围：
(1)灵活的订阅要求复杂的过滤和路由算法
(2)高可用性开销(事件侦听、日志重传)
(3)消息认可带来的网络流量消耗
(4)庞大的订阅者数据带来的系统开销
基于事件的pub/sub中间件的开发与利用在一定程度上可以提高系统的效率。
Redis中的实现流程:
(1)订阅者订阅两个topic，范例：
(2)发布者发布信息，范例：
(3)订阅者取消一个topic，返利：
unsubscribeHi，亲爱的小伙伴！
欢迎来到社区!
Tools Online | 在线开发工具
RankList | 热门文章
扫码关注 PHP1 官方微信号
Recommend | 推荐阅读
| 中国最专业的PHP中文社区 |
Copyright (C) 1998 - . All Rights Reserved
第一PHP社区