轉：Redis消息隊列的若干實現方式

  微信號:neihanrukou

Redis消息隊列的若干實現方式

最近忙着用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);
}

?>

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命令行來演示一下用戶是如何查詢消息的。

先插入三條消息，其<MSGID>分別是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

再把這三條消息發送給某個用戶，其<USRID>是123：

redis> SADD usr:123:msg 1
redis> SADD usr:123:msg 2
redis> SADD usr:123:msg 3

此時若是簡單查詢用戶有哪些消息的話，無疑只能查到一些<MSGID>：

redis> SMEMBERS usr:123:msg
1) "1"
2) "2"
3) "3"

若是還須要用程序根據<MSGID>再來一次查詢無疑有點低效，好在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了。

實際暴露查詢接口的時候，不會使用PHP等程序來封裝，由於那會成倍下降RPS，推薦使用Webdis，它是一個Redis的Web代理，效率沒得說。

…

最近Tumblr發表了一篇相似的文章：Staircar: Redis-powered notifications，介紹了他們使用Redis實現消息通知系統的一些狀況，有興趣的不妨一塊兒看看。