基於Redis消息隊列實現的消息推送

基於Redis消息隊列實現的消息推送

基本知識點

重點用到了如下命令實現咱們的消息推送

• brpop 阻塞模式 從隊列右邊獲取值以後刪除
• brpoplpush 從隊列A的右邊取值以後刪除，從左側放置到隊列B中

邏輯分析

• 在普通的任務腳本中寫入push_queue隊列要發送消息的目標，併爲目標設置一個要推送的內容，永不過時
• RedisPushQueue中brpoplpush處理，處理後的值放到temp_queue，主要防止程序崩潰形成推送失敗
• RedisAutoDeleteTempqueueItems處理temp_queue,這裏用到了brpop

代碼實現

• 普通任務腳本

  <?php
  foreach ($user_list as $item) {
      //命名規則 業務類型_操做_ID_隨機6位 值 自定義 我自定義的是"推送內容"
      $k_name = 'rabbit_push_' . $item['uid'].'_'.rand(100000,999999);
      $redis->lPush('push_queue',$k_name);//左進隊列
      $redis->set($k_name, '推送內容');
  }

• RedisPushQueue

  <?php
  //消息隊列處理推送~
  //
   // 守護進程運行
   // nohup php YOURPATH/RedisPushQueue.php & 開啓守護進程運行，修改文件以後須要重新啓動
  // blpop 有值則回去 沒值則阻塞 主要就是這個函數在起做用 不過並不安全，程序在執行過程當中崩潰就會致使隊列中的內容
   // 永久丟失~
   // BRPOPLPUSH 阻塞模式 右邊出 左邊進 在填寫隊列內容的時候要求從左進入
   //
  ini_set('default_socket_timeout', -1);  //不超時
  require_once 'YOURPARH/Rongcloud.php';
  
  $redis = new \Redis();
  $redis->connect('127.0.0.1', 6379);
  $redis->select(2);//切換到db2
  $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);
  
  // temp_queue臨時隊列防止程序崩潰致使隊列中內容丟失 0表明永不超時！
  While ($key = $redis->brpoplpush('push_queue', 'temp_queue', 0)) {
      if ($val = $redis->get($key)) {
          //rabbit_push_20_175990
          $arr = explode('_', $key);
          if (count($arr) != 4) {
              continue;
          }
          $id = $arr[2];
          push($id, $val);
          //刪除key內容
          $redis->del($key);
      }  
  }
  function push($id, $v)
  {
    //推送操做~
  }

• RedisAutoDeleteTempqueueItems

  <?php
  /* 自動處理temp_queue中的元素，這個操做是防止RedisPushQueue崩潰的時候作處理
   處理思路是 使用brpop 命令阻塞處理temp_queue這個隊列中的值，若是能獲取到"值"對應的"值"，說明RedisPushQueue執行失敗了
   將值還lpush到push_queue中，以備重新處理
   至於爲何使用brpop命令，是由於在RedisPushQueue中咱們使用的是brpoplpush
   nohup php YOURPATH/RedisAutoDeleteTempqueueItems.php & 開啓守護進程運行，修改文件以後須要重新啓動
  */
  ini_set('default_socket_timeout', -1);  //不超時
  $redis = new \Redis();
  $redis->connect('127.0.0.1', 6379);
  $redis->select(2);//切換到db2
  $redis->setOption(\Redis::OPT_READ_TIMEOUT, -1);
      while($key_arr = $redis->brPop('temp_queue',0)){
      if(count($key_arr) != 2){
          continue;
      }
      $key =$key_arr[1];
      if($redis->get($key)){//能獲取到值 說明RedisPushQueue執行失敗
          $redis->lPush('push_queue',$key);
      }
  }