php中Redis的应用--消息传递

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,-1); #一次性将全部旧消息取出来 foreach ($json_pre as $k => $v) { $temp=json_decode($v); #json反编码 $timeout=$temp->time+60*60*24*7; #数据过期时间 七天过期 if($timeoutredis->del('preMessage_'.$user); break; } $this->redis->ltrim('preMessage_'.$user,$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).'
'; echo '消息内容：'.$v2->message.'
'; } echo "

---

"; } } }

$sender,'message'=>$message,'time'=>$time); echo $object->sendSingle($to,$arr);

getNewMessage('jane'); if($arr) { echo $arr['count']."个联系人发来新消息

---

"; $object->dealArr($arr['messageArr']); } else echo "无新消息";

getPreMessage('jane'); if($arr) { $object->dealArr($arr); } else echo "无旧数据";

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,$id); foreach ($addUser as $v) #创建群组的同时需要添加的用户成员 { $this->redis->zadd('groupChat_'.$id,2,$v); $this->redis->zadd('hasGroupChat_'.$v,$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,$v); #添加进群 $this->redis->zadd('hasGroupChat_'.$v,$groupChatID); #添加群名到用户的有序集合中 } $pipe->exec(); return true; } return false; } /** * @desc 群主删除成员 * * @param $user string | 群主名 * @param $groupChatID int | 群组ID * @param $delMembers array | 需要删除的成员名字 * * @return bool */ public function delMembers($user,$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,$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,-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,$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,$page=1,$size=10) { $start=$page*$size-$size; #开始截取数据位置 $stop=$page*$size-1; #结束截取数据位置 $json_message=$this->redis->zrevrange('groupChatMessage_'.$groupChatID,$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; #其他客户端修改了锁，不能删除别人的锁 } }

}
?>

createGroupChat($user,$arr); echo "

";
print_r($a);
echo "

";die;

addMembers('jack','568',array('jane1','jane2','jane3','jane4')); echo "

";
 print_r($b);
 echo "

";die;

delMembers('jack','jane4')); echo "

";
print_r($c);
echo "

";die;

$user,'time'=>time()); $d=$object->sendMessage($user,$arr); echo "

";
print_r($d);
echo "

";die;

getNewMessage('jane2'); echo "

";
print_r($e);
echo "

";die;

getPartMessage('jane2',568,10); echo "

";
print_r($f);
echo "

";die;