php中Redis應(yīng)用之消息傳遞的示例分析

這篇文章給大家分享的是有關(guān)php中Redis應(yīng)用之消息傳遞的示例分析的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過來看看吧。

1、摘要

消息傳遞這一應(yīng)用廣泛存在于各個(gè)網(wǎng)站中，這個(gè)功能也是一個(gè)網(wǎng)站必不可少的。常見的消息傳遞應(yīng)用有，新浪微博中的@我呀、給你評(píng)論然后的提示呀、贊贊贊提示、私信呀、甚至是發(fā)微博分享的新鮮事；知乎中的私信呀、live發(fā)送過來的消息、知乎團(tuán)隊(duì)消息呀等等。

2、實(shí)現(xiàn)方法

消息傳遞即兩個(gè)或者多個(gè)客戶端在相互發(fā)送和接收消息。

通常有兩種方法實(shí)現(xiàn)：

第一種為消息推送。Redis內(nèi)置有這種機(jī)制，publish往頻道推送消息、subscribe訂閱頻道。這種方法有一個(gè)缺點(diǎn)就是必須保證接收者時(shí)刻在線（即是此時(shí)程序不能停下來，一直保持監(jiān)控狀態(tài)，假若斷線后就會(huì)出現(xiàn)客戶端丟失信息）

第二種為消息拉取。所謂消息拉取，就是客戶端自主去獲取存儲(chǔ)在服務(wù)器中的數(shù)據(jù)。Redis內(nèi)部沒有實(shí)現(xiàn)消息拉取這種機(jī)制。因此我們需要自己手動(dòng)編寫代碼去實(shí)現(xiàn)這個(gè)功能。

在這里我們，我們進(jìn)一步將消息傳遞再細(xì)分為一對(duì)一的消息傳遞，多對(duì)多的消息傳遞（群組消息傳遞）。

【注：兩個(gè)類的代碼相對(duì)較多，因此將其折疊起來了】

3、一對(duì)一消息傳遞

例子1：一對(duì)一消息發(fā)送與獲取

模塊要求：

1、提示有多少個(gè)聯(lián)系人發(fā)來新消息

2、信息包含發(fā)送人、時(shí)間、信息內(nèi)容

3、能夠獲取之前的舊消息

4、并且消息能夠保持7天，過期將會(huì)被動(dòng)觸發(fā)刪除

Redis實(shí)現(xiàn)思路：

1、新消息與舊消息分別采用兩個(gè)鏈表來存儲(chǔ)

2、原始消息的結(jié)構(gòu)采用數(shù)組的形式存放，并且含有發(fā)送人、時(shí)間戳、信息內(nèi)容

3、在推入redis的鏈表前，需要將數(shù)據(jù)轉(zhuǎn)換為json類型然后再進(jìn)行存儲(chǔ)

4、在取出新信息時(shí)應(yīng)該使用rpoplpush來實(shí)現(xiàn)，將已讀的新消息推入舊消息鏈表中

5、取出舊消息時(shí)，應(yīng)該用舊消息的時(shí)間與現(xiàn)在的時(shí)間進(jìn)行對(duì)比，若超時(shí)，則直接刪除后面的全部數(shù)據(jù)（因?yàn)閿?shù)據(jù)是按時(shí)間一個(gè)一個(gè)壓進(jìn)鏈表中的，所以對(duì)于時(shí)間是有序排列的）

數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)圖：

PHP的實(shí)現(xiàn)代碼：

#SinglePullMessage.class.php

