Detaillierte Erläuterung der Anwendungsnachrichten von Redis in PHP-PHP-Tutorial-php.cn

Messaging wird auf verschiedenen Websites häufig verwendet, und diese Funktion ist auch für eine Website unerlässlich. In diesem Artikel wird hauptsächlich die Anwendung von Redis bei der PHP-Nachrichtenübermittlung vorgestellt.

Inhalt lesen

1. Zusammenfassung

2 Methode

3. Eins-zu-Eins-Nachrichten

4 -Viele Nachrichtenübermittlung

1. Zusammenfassung

Die Nachrichtenübermittlung wird häufig auf verschiedenen Websites verwendet, und diese Funktion ist auch für eine Website unerlässlich. Zu den gängigen Messaging-Anwendungen gehören @me auf Sina Weibo, Eingabeaufforderungen nach dem Kommentieren, wie Eingabeaufforderungen, private Nachrichten und sogar neue Dinge, die auf Weibo gesendet und live auf Zhihu gesendet werden, Nachrichten des Zhihu-Teams usw.

2. Implementierungsmethode

Message Passing bedeutet, dass zwei oder mehr Clients einander Nachrichten senden und empfangen.

Es gibt normalerweise zwei Möglichkeiten, es umzusetzen:

Die erste ist Message Push. Redis verfügt über diesen integrierten Mechanismus, veröffentlicht Push-Nachrichten an Kanäle und abonniert Kanäle. Ein Nachteil dieser Methode besteht darin, dass der Empfänger jederzeit online sein muss (d. h. das Programm kann zu diesem Zeitpunkt nicht gestoppt werden und bleibt im Überwachungszustand. Wenn die Verbindung getrennt wird, verliert der Client Informationen)

Der zweite Typ ist das Message Pulling. Das sogenannte Message Pulling bedeutet, dass der Client selbstständig die im Server gespeicherten Daten erhält. Redis implementiert intern keinen Message-Pull-Mechanismus. Daher müssen wir den Code selbst manuell schreiben, um diese Funktion zu implementieren.

Hier unterteilen wir Messaging weiter in Eins-zu-Eins-Messaging und Viele-zu-Viele-Messaging (Gruppen-Messaging).

[Hinweis: Der Code der beiden Klassen ist relativ groß, daher sind sie gefaltet]

Beispiel 1: Eins-zu-eins-Nachrichtenversand und -abruf

Modulanforderungen:

1. Abfrage, wie viele Kontakte neue Nachrichten gesendet haben

2. Die Nachricht enthält den Absender, die Uhrzeit und den Inhalt der Nachricht

3. Die alte Nachricht kann abgerufen werden

4. Die Nachricht kann 7 Tage lang aufbewahrt werden , und der Ablauf wird passiv ausgelöst. Löschen

Redis-Implementierungsidee:

1 Listen bzw.

2 , Die Struktur der Originalnachricht wird in Form eines Arrays gespeichert und enthält den Absender, den Zeitstempel und den Informationsinhalt

3 Vor dem Einfügen in die Redis-verknüpfte Liste , die Daten müssen in den JSON-Typ konvertiert und dann gespeichert werden

4 Beim Abrufen neuer Informationen sollte rpoplpush verwendet werden, um die gelesene neue Nachricht in die alte Nachrichtenliste zu verschieben

5 Beim Abrufen alter Nachrichten sollten die Zeit und die Gegenwart der alten Nachricht verwendet werden. Vergleichen Sie die Zeit. Wenn das Zeitlimit überschritten wird, werden alle nachfolgenden Daten direkt gelöscht (da die Daten einzeln nach Zeit in die verknüpfte Liste verschoben werden). Zeit ist in der richtigen Reihenfolge angeordnet)

Datenspeicherstrukturdiagramm:

PHP-Implementierungscode:

#SinglePullMessage.class.php

