消息传递这一应用广泛存在于各个网站中，这个功能也是一个网站必不可少的。本文主要介绍了php中redis的应用--消息传递。下面跟着小编一起来看下吧
阅读目录
1、摘要
2、实现方法
3、一对一消息传递
4、多对多消息传递
1、摘要
消息传递这一应用广泛存在于各个网站中，这个功能也是一个网站必不可少的。常见的消息传递应用有，新浪微博中的@我呀、给你评论然后的提示呀、赞赞赞提示、私信呀、甚至是发微博分享的新鲜事；知乎中的私信呀、live发送过来的消息、知乎团队消息呀等等。
2、实现方法
消息传递即两个或者多个客户端在相互发送和接收消息。
通常有两种方法实现：
第一种为消息推送。redis内置有这种机制，publish往频道推送消息、subscribe订阅频道。这种方法有一个缺点就是必须保证接收者时刻在线（即是此时程序不能停下来，一直保持监控状态，假若断线后就会出现客户端丢失信息）
第二种为消息拉取。所谓消息拉取，就是客户端自主去获取存储在服务器中的数据。redis内部没有实现消息拉取这种机制。因此我们需要自己手动编写代码去实现这个功能。
在这里我们，我们进一步将消息传递再细分为一对一的消息传递，多对多的消息传递（群组消息传递）。
【注：两个类的代码相对较多，因此将其折叠起来了】
3、一对一消息传递
例子1：一对一消息发送与获取
模块要求：
1、提示有多少个联系人发来新消息
2、信息包含发送人、时间、信息内容
3、能够获取之前的旧消息
4、并且消息能够保持7天，过期将会被动触发删除
redis实现思路：
1、新消息与旧消息分别采用两个链表来存储
2、原始消息的结构采用数组的形式存放，并且含有发送人、时间戳、信息内容
3、在推入redis的链表前，需要将数据转换为json类型然后再进行存储
4、在取出新信息时应该使用rpoplpush来实现，将已读的新消息推入旧消息链表中
5、取出旧消息时，应该用旧消息的时间与现在的时间进行对比，若超时，则直接删除后面的全部数据（因为数据是按时间一个一个压进链表中的，所以对于时间是有序排列的）
数据存储结构图：
php的实现代码：
#singlepullmessage.class.php
<?php
#单接接收者接收消息
class singlepullmessage
{
private $redis=''; #存储redis对象
/**
* @desc 构造函数
*
* @param $host string | redis主机
* @param $port int | 端口
*/
public function construct($host,$port=6379)
{
$this->redis=new redis();
$this->redis->connect($host,$port);
}
/**
* @desc 发送消息（一个人）
*
* @param $touser string | 接收人
* @param $messagearr array | 发送的消息数组，包含sender、message、time
*
* @return bool
*/
public function sendsingle($touser,$messagearr)
{
$json_message=json_encode($messagearr); #编码成json数据
return $this->redis->lpush($touser,$json_message); #将数据推入链表
}
/**
* @desc 用户获取新消息
*
* @param $user string | 用户名
*
* @return array 返回数组，包含多少个用户发来新消息，以及具体消息
*/
public function getnewmessage($user)
{
#接收新信息数据，并且将数据推入旧信息数据链表中，并且在原链表中删除
$messagearr=array();
while($json_message=$this->redis->rpoplpush($user, 'premessage_'.$user))
{
$temp=json_decode($json_message); #将json数据变成对象
$messagearr[$temp->sender][]=$temp; #转换成数组信息
}
if($messagearr)
{
$arr['count']=count($messagearr); #统计有多少个用户发来信息
$arr['messagearr']=$messagearr;
return $arr;
}
return false;
}
public function getpremessage($user)
{
##取出旧消息
$messagearr=array();
$json_pre=$this->redis->lrange('premessage_'.$user, 0, -1); #一次性将全部旧消息取出来
foreach ($json_pre as $k => $v)
{
$temp=json_decode($v); #json反编码
$timeout=$temp->time+60*60*24*7; #数据过期时间 七天过期
if($timeout<time()) #判断数据是否过期
{
if($k==0) #若是最迟插入的数据都过期了，则将所有数据删除
{
$this->redis->del('premessage_'.$user);
break;
}
$this->redis->ltrim('premessage_'.$user, 0, $k); #若检测出有过期的，则将比它之前插入的所有数据删除
break;
}
$messagearr[$temp->sender][]=$temp;
}
return $messagearr;
}
/**
* @desc 消息处理，没什么特别的作用。在这里这是用来处理数组信息，然后将其输出。
*
* @param $arr array | 需要处理的信息数组
*
* @return 返回打印输出
*/
public function dealarr($arr)
{
foreach ($arr as $k => $v)
{
foreach ($v as $k1 => $v2)
{
echo '发送人：'.$v2->sender.' 发送时间：'.date('y-m-d h:i:s',$v2->time).'<br/>';
echo '消息内容：'.$v2->message.'<br/>';
}
echo "<hr/>";
}
}
}
测试：
1、发送消息
#建立test1.php
include './singlepullmessage.class.php';
$object=new singlepullmessage('192.168.95.11');
#发送消息
$sender='boss'; #发送者
$to='jane'; #接收者
$message='how are you'; #信息
$time=time();
$arr=array('sender'=>$sender,'message'=>$message,'time'=>$time);
echo $object->sendsingle($to,$arr);
2、获取新消息
#建立test2.php
include './singlepullmessage.class.php';
$object=new singlepullmessage('192.168.95.11');
#获取新消息
$arr=$object->getnewmessage('jane');
if($arr)
{
echo $arr['count']."个联系人发来新消息<br/><hr/>";
$object->dealarr($arr['messagearr']);
}
else
echo "无新消息";
访问结果:
3、获取旧消息
#建立test3.php
include './singlepullmessage.class.php';
$object=new singlepullmessage('192.168.95.11');
#获取旧消息
$arr=$object->getpremessage('jane');
if($arr)
{
$object->dealarr($arr);
}
else
echo "无旧数据";
4、多对多消息传递
例子2：多对多消息发送与获取（即是群组）
模块要求：
1、用户能够自行创建群组，并成为群主
2、群主可以拉人进来作为群组成员、并且可以踢人
3、用户可以直接退出群组
4、可以发送消息，每一位成员都可以拉取消息
5、群组的消息最大容纳量为5000条
6、成员可以拉取新消息，并提示有多少新消息
7、成员可以分页获取之前已读的旧消息
。。。。。功能就写这几个吧，有需要或者想练习的同学们可以增加其他功能，例如禁言、匿名消息发送、文件发送等等。
redis实现思路：
1、群组的消息以及群组的成员组成采用有序集合进行存储。群组消息有序集合的member存储用户发送的json数据消息，score存储唯一值，将采用原子操作incr获取string中的自增长值进行存储；群组成员有序集合的member存储user，score存储非零数字（在这里这个score意义不大，我的例子代码中使用数字1为群主的score，其他的存储为2。当然这使用这个数据还可以扩展别的功能，例如群组中成员等级）可参考下面数据存储结构简图。
2、用户所加入的群组也是采用有序集合进行存储。其中，member存储群组id，score存储用户已经获取该群组的最大消息分值（对应群组消息的score值）
3、用户创建群组的时候，通过原子操作incr从而获取一个唯一id
4、用户在群中发送消息时，也是通过原子操作incr获取一个唯一自增长有序id
5、在执行incr时，为防止并发导致竞争关系，因此需要进行加锁操作【redis详细锁的讲解可以参考：redis构建分布式锁http://www.jb51.net/article/109704.htm】
6、创建群组方法简要思路，任何一个用户都可以创建群组聊天，在创建的同时，可以选择时是否添加群组成员（参数通过数组的形式）。创建过程将会为这个群组建立一个群组成员有序集合（群组信息有序集合暂时不创建），接着将群主添加进去，再将群id添加用户所参加的群组有序集合中。
数据存储结构图：
php的代码实现：
#manypullmessage.class.php
<?php
class manypullmessage
{
private $redis=''; #存储redis对象
/**
* @desc 构造函数
*
* @param $host string | redis主机
* @param $port int | 端口
*/
public function construct($host,$port=6379)
{
$this->redis=new redis();
$this->redis->connect($host,$port);
}
/**
* @desc 用于创建群组的方法，在创建的同时还可以拉人进群组
*
* @param $user string | 用户名，创建群组的主人
* @param $adduser array | 其他用户构成的数组
*
* @param $lockname string | 锁的名字，用于获取群组id的时候用
* @return int 返回群组id
*/
public function creategroupchat($user, $adduser=array(), $lockname='chatidlock')
{
$identifier=$this->getlock($lockname); #获取锁
if($identifier)
{
$id=$this->redis->incr('groupchatid'); #获取群组id
$this->releaselock($lockname,$identifier); #释放锁
}
else
return false;
$messagecount=$this->redis->set('countmessage_'.$id, 0); #初始化这个群组消息计数器
#开启非事务型流水线，一次性将所有redis命令传给redis，减少与redis的连接
$pipe=$this->redis->pipeline();
$this->redis->zadd('groupchat_'.$id, 1, $user); #创建群组成员有序集合，并添加群主
#将这个群组添加到user所参加的群组有序集合中
$this->redis->zadd('hasgroupchat_'.$user, 0, $id);
foreach ($adduser as $v) #创建群组的同时需要添加的用户成员
{
$this->redis->zadd('groupchat_'.$id, 2, $v);
$this->redis->zadd('hasgroupchat_'.$v, 0, $id);
}
$pipe->exec();
return $id; #返回群组id
}
/**
* @desc 群主主动拉人进群
*
* @param $user string | 群主名
* @param $groupchatid int | 群组id
* @param $addmembers array | 需要拉进群的用户
*
* @return bool
*/
public function addmembers($user, $groupchatid, $addmembers=array())
{
$groupmasterscore=$this->redis->zscore('groupchat_'.$groupchatid, $user); #将groupchatname的群主取出来
if($groupmasterscore==1) #判断user是否是群主
{
$pipe=$this->redis->pipeline(); #开启非事务流水线
foreach ($addmembers as $v)
{
$this->redis->zadd('groupchat_'.$groupchatid, 2, $v); #添加进群
$this->redis->zadd('hasgroupchat_'.$v, 0, $groupchatid); #添加群名到用户的有序集合中
}
$pipe->exec();
return true;
}
return false;
}
/**
* @desc 群主删除成员
*
* @param $user string | 群主名
* @param $groupchatid int | 群组id
* @param $delmembers array | 需要删除的成员名字
*
* @return bool
*/
public function delmembers($user, $groupchatid, $delmembers=array())
{
$groupmasterscore=$this->redis->zscore('groupchat_'.$groupchatid, $user);
if($groupmasterscore==1) #判断user是否是群主
{
$pipe=$this->redis->pipeline(); #开启非事务流水线
foreach ($delmembers as $v)
{
$this->redis->zrem('groupchat_'.$groupchatid, $v);
$this->redis->zrem('hasgroupchat_'.$v, $groupchatid);
}
$pipe->exec();
return true;
}
return false;
}
/**
* @desc 退出群组
*
* @param $user string | 用户名
* @param $groupchatid int | 群组名
*/
public function quitgroupchat($user, $groupchatid)
{
$this->redis->zrem('groupchat_'.$groupchatid, $user);
$this->redis->zrem('hasgroupchat_'.$user, $groupchatid);
return true;
}
/**
* @desc 发送消息
*
* @param $user string | 用户名
* @param $groupchatid int | 群组id
* @param $messagearr array | 包含发送消息的数组
* @param $prelockname string | 群消息锁前缀，群消息锁全名为countlock_群id
*
* @return bool
*/
public function sendmessage($user, $groupchatid, $messagearr, $prelockname='countlock_')
{
$memberscore=$this->redis->zscore('groupchat_'.$groupchatid, $user); #成员score
if($memberscore)
{
$identifier=$this->getlock($prelockname.$groupchatid); #获取锁
if($identifier) #判断获取锁是否成功
{
$messagecount=$this->redis->incr('countmessage_'.$groupchatid);
$this->releaselock($prelockname.$groupchatid,$identifier); #释放锁
}
else
return false;
$json_message=json_encode($messagearr);
$this->redis->zadd('groupchatmessage_'.$groupchatid, $messagecount, $json_message);
$count=$this->redis->zcard('groupchatmessage_'.$groupchatid); #查看信息量大小
if($count>5000) #判断数据量有没有达到5000条
{ #数据量超5000，则需要清除旧数据
$start=5000-$count;
$this->redis->zremrangebyrank('groupchatmessage_'.$groupchatid, $start, $count);
}
return true;
}
return false;
}
/**
* @desc 获取新信息
*
* @param $user string | 用户名
*
* @return 成功则放回json数据数组，无新信息返回false
*/
public function getnewmessage($user)
{
$arrid=$this->redis->zrange('hasgroupchat_'.$user, 0, -1, 'withscores'); #获取用户拥有的群组id
$json_message=array(); #初始化
foreach ($arrid as $k => $v) #遍历循环所有群组，查看是否有新消息
{
$messagecount=$this->redis->get('countmessage_'.$k); #群组最大信息分值数
if($messagecount>$v) #判断用户是否存在未读新消息
{
$json_message[$k]['message']=$this->redis->zrangebyscore('groupchatmessage_'.$k, $v+1, $messagecount);
$json_message[$k]['count']=count($json_message[$k]['message']); #统计新消息数量
$this->redis->zadd('hasgroupchat_'.$user, $messagecount, $k); #更新已获取消息
}
}
if($json_message)
return $json_message;
return false;
}
/**
* @desc 分页获取群组信息
*
* @param $user string | 用户名
* @param $groupchatid int | 群组id
* @param $page int | 第几页
* @param $size int | 每页多少条数据
*
* @return 成功返回json数据，失败返回false
*/
public function getpartmessage($user, $groupchatid, $page=1, $size=10)
{
$start=$page*$size-$size; #开始截取数据位置
$stop=$page*$size-1; #结束截取数据位置
$json_message=$this->redis->zrevrange('groupchatmessage_'.$groupchatid, $start, $stop);
if($json_message)
return $json_message;
return false;
}
/**
* @desc 加锁方法
*
* @param $lockname string | 锁的名字
* @param $timeout int | 锁的过期时间
*
* @return 成功返回identifier/失败返回false
*/
public function getlock($lockname, $timeout=2)
{
$identifier=uniqid(); #获取唯一标识符
$timeout=ceil($timeout); #确保是整数
$end=time()+$timeout;
while(time()<$end) #循环获取锁
{
/*
#这里的set操作可以等同于下面那个if操作，并且可以减少一次与redis通讯
if($this->redis->set($lockname, $identifier array('nx', 'ex'=>$timeout)))
return $identifier;
*/
if($this->redis->setnx($lockname, $identifier)) #查看$lockname是否被上锁
{
$this->redis->expire($lockname, $timeout); #为$lockname设置过期时间
return $identifier; #返回一维标识符
}
elseif ($this->redis->ttl($lockname)===-1)
{
$this->redis->expire($lockname, $timeout); #检测是否有设置过期时间，没有则加上
}
usleep(0.001); #停止0.001ms
}
return false;
}
/**
* @desc 释放锁
*
* @param $lockname string | 锁名
* @param $identifier string | 锁的唯一值
*
* @param bool
*/
public function releaselock($lockname,$identifier)
{
if($this->redis->get($lockname)==$identifier) #判断是锁有没有被其他客户端修改
{
$this->redis->multi();
$this->redis->del($lockname); #释放锁
$this->redis->exec();
return true;
}
else
{
return false; #其他客户端修改了锁，不能删除别人的锁
}
}
}
?>
测试：
1、建立creategroupchat.php(测试创建群组功能)
执行代码并创建568、569群组（群主为jack）
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
#创建群组
$user='jack';
$arr=array('jane1','jane2');
$a=$object->creategroupchat($user,$arr);
echo "<pre>";
print_r($a);
echo "</pre>";die;
2、建立addmembers.php(测试添加成员功能)
执行代码并添加新成员
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
$b=$object->addmembers('jack','568',array('jane1','jane2','jane3','jane4'));
echo "<pre>";
print_r($b);
echo "</pre>";die;
3、建立delete.php(测试群主删除成员功能)
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
#群主删除成员
$c=$object->delmembers('jack', '568', array('jane1','jane4'));
echo "<pre>";
print_r($c);
echo "</pre>";die;
4、建立sendmessage.php(测试发送消息功能)
多执行几遍，568、569都发几条
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
#发送消息
$user='jane2';
$message='go go go';
$groupchatid=568;
$arr=array('sender'=>$user, 'message'=>$message, 'time'=>time());
$d=$object->sendmessage($user,$groupchatid,$arr);
echo "<pre>";
print_r($d);
echo "</pre>";die;
5、建立getnewmessage.php(测试用户获取新消息功能)
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
#用户获取新消息
$e=$object->getnewmessage('jane2');
echo "<pre>";
print_r($e);
echo "</pre>";die;
6、建立getpartmessage.php(测试用户获取某个群组部分消息)
（多发送几条消息，用于测试。568中共18条数据）
include './manypullmessage.class.php';
$object=new manypullmessage('192.168.95.11');
#用户获取某个群组部分消息
$f=$object->getpartmessage('jane2', 568, 1, 10);
echo "<pre>";
print_r($f);
echo "</pre>";die;
page=1，size=10
page=2，size=10
测试完毕，还需要别的功能可以自己进行修改添加测试。
以上就是php中redis的应用-消息传递的示例代码分享（图）的详细内容。