<?php
#單接接收者接收消息
class SinglePullMessage
{
 private $redis=''; #存儲(chǔ)redis對(duì)象
 /**
 * @desc 構(gòu)造函數(shù)
 * 
 * @param $host string | redis主機(jī)
 * @param $port int | 端口
 */
 public function __construct($host,$port=6379)
 {
 $this->redis=new Redis();
 $this->redis->connect($host,$port);
 } 
 /**
 * @desc 發(fā)送消息（一個(gè)人）
 * 
 * @param $toUser string | 接收人
 * @param $messageArr array | 發(fā)送的消息數(shù)組，包含sender、message、time 
 *
 * @return bool
 */
 public function sendSingle($toUser,$messageArr)
 {
 $json_message=json_encode($messageArr); #編碼成json數(shù)據(jù)
 return $this->redis->lpush($toUser,$json_message); #將數(shù)據(jù)推入鏈表 
 }
 /**
 * @desc 用戶獲取新消息
 *
 * @param $user string | 用戶名
 *
 * @return array 返回?cái)?shù)組，包含多少個(gè)用戶發(fā)來新消息，以及具體消息
 */
 public function getNewMessage($user)
 {
 #接收新信息數(shù)據(jù)，并且將數(shù)據(jù)推入舊信息數(shù)據(jù)鏈表中，并且在原鏈表中刪除
 $messageArr=array();
 while($json_message=$this->redis->rpoplpush($user, 'preMessage_'.$user))
 {
  $temp=json_decode($json_message); #將json數(shù)據(jù)變成對(duì)象
  $messageArr[$temp->sender][]=$temp; #轉(zhuǎn)換成數(shù)組信息
 }
 if($messageArr)
 {
  $arr['count']=count($messageArr); #統(tǒng)計(jì)有多少個(gè)用戶發(fā)來信息
  $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; #數(shù)據(jù)過期時(shí)間 七天過期
  if($timeout<time())  #判斷數(shù)據(jù)是否過期
  {
  if($k==0)   #若是最遲插入的數(shù)據(jù)都過期了，則將所有數(shù)據(jù)刪除
  {
   $this->redis->del('preMessage_'.$user);
   break;
  }
  $this->redis->ltrim('preMessage_'.$user, 0, $k); #若檢測出有過期的，則將比它之前插入的所有數(shù)據(jù)刪除
  break;
  }
  $messageArr[$temp->sender][]=$temp;
 }
 return $messageArr;
 }
 /**
 * @desc 消息處理，沒什么特別的作用。在這里這是用來處理數(shù)組信息，然后將其輸出。 
 *
 * @param $arr array | 需要處理的信息數(shù)組
 *
 * @return 返回打印輸出
 */
 public function dealArr($arr)
 {
 foreach ($arr as $k => $v) 
 {
  foreach ($v as $k1 => $v2) 
  {
  echo '發(fā)送人：'.$v2->sender.' 發(fā)送時(shí)間：'.date('Y-m-d h:i:s',$v2->time).'<br/>';
  echo '消息內(nèi)容：'.$v2->message.'<br/>';
  }
  echo "<hr/>";
 }
 }
}

測試：

1、發(fā)送消息

#建立test1.php

include './SinglePullMessage.class.php';
$object=new SinglePullMessage('192.168.95.11');
#發(fā)送消息
$sender='boss'; #發(fā)送者
$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']."個(gè)聯(lián)系人發(fā)來新消息<br/><hr/>";
 $object->dealArr($arr['messageArr']); 
}
else
 echo "無新消息";

訪問結(jié)果:

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 "無舊數(shù)據(jù)";

4、多對(duì)多消息傳遞

例子2：多對(duì)多消息發(fā)送與獲?。词侨航M）

模塊要求：

1、用戶能夠自行創(chuàng)建群組，并成為群主

2、群主可以拉人進(jìn)來作為群組成員、并且可以踢人

3、用戶可以直接退出群組

4、可以發(fā)送消息，每一位成員都可以拉取消息

5、群組的消息最大容納量為5000條

6、成員可以拉取新消息，并提示有多少新消息

7、成員可以分頁獲取之前已讀的舊消息

。。。。。功能就寫這幾個(gè)吧，有需要或者想練習(xí)的同學(xué)們可以增加其他功能，例如禁言、匿名消息發(fā)送、文件發(fā)送等等。

Redis實(shí)現(xiàn)思路：

1、群組的消息以及群組的成員組成采用有序集合進(jìn)行存儲(chǔ)。群組消息有序集合的member存儲(chǔ)用戶發(fā)送的json數(shù)據(jù)消息，score存儲(chǔ)唯一值，將采用原子操作incr獲取string中的自增長值進(jìn)行存儲(chǔ)；群組成員有序集合的member存儲(chǔ)user，score存儲(chǔ)非零數(shù)字（在這里這個(gè)score意義不大，我的例子代碼中使用數(shù)字1為群主的score，其他的存儲(chǔ)為2。當(dāng)然這使用這個(gè)數(shù)據(jù)還可以擴(kuò)展別的功能，例如群組中成員等級(jí)）可參考下面數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)簡圖。