<?php
#单接接收者接收消息
class SinglePullMessage
{
 private $redis=&#39;&#39;; #存储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, &#39;preMessage_&#39;.$user))
 {
  $temp=json_decode($json_message); #将json数据变成对象
  $messageArr[$temp->sender][]=$temp; #转换成数组信息
 }
 if($messageArr)
 {
  $arr[&#39;count&#39;]=count($messageArr); #统计有多少个用户发来信息
  $arr[&#39;messageArr&#39;]=$messageArr;
  return $arr;
 }
 return false;
 }
 public function getPreMessage($user)
 {
 ##取出旧消息
 $messageArr=array();
 $json_pre=$this->redis->lrange(&#39;preMessage_&#39;.$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(&#39;preMessage_&#39;.$user);
   break;
  }
  $this->redis->ltrim(&#39;preMessage_&#39;.$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 &#39;发送人：&#39;.$v2->sender.&#39; 发送时间：&#39;.date(&#39;Y-m-d h:i:s&#39;,$v2->time).&#39;<br/>&#39;;
  echo &#39;消息内容：&#39;.$v2->message.&#39;<br/>&#39;;
  }
  echo "<hr/>";
 }
 }
}

Nach dem Login kopieren

Test:

1. Nachricht senden

# Test1.php erstellen

