消息隊列簡介

消息隊列：是消息的順序集合。
比如網站的PV統計和查看，傳統方式就是每個頁面發一個AJAX然后MySQL給PV+1。用戶量非常大的時候，沒有辦法實時插入PV。
結合redis消息隊列的實現，也是每個用戶訪問的時候發送ajax到控制器，這個時候redis每次rpush pvlog，相當于直接往數組后面插入一萬個行為，接下來用一個腳本運輸處理pvlog，set pv查看時候get pv，如果想處理用戶請求時間等等，同樣可以這樣異步處理。

常見場景和解決的問題：

應對流量峰值
異步消費（不定速的插入，生產和勻速的處理，消費）
解耦應用（不同來源的生產和同步去向的消費，基于publish/subscribe實現），即消息隊列作為消息池，同時往里面寫入的可能有多種數據，根據不同的場景來進行消費。
redis實現消息隊列原理
使用redis實現的最主要優勢是簡單快捷，性能沒有kafka高，但是安裝簡便，kafka性能高但是比較重，如果消息隊列不是很多，比如說一個博客計算pv，那么kafka可能比整個項目還要大。

實現方式：

方法一：基于list的實現方式

Screen-Shot-2019-04-06-at-7.01.54-PM.png

核心代碼
沒有用消息隊列的方式，使用incrBy大概上限在1000萬

<?php$redis = new Redis();$redis->connect('127.0.0.1', 6380);$res->select(0);$key = 'pv:index';//  看一下是不是沒有，如果沒有的話就設置成0if(false === $redis->get($key)){    $redis->set($key, 0);}// 如果有了就增加1 注意incrBy的上線大概在1000萬$redis->incrBy($key,1);`

用list來實現

// 用list消息隊列實現$key = 'listpv:index';$redis->rPush($key, 1);// 后臺的cron來實現// 先連接redis$redis = new Redis();$redis->connect('127.0.0.1', 6380);$res->select(0);$key = 'pv:index';// 用一個死循環while(true){    if(false !== $redis->lPop($key))    {        $redis->incrBy('pv:index');    }}// 然后在terminal中無線循環這個腳本

基于list來實現消息隊列的特點：

與水庫相似的地方：

水庫的容量決定承載能力 — redis的容量決定業務的承載能力
每一滴隨只可能經過一個閘門 — 每條消息只能被一個消費者消費
與水庫不同的地方：

水庫用于蓄水 — 一般要把消息全部消費掉
不要的隨扔掉 — 處理失敗的消息要做容錯

方法二：基于publish/subscribe

Screen-Shot-2019-04-06-at-7.53.22-PM.png

頻道固定，生產者和消費者不固定，可能一對多，也可能多對一，也可能多對多。
命令行的實現方式：

首先起兩個redis cli，一個作為訂閱，一個作為發布

訂閱者：SUBSCRIBE channel1 channel2

發布者:
PUBLISH channel1 helloChannel
PUBLISH channel2 helloChannel2

如果這個時候發布到一個沒有被訂閱的channel，那么這條消息就會丟失。
如果有多個訂閱了同一個channel，但有信息發布到同一個channel的時候，他們都會受到
代碼實現：

發布者：

<?php// 發布者$redis = new Redis();$redis->connect('127.0.0.1', 6380);$res = $redis->publish('c1','hello c1');echo "clients reading c1:{$res}n";$res = $redis->publish('c2','hello c2');echo "clients reading c2:{$res}n";$res = $redis->publish('c3','hello c3');echo "clients reading c3:{$res}n";

監聽者：

<?php$redis = new Redis();$redis->connect('127.0.0.1', 6380);// 超時控制$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);// 訂閱$redis->subscribe(['c1','c2'],function(Redis $instance, $channel, $message){    echo "received message form {$channel} : {$message}n";})?```php### 實戰部分-生成內容頁的質量分：實現三個功能：- 統計首頁、列表頁、內容頁的PV- 統計瀏覽時間超過5s的內容頁- 內容頁的PV+1分，瀏覽時間超過5s+5分，不超過5秒-1分，生成內容頁的質量分前端部分：?```JAVAscript<script>// ajax 訪問ajax.php，給內容頁增加PV$get('ajax.php?action=pv&from=article&aid=<?=$aid?>');// 如果頁面打開時間超過5秒，則發出統計setTimeout(function(){    $.get('ajax.php?action=get5&aid=<?=$aid?>');},5000);</script>