2、用戶所加入的群組也是采用有序集合進(jìn)行存儲(chǔ)。其中，member存儲(chǔ)群組ID，score存儲(chǔ)用戶已經(jīng)獲取該群組的最大消息分值（對(duì)應(yīng)群組消息的score值）

3、用戶創(chuàng)建群組的時(shí)候，通過原子操作incr從而獲取一個(gè)唯一ID

4、用戶在群中發(fā)送消息時(shí)，也是通過原子操作incr獲取一個(gè)唯一自增長有序ID

5、在執(zhí)行incr時(shí)，為防止并發(fā)導(dǎo)致競爭關(guān)系，因此需要進(jìn)行加鎖操作【redis詳細(xì)鎖的講解可以參考：Redis構(gòu)建分布式鎖https://www.jb51.net/article/109704.htm】

6、創(chuàng)建群組方法簡要思路，任何一個(gè)用戶都可以創(chuàng)建群組聊天，在創(chuàng)建的同時(shí)，可以選擇時(shí)是否添加群組成員（參數(shù)通過數(shù)組的形式）。創(chuàng)建過程將會(huì)為這個(gè)群組建立一個(gè)群組成員有序集合（群組信息有序集合暫時(shí)不創(chuàng)建），接著將群主添加進(jìn)去，再將群ID添加用戶所參加的群組有序集合中。

數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)圖：

PHP的代碼實(shí)現(xiàn)：

#ManyPullMessage.class.php

