Redis消息通知系统的实现

转载:http://huoding.com/2012/02/29/146

最近忙着用Redis实现一个消息通知系统，今天大概总结了一下技术细节，其中演示代码若是没有特殊说明，使用的都是PhpRedis扩展来实现的。

内存

好比要推送一条全局消息，若是真的给全部用户都推送一遍的话，那么会占用很大的内存，实际上无论粘性有多高的产品，活跃用户同所有用户比起来，都会小不少，因此若是只处理登陆用户的话，那么至少在内存消耗上是至关划算的，至于未登陆用户，能够推迟到用户下次登陆时再处理，若是用户一直不登陆，就一了百了了。

队列

当大量用户同时登陆的时候，若是所有都即时处理，那么很容易就崩溃了，此时可使用一个队列来保存待处理的登陆用户，如此一来顶可能是反应慢点，但不会崩溃。

Redis的LIST数据类型能够很天然的建立一个队列，代码以下：

<?php

 

$redis = new Redis;

$redis->connect('/tmp/redis.sock');

 

$redis->lPush('usr', <USRID>);

 

while ($usr = $redis->rPop('usr')) {

    var_dump($usr);

}

 

?>

出于相似的缘由，咱们还须要一个队列来保存待处理的消息。固然也可使用LIST来实现，但LIST只能按照插入的前后顺序实现相似FIFO或LIFO形式的队列，然而消息其实是有优先级的：好比说我的消息优先级高，全局消息优先级低。此时可使用ZSET来实现，它里面分数的概念很天然的实现了优先级。不过ZSET没有原生的POP操做，因此咱们须要模拟实现，代码以下：

<?php

 

class RedisClient extends Redis

{

    const POSITION_FIRST = 0;

    const POSITION_LAST = -1;

 

    public function zPop($zset)

    {

        return $this->zsetPop($zset, self::POSITION_FIRST);

    }

 

    public function zRevPop($zset)

    {

        return $this->zsetPop($zset, self::POSITION_LAST);

    }

 

    private function zsetPop($zset, $position)

    {

        $this->watch($zset);

 

        $element = $this->zRange($zset, $position, $position);

 

        if (!isset($element[0])) {

            return false;

        }

 

        if ($this->multi()->zRem($zset, $element[0])->exec()) {

            return $element[0];

        }

 

        return $this->zsetPop($zset, $position);

    }

}

 

?>

模拟实现了POP操做后，咱们就可使用ZSET实现队列了，代码以下：

<?php

 

$redis = new RedisClient;

$redis->connect('/tmp/redis.sock');

 

$redis->zAdd('msg', <PRIORITY>, <MSGID>);

 

while ($msg = $redis->zRevPop('msg')) {

    var_dump($msg);

}

 

?>

推拉

之前微博架构中推拉选择的问题已经被你们讨论过不少次了。实际上消息通知系统和微博差很少，也存在推拉选择的问题，一样答案也是相似的，那就是应该推拉结合。具体点说：在登录用户获取消息的时候，就是一个拉消息的过程；在把消息发送给登录用户的时候，就是一个推消息的过程。

速度

假设要推送一百万条消息的话，那么最直白的实现就是不断的插入，代码以下：

<?php

 

for ($msgid = 1; $msgid <= 1000000; $msgid++) {

    $redis->sAdd('usr:<USRID>:msg', $msgid);

}

 

?>

说明：这里我使用了SET数据类型，固然你也能够视需求换成LIST或者ZSET。Redis的速度是很快的，可是借助PIPELINE，会更快，代码以下：

<?php

 

for ($i = 1; $i <= 100; $i++) {

    $redis->multi(Redis::PIPELINE);

    for ($j = 1; $j <= 10000; $j++) {

        $msgid = ($i - 1) * 10000 + $j;

        $redis->sAdd('usr:<USRID>:msg', $msgid);

    }

    $redis->exec();

}

 

?>

说明：所谓PIPELINE，就是省略了无谓的折返跑，把命令打包给服务端统一处理。先后两段代码在个人测试里，使用PIPELINE的速度大概是不使用PIPELINE的十倍。

查询

咱们用Redis命令行来演示一下用户是如何查询消息的。先插入三条消息，其分别是1，2，3：

先插入三条消息，其分别是1，2，3：

redis> HMSET msg:1 title title1 content content1

redis> HMSET msg:2 title title2 content content2

redis> HMSET msg:3 title title3 content content3

再把这三条消息发送给某个用户，其是123：

redis> SADD usr:123:msg 1

redis> SADD usr:123:msg 2

redis> SADD usr:123:msg 3

此时若是简单查询用户有哪些消息的话，无疑只能查到一些：

redis> SMEMBERS usr:123:msg

1) "1"

2) "2"

3) "3"

若是还须要用程序根据再来一次查询无疑有点低效，好在Redis内置的SORT命令能够达到事半功倍的效果，实际上它相似于SQL中的JOIN：

redis> SORT usr:123:msg GET msg:*->title

1) "title1"

2) "title2"

3) "title3"

redis> SORT usr:123:msg GET msg:*->content

1) "content1"

2) "content2"

3) "content3"

SORT的缺点是它只能GET出字符串类型的数据，若是你想要多个数据，就要屡次GET：

redis> SORT usr:123:msg GET msg:*->title GET msg:*->content

1) "title1"

2) "content1"

3) "title2"

4) "content2"

5) "title3"

6) "content3"

不少状况下这显得不够灵活，好在咱们能够采用其余一些方法平衡一下利弊，好比说新加一个字段，冗余保存完整消息的序列化，接着只GET这个字段就OK了。