PHP消息隊列實現及運用的方法是什么

這篇文章主要講解了“PHP消息隊列實現及運用的方法是什么”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“PHP消息隊列實現及運用的方法是什么”吧！

消息隊列的概念、原理、實現方式

概念

• 隊列結構的一個中間件

• 不需要立即消費消息

• 由消費者或者訂閱者進行按順序消費

基本的流程圖如下所示

• 流程
  

應用場景

• 冗余

• 解耦

• 流量削峰

• 異步通信

實現方式

• mysql：可靠、速度慢

• redis：速度快，對于大消息包處理較慢

• 消息系統：可靠、專業性強

消息的觸發機制

• 死循環的方式，故障時無法及時恢復

• 定時任務：壓力均分、但是處理量有上限

• 守護進程的方式

解耦 （訂單和配送系統）

• 架構設計1 采用定時任務的方式
  

  php入門到就業線上直播課：進入學習
  Apipost = Postman + Swagger + Mock + Jmeter 超好用的API調試工具：點擊使用

• 使用配送處理系統進行處理時，將當前數據庫里需要處理的訂單狀態更新為2，待處理完成后將狀態設為1

• 可以每次指定更新多少條數據

流量削鋒 （redis實現秒殺）

• 使用隊列的數據結構

• lpush/rpush 將數據放入列表中

• lpop/rpop 將數據移除列表并獲取到移除的值

• ltrim 保留指定區間內的元素

• llen 獲取列表長度

• lset 通過索引設置列表的值

• lindex 通過索引獲取列表中的值

• lrange 獲取指定范圍的元素

• 圖示如下
  

• 代碼流程如下

• 秒殺程序將請求寫入redis(uid,time)

• 檢查redis列表存放的長度，超過10個直接舍棄

• 通過死循環讀取redis數據，并存入數據庫

  
  

  // Spike.php 秒殺程序if(Redis::llen('lottery') < 10){
      // 成功
      Redis::lpush('lottery', $uid.'%'.microtime());}else{
      // 失敗}

• // Warehousing.php 入庫程序while(true){
      $user = Redis::rpop('lottery');
      if (!$user || $user == 'nil') {
          sleep(2);
          continue;
      }
      $user_arr = explode($user, '%');
      $insert_user = [
          'uid' => $user_arr[0],
          'time' => $user_arr[1]
      ];
      $res = DB::table('lottery_queue')->insert($insert_user);
      if (!$res) {
          Redis::lpush('lottery', $user);
      }}
  

• 上述代碼中假如并發過大的話會存在超賣的情況，此時可以使用文件鎖或者redis分布式鎖進行控制，先將商品放入redis list中 使用rpop進行取出，如果取不到則說明已經賣完

• 具體的思路及偽代碼如下

  
  

    // 先將商品放入redis中
    $goods_id = 2;
  
    $sql = select id,num from goods where id = $goods_id;
    $res = DB::select($sql);
    if (!empty($res)) {
        // 也可以指定多少件
        Redis::del('lottery_goods' . $goods_id);
        for($i=0;$i<$res['num'];$i++){
            Redis::lpush('lottery_goods . $goods_id', $i);
        }
        LOG::info('商品存入隊列成功，數量：' . Redis::llen('lottery_goods . $goods_id'));
    } else {
        LOG::info($goods_id . '加入失敗');
    }

•   // 開始秒殺
    $count = Redis::rpop('lottery_goods' . $goods_id);
    if (!$count) {
        // 商品已搶完
        ...
    }
  
    // 用戶搶購隊列
    $user_list = 'user_goods_id_' . $goods_id;
    $user_status = Redis::sismember($user_list, $user_id);
    if ($user_status) {
        // 已搶過
        ...
    }
  
    // 將搶到的放到列表中
    Redis::sadd($user_list, $uid);
    $msg = '用戶：' . $uid . '順序' . $count;
    Log::info($msg);
    // 生成訂單等
    ...
    // 減庫存
    $sql = update goods set num = num -1 where id = $goods_id and num > 0; // 防止超賣
    DB::update($sql)
    // 搶購成功
  

rabbitmq

• 架構及原理
  
  其中P代表生產者，X為交換機(channal)，C代表消費者

• 簡單使用

  
  

    // Send.php
    require_once __DIR__.'/vendor/autoload.php';
  
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
  
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
  
    // 創建通道
    $channel = $connection->channel();
    // 聲明一個隊列
    $channel->queue_declare('user_email', false, false, false, false);
    // 制作消息
    $msg = new AMQPMessage('send email');
    // 將消息推送到隊列
    $channel->basic_publish($msg, '', 'user_email');
  
    echo '[x] send email';
  
    $channel->close();
    $connection->close();

•   // Receive.php
    require_once __DIR__.'/vendor/autoload.php';
  
    use PhpAmqpLib\Connection\AMQPStreamConnection;
    use PhpAmqpLib\Message\AMQPMessage;
  
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
  
    //創建通道
    $channel = $connection->channel();
  
    $channel->queue_declare('user_email', false, false, false, false);
  
    // 當收到消息時的回調函數
    $callback = function($msg){
        //發送郵件
        echo 'Received '.$msg->body.'\n';
    };
  
    $channel->basic_consume('user_email', '', false, true, false, false, $callback);
  
    // 保持監聽狀態
    while($channel->is_open()){
        $channel->wait();
    }
  

感謝各位的閱讀，以上就是“PHP消息隊列實現及運用的方法是什么”的內容了，經過本文的學習后，相信大家對PHP消息隊列實現及運用的方法是什么這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！