后端部分：
發布的實現：

<?php$action = $_GET['action'];$redis = new Redis();$redis->connect('127.0.0.1', 6380);// 首頁的PV$channelPvIndex = 'pv:index'; //內容頁的pv$channelPvList ='pv:list'; // 內容頁的PV$channelPvArticle = 'pv:article'; // 內容頁瀏覽超過5秒$channelGT5 = 'gt5:article';if('pv' === $action){    $from = $_GET['from'];    if('index' === $from)    {        $redis->publish($channelPVIndex, 1);    }    else if('list' === $from)    {        $tid = intval($_GET['tid']);        $redis->publish($channelPvList, $tid);    }    else if('articel' === $_GET['aid'])    {        $aid = intval($_GET['aid']);        $redis->publish($channelPvArticle, $aid);    }}else if('gt5' == $action){    $aid = intval($GET['aid'])    {        $redis->publish($channelGT5, $aid);    }}else{    // unknown action}

訂閱的實現：

<?php// 訂閱的實現$redis = new Redis();$redis->connect('127.0.0.1', 6380);// 首頁的PV$channelPvIndex = 'pv:index'; //內容頁的pv$channelPvList ='pv:list'; // 內容頁的PV$channelPvArticle = 'pv:article'; // 內容頁瀏覽超過5秒$channelGT5 = 'gt5:article';// 頻道和PV的key的映射$keyMap = [    $channelPVIndex => 'realtimepv:index',    $channelPvList => 'realtimepv:list',    $channelPvArticle => 'realtimepv:article'];$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);$redis->subscribe(    [$channelPVIndex,$channelPvList,$channelPvArticle,$channelGT5],    function(Redis $instance, $channel, $message){    // 注意在subscribe的回調中只能夠執行訂閱、取消訂閱、模式訂閱、模式取消訂閱，無法執行incrBy    // 嘗試取消訂閱命令(證明上面一句話)    // $instance->unsubscribe([$channelName]);    // 因此要想incryBy只能重新實例化一個redis    $redis2 = new Redis();    $redis2->connect('127.0.0.1',6380);    global $keyMap; //這里可以使用閉包實現    if(!isset($$keyMap[$channelName]))    {        $realTimePvKey = $keyMap[$channelName]; // 映射過來        $redis2->incrBy($realTimePvKey, 1);    }}    )

注意在subscribe的回調中只能夠執行訂閱、取消訂閱、模式訂閱、模式取消訂閱，無法執行incrBy，因此要想incryBy只能重新實例化一個redis

計算質量分：

<?php// 訂閱的實現$redis = new Redis();$redis->connect('127.0.0.1', 6380);// 內容頁的PV$channelPvArticle = 'pv:article'; // 內容頁瀏覽超過5秒$channelGT5 = 'gt5:article';$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);$redis->subscribe([$channelPVIndex,$channelPvList,$channelPvArticle,$channelGT5],    function(Redis $instance, $channel, $message){        /**         * 使用pv和gt5的數量計算文章的質量分         * 1. 如何計算？ = gt5*6         * 2. 以什么形式保存? HASH         * gt5: int         * score: int         */        $redis2 = new Redis();        $redis2->connect('127.0.0.1',6380);        global $keyMap; //這里可以使用閉包實現        if('gt5:article' === $channelName)        {            echo "${channelName}n";            $key = 'realtimescore:'.intval($message);            $res = $redis2->hIncrBy($keym 'gt5', 1);            echo "${channelName}n";            if($res)            {                $score = $res * 6;                $redis2->hSet($key, 'score', $score);                echo "{$score}n";            }            else            {                // 報警            }        }    })

本人文章均為原創，轉載請注明出處。