include &#39;./SinglePullMessage.class.php&#39;;
$object=new SinglePullMessage(&#39;192.168.95.11&#39;);
#发送消息
$sender=&#39;boss&#39;; #发送者
$to=&#39;jane&#39;;  #接收者
$message=&#39;How are you&#39;; #信息
$time=time();
$arr=array(&#39;sender&#39;=>$sender,&#39;message&#39;=>$message,&#39;time&#39;=>$time);
echo $object->sendSingle($to,$arr);

Nach dem Login kopieren

2. Neue Nachrichten erhalten

# Test2.php erstellen

include &#39;./SinglePullMessage.class.php&#39;;
$object=new SinglePullMessage(&#39;192.168.95.11&#39;);
#获取新消息
$arr=$object->getNewMessage(&#39;jane&#39;);
if($arr)
{
 echo $arr[&#39;count&#39;]."个联系人发来新消息<br/><hr/>";
 $object->dealArr($arr[&#39;messageArr&#39;]); 
}
else
 echo "无新消息";

Nach dem Login kopieren

Zugriff auf Ergebnisse:

3. Alte Nachrichten abrufen

#Create test3.php

include &#39;./SinglePullMessage.class.php&#39;;
$object=new SinglePullMessage(&#39;192.168.95.11&#39;);
#获取旧消息
$arr=$object->getPreMessage(&#39;jane&#39;);
if($arr)
{
 $object->dealArr($arr);
}
else
 echo "无旧数据";

Nach dem Login kopieren

4. Many-to-many-Nachrichten

Beispiel 2: Many-to-many-Nachrichtenversand und -erfassung (d. h. Gruppe)

Modulvoraussetzungen:

1. Benutzer können selbst eine Gruppe erstellen und Gruppenleiter werden

2. Der Gruppenleiter kann Personen als Gruppenmitglieder hinzufügen und Personen ausschließen

3. Benutzer können die Gruppe direkt verlassen

4 Sie können Nachrichten senden und jedes Mitglied kann Nachrichten abrufen

5. Die maximale Nachrichtenkapazität der Gruppe beträgt 5.000 🎜>

6. Mitglieder können neue Nachrichten abrufen

7. Mitglieder können alte Nachrichten, die zuvor gelesen wurden, in Seiten abrufen

. . . . . Schreiben wir einfach diese Funktionen. Schüler, die Bedarf haben oder üben möchten, können weitere Funktionen hinzufügen, z. B. Stummschaltung, anonymes Senden von Nachrichten, Senden von Dateien usw.

Ideen für die Redis-Implementierung:

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=&#39;&#39;; #存储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=&#39;chatIdLock&#39;)
 {
 $identifier=$this->getLock($lockName); #获取锁
 if($identifier)
 {
  $id=$this->redis->incr(&#39;groupChatID&#39;); #获取群组ID
  $this->releaseLock($lockName,$identifier); #释放锁
 }
 else
  return false;
 $messageCount=$this->redis->set(&#39;countMessage_&#39;.$id, 0); #初始化这个群组消息计数器
 #开启非事务型流水线，一次性将所有redis命令传给redis，减少与redis的连接
 $pipe=$this->redis->pipeline(); 
 $this->redis->zadd(&#39;groupChat_&#39;.$id, 1, $user); #创建群组成员有序集合，并添加群主
 #将这个群组添加到user所参加的群组有序集合中
 $this->redis->zadd(&#39;hasGroupChat_&#39;.$user, 0, $id); 
 foreach ($addUser as $v) #创建群组的同时需要添加的用户成员
 {
  $this->redis->zadd(&#39;groupChat_&#39;.$id, 2, $v);
  $this->redis->zadd(&#39;hasGroupChat_&#39;.$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(&#39;groupChat_&#39;.$groupChatID, $user); #将groupChatName的群主取出来
 if($groupMasterScore==1) #判断user是否是群主
 {
  $pipe=$this->redis->pipeline(); #开启非事务流水线
  foreach ($addMembers as $v) 
  {
  $this->redis->zadd(&#39;groupChat_&#39;.$groupChatID, 2, $v);   #添加进群
  $this->redis->zadd(&#39;hasGroupChat_&#39;.$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(&#39;groupChat_&#39;.$groupChatID, $user); 
 if($groupMasterScore==1) #判断user是否是群主
 {
  $pipe=$this->redis->pipeline(); #开启非事务流水线
  foreach ($delMembers as $v) 
  {
  $this->redis->zrem(&#39;groupChat_&#39;.$groupChatID, $v);   
  $this->redis->zrem(&#39;hasGroupChat_&#39;.$v, $groupChatID); 
  }
  $pipe->exec();
  return true;
 }
 return false;
 }
 /**
 * @desc 退出群组
 *
 * @param $user string | 用户名
 * @param $groupChatID int | 群组名
 */
 public function quitGroupChat($user, $groupChatID)
 {
 $this->redis->zrem(&#39;groupChat_&#39;.$groupChatID, $user);
 $this->redis->zrem(&#39;hasGroupChat_&#39;.$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=&#39;countLock_&#39;)
 {
 $memberScore=$this->redis->zscore(&#39;groupChat_&#39;.$groupChatID, $user); #成员score
 if($memberScore)
 {
  $identifier=$this->getLock($preLockName.$groupChatID); #获取锁
  if($identifier) #判断获取锁是否成功
  {
  $messageCount=$this->redis->incr(&#39;countMessage_&#39;.$groupChatID);
  $this->releaseLock($preLockName.$groupChatID,$identifier); #释放锁
  }
  else
  return false;
  $json_message=json_encode($messageArr);
  $this->redis->zadd(&#39;groupChatMessage_&#39;.$groupChatID, $messageCount, $json_message);
  $count=$this->redis->zcard(&#39;groupChatMessage_&#39;.$groupChatID); #查看信息量大小
  if($count>5000) #判断数据量有没有达到5000条
  { #数据量超5000，则需要清除旧数据
  $start=5000-$count;
  $this->redis->zremrangebyrank(&#39;groupChatMessage_&#39;.$groupChatID, $start, $count);
  }
  return true;
 }
 return false;
 }
 /**
 * @desc 获取新信息
 *
 * @param $user string | 用户名
 *
 * @return 成功则放回json数据数组，无新信息返回false
 */
 public function getNewMessage($user)
 {
 $arrID=$this->redis->zrange(&#39;hasGroupChat_&#39;.$user, 0, -1, &#39;withscores&#39;); #获取用户拥有的群组ID
 $json_message=array(); #初始化
 foreach ($arrID as $k => $v) #遍历循环所有群组，查看是否有新消息
 {
  $messageCount=$this->redis->get(&#39;countMessage_&#39;.$k); #群组最大信息分值数
  if($messageCount>$v) #判断用户是否存在未读新消息
  {
  $json_message[$k][&#39;message&#39;]=$this->redis->zrangebyscore(&#39;groupChatMessage_&#39;.$k, $v+1, $messageCount);
  $json_message[$k][&#39;count&#39;]=count($json_message[$k][&#39;message&#39;]); #统计新消息数量
  $this->redis->zadd(&#39;hasGroupChat_&#39;.$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(&#39;groupChatMessage_&#39;.$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(&#39;nx&#39;, &#39;ex&#39;=>$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; #其他客户端修改了锁，不能删除别人的锁
 }
 }

}
?>

Nach dem Login kopieren

测试：

1、建立createGroupChat.php(测试创建群组功能)

执行代码并创建568、569群组（群主为jack）

include &#39;./ManyPullMessage.class.php&#39;;
$object=new ManyPullMessage(&#39;192.168.95.11&#39;);
#创建群组
$user=&#39;jack&#39;;
$arr=array(&#39;jane1&#39;,&#39;jane2&#39;);
$a=$object->createGroupChat($user,$arr);
echo "<pre class="brush:php;toolbar:false">";
print_r($a);
echo "

";die;

Nach dem Login kopieren

2、建立addMembers.php(测试添加成员功能)

执行代码并添加新成员

 include &#39;./ManyPullMessage.class.php&#39;;
 $object=new ManyPullMessage(&#39;192.168.95.11&#39;);
 $b=$object->addMembers(&#39;jack&#39;,&#39;568&#39;,array(&#39;jane1&#39;,&#39;jane2&#39;,&#39;jane3&#39;,&#39;jane4&#39;));
 echo "<pre class="brush:php;toolbar:false">";
 print_r($b);
 echo "

";die;

Nach dem Login kopieren

3、建立delete.php(测试群主删除成员功能)

include &#39;./ManyPullMessage.class.php&#39;;
$object=new ManyPullMessage(&#39;192.168.95.11&#39;);
#群主删除成员
$c=$object->delMembers(&#39;jack&#39;, &#39;568&#39;, array(&#39;jane1&#39;,&#39;jane4&#39;));
echo "<pre class="brush:php;toolbar:false">";
print_r($c);
echo "

";die;

Nach dem Login kopieren

4、建立sendMessage.php(测试发送消息功能)

多执行几遍，568、569都发几条

include &#39;./ManyPullMessage.class.php&#39;;
$object=new ManyPullMessage(&#39;192.168.95.11&#39;);
#发送消息
$user=&#39;jane2&#39;;
$message=&#39;go go go&#39;;
$groupChatID=568;
$arr=array(&#39;sender&#39;=>$user, &#39;message&#39;=>$message, &#39;time&#39;=>time());
$d=$object->sendMessage($user,$groupChatID,$arr);
echo "<pre class="brush:php;toolbar:false">";
print_r($d);
echo "

";die;

Nach dem Login kopieren

5、建立getNewMessage.php(测试用户获取新消息功能)

include &#39;./ManyPullMessage.class.php&#39;;
$object=new ManyPullMessage(&#39;192.168.95.11&#39;);
#用户获取新消息
$e=$object->getNewMessage(&#39;jane2&#39;);
echo "<pre class="brush:php;toolbar:false">";
print_r($e);
echo "

";die;

Nach dem Login kopieren

6、建立getPartMessage.php(测试用户获取某个群组部分消息)

（多发送几条消息，用于测试。568中共18条数据）

include &#39;./ManyPullMessage.class.php&#39;;
$object=new ManyPullMessage(&#39;192.168.95.11&#39;);
#用户获取某个群组部分消息
$f=$object->getPartMessage(&#39;jane2&#39;, 568, 1, 10); 
echo "<pre class="brush:php;toolbar:false">";
print_r($f);
echo "

";die;

Nach dem Login kopieren

page=1，size=10