<?php
class ManyPullMessage
{
 private $redis=''; #存儲(chǔ)redis對(duì)象
 /**
 * @desc 構(gòu)造函數(shù)
 * 
 * @param $host string | redis主機(jī)
 * @param $port int | 端口
 */
 public function __construct($host,$port=6379)
 {
 $this->redis=new Redis();
 $this->redis->connect($host,$port);
 } 
 /**
 * @desc 用于創(chuàng)建群組的方法，在創(chuàng)建的同時(shí)還可以拉人進(jìn)群組
 * 
 * @param $user string | 用戶名，創(chuàng)建群組的主人
 * @param $addUser array | 其他用戶構(gòu)成的數(shù)組
 *
 * @param $lockName string | 鎖的名字，用于獲取群組ID的時(shí)候用
 * @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); #初始化這個(gè)群組消息計(jì)數(shù)器
 #開啟非事務(wù)型流水線，一次性將所有redis命令傳給redis，減少與redis的連接
 $pipe=$this->redis->pipeline(); 
 $this->redis->zadd('groupChat_'.$id, 1, $user); #創(chuàng)建群組成員有序集合，并添加群主
 #將這個(gè)群組添加到user所參加的群組有序集合中
 $this->redis->zadd('hasGroupChat_'.$user, 0, $id); 
 foreach ($addUser as $v) #創(chuàng)建群組的同時(shí)需要添加的用戶成員
 {
  $this->redis->zadd('groupChat_'.$id, 2, $v);
  $this->redis->zadd('hasGroupChat_'.$v, 0, $id);
 }
 $pipe->exec();
 return $id; #返回群組ID
 }
 /**
 * @desc 群主主動(dòng)拉人進(jìn)群
 *
 * @param $user string | 群主名
 * @param $groupChatID int | 群組ID
 * @param $addMembers array | 需要拉進(jìn)群的用戶
 *
 * @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(); #開啟非事務(wù)流水線
  foreach ($addMembers as $v) 
  {
  $this->redis->zadd('groupChat_'.$groupChatID, 2, $v);   #添加進(jìn)群
  $this->redis->zadd('hasGroupChat_'.$v, 0, $groupChatID); #添加群名到用戶的有序集合中
  }
  $pipe->exec();
  return true;
 }
 return false;
 }
 /**
 * @desc 群主刪除成員
 *
 * @param $user string | 群主名
 * @param $groupChatID int | 群組ID
 * @param $delMembers array | 需要?jiǎng)h除的成員名字
 *
 * @return bool
 */
 public function delMembers($user, $groupChatID, $delMembers=array())
 {
 $groupMasterScore=$this->redis->zscore('groupChat_'.$groupChatID, $user); 
 if($groupMasterScore==1) #判斷user是否是群主
 {
  $pipe=$this->redis->pipeline(); #開啟非事務(wù)流水線
  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 發(fā)送消息
 *
 * @param $user string | 用戶名
 * @param $groupChatID int | 群組ID
 * @param $messageArr array | 包含發(fā)送消息的數(shù)組
 * @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) #判斷數(shù)據(jù)量有沒有達(dá)到5000條
  { #數(shù)據(jù)量超5000，則需要清除舊數(shù)據(jù)
  $start=5000-$count;
  $this->redis->zremrangebyrank('groupChatMessage_'.$groupChatID, $start, $count);
  }
  return true;
 }
 return false;
 }
 /**
 * @desc 獲取新信息
 *
 * @param $user string | 用戶名
 *
 * @return 成功則放回json數(shù)據(jù)數(shù)組，無新信息返回false
 */
 public function getNewMessage($user)
 {
 $arrID=$this->redis->zrange('hasGroupChat_'.$user, 0, -1, 'withscores'); #獲取用戶擁有的群組ID
 $json_message=array(); #初始化
 foreach ($arrID as $k => $v) #遍歷循環(huán)所有群組，查看是否有新消息
 {
  $messageCount=$this->redis->get('countMessage_'.$k); #群組最大信息分值數(shù)
  if($messageCount>$v) #判斷用戶是否存在未讀新消息
  {
  $json_message[$k]['message']=$this->redis->zrangebyscore('groupChatMessage_'.$k, $v+1, $messageCount);
  $json_message[$k]['count']=count($json_message[$k]['message']); #統(tǒng)計(jì)新消息數(shù)量
  $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 | 每頁多少條數(shù)據(jù)
 *
 * @return 成功返回json數(shù)據(jù)，失敗返回false
 */
 public function getPartMessage($user, $groupChatID, $page=1, $size=10)
 {
 $start=$page*$size-$size; #開始截取數(shù)據(jù)位置
 $stop=$page*$size-1; #結(jié)束截取數(shù)據(jù)位置
 $json_message=$this->redis->zrevrange('groupChatMessage_'.$groupChatID, $start, $stop);
 if($json_message)
  return $json_message;
 return false;
 }
 /**
 * @desc 加鎖方法
 *
 * @param $lockName string | 鎖的名字
 * @param $timeout int | 鎖的過期時(shí)間
 *
 * @return 成功返回identifier/失敗返回false
 */
 public function getLock($lockName, $timeout=2)
 {
 $identifier=uniqid(); #獲取唯一標(biāo)識(shí)符
 $timeout=ceil($timeout); #確保是整數(shù)
 $end=time()+$timeout;
 while(time()<$end)  #循環(huán)獲取鎖
 {
  /*
  #這里的set操作可以等同于下面那個(gè)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設(shè)置過期時(shí)間
  return $identifier;    #返回一維標(biāo)識(shí)符
  }
  elseif ($this->redis->ttl($lockName)===-1) 
  {
  $this->redis->expire($lockName, $timeout); #檢測是否有設(shè)置過期時(shí)間，沒有則加上
  }
  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(測試創(chuàng)建群組功能)

執(zhí)行代碼并創(chuàng)建568、569群組（群主為jack）

include './ManyPullMessage.class.php';
$object=new ManyPullMessage('192.168.95.11');
#創(chuàng)建群組
$user='jack';
$arr=array('jane1','jane2');
$a=$object->createGroupChat($user,$arr);
echo "<pre>";
print_r($a);
echo "</pre>";die;

2、建立addMembers.php(測試添加成員功能)

執(zhí)行代碼并添加新成員

 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(測試發(fā)送消息功能)

多執(zhí)行幾遍，568、569都發(fā)幾條

include './ManyPullMessage.class.php';
$object=new ManyPullMessage('192.168.95.11');
#發(fā)送消息
$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(測試用戶獲取某個(gè)群組部分消息)

（多發(fā)送幾條消息，用于測試。568中共18條數(shù)據(jù)）

include './ManyPullMessage.class.php';
$object=new ManyPullMessage('192.168.95.11');
#用戶獲取某個(gè)群組部分消息
$f=$object->getPartMessage('jane2', 568, 1, 10); 
echo "<pre>";
print_r($f);
echo "</pre>";die;

page=1，size=10

page=2，size=10

測試完畢，還需要?jiǎng)e的功能可以自己進(jìn)行修改添加測試。

感謝各位的閱讀！關(guān)于“php中Redis應(yīng)用之消息傳遞的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！