ThinkPHP怎么使用think-queue實現redis消息隊列

本篇內容主要講解“ThinkPHP怎么使用think-queue實現redis消息隊列”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“ThinkPHP怎么使用think-queue實現redis消息隊列”吧!

簡單介紹：

消息隊列中間件是大型系統中的重要組件，已經逐漸成為企業系統內部通信的核心手段。它具有松耦合、異步消息、流量削峰、可靠投遞、廣播、流量控制、最終一致性等一系列功能，已經成為異步RPC的主要手段之一。

大白話：

消息隊列有兩個角色和一個容器，角色分別為生產者（負責發布任務）和消費者（負責執行任務），容器這是用來存放/堆積生產者發布的任務，將發布和執行兩個步驟分開且互不影響。

消息隊列的大致流程為：

生產者發布任務存放/堆積在消息隊列中，由消費者主動去消息隊列中取出任務并執行，先發布的先執行（隊列：先進先出），在沒有消費者的情況下任務會堆積在隊列中等待被取出執行。

優點：

消息隊列適用于大并發或者處理時間長并需要批量操作的第三方接口，可用于但不僅限于短信發送、郵件發送、APP推送等，支持跨系統，即本系統發布的消息隊列可以由自己或者給其他系統執行任務，同理本系統也可以作為消費者執行自己或者其他系統發布的消息隊列任務。

接下來主要介紹一下 think-queue 的使用

ThinkPHP的Queue內置了 Redis、Database、Topthink、Sync四種驅動，這里使用的是 Redis，也推薦使用 Redis

think-queue 隊列消息可以進行任務的發布、獲取、執行、刪除、重新發布、延遲發布、超時控制等操作

消息隊列基本配置

在 extra 目錄下創建 queue.php 配置文件

<?phpreturn [
    'connector'  => 'Redis',
    'expire'     => null,   // 任務過期時間，默認為60秒，若要禁用，則設置為 null
    'default'    => 'REDIS_QUEUE',  // 默認的隊列名
    'host'       => '127.0.0.1',   // redis 主機ip
    'port'       => 6379,   // redis 端口
    'password'   => '',   // redis 密碼
    'select'     => 0,   // 使用哪里一個 db，默認為 db0
    'timeout'    => 0,   // redis 連接的超時時間
    'persistent' => false,   // 是否是長連接];

至于為什么放在這里，是因為 Queue 源代碼默認從 extra 讀取 queue 文件獲取配置信息，如果想要將配置文件放置其他地方，則需要對應去修改源代碼中的默認獲取配置

生產者

創建一個測試類，寫入生產者方法

<?phpnamespace app\api\controller;use think\Controller;use think\Queue;class Test extends Controller{
    // 生產者，添加消息隊列
    public function addQueue()
    {
        // 參數
        $data = [
            'id' => rand(0, 99),
            'userName' => '一起摸魚'
        ];

        // 消息隊列名
        $queueName = 'testQueue';

        // 推入消息隊列，注意這里的 ::class 是PHP5.5才有的寫法
        $isPushed = Queue::push(TestQueue::class, $data, $queueName);
		// PHP5.5以下的可以直接寫命名空間
        // $isPushed = Queue::push('app\common\queue\TestQueue', $data, $queueName);
        
        if ($isPushed !== false) {
            // 成功之后的業務
            echo '隊列加入成功';
        } else {
            // 失敗之后的業務
            echo '隊列加入失敗';
        }
    }}

消費者

創建一個 TestQueue 類，用做消費者，執行消息隊列中的任務

<?phpnamespace app\common\queue;use think\Log;use think\queue\Job;class TestQueue{
    // 消費者執行入口
    public function fire(Job $job, $data)
    {
        // 具體執行業務
        $isJobDone = $this->doJob($data);
        
        if ($isJobDone) {
            // 消息隊列執行成功，刪除隊列，否則會一直執行
            $job->delete();
        } else {
            // 消息隊列執行失敗
            // 獲取消息隊列已經重試了幾遍
            $attempts = $job->attempts();
            if ($attempts == 0 || $attempts == 1) {
                // 重新發布，參數 delay 是延時發布的時間
                $job->release(2);
            }
        }
    }

    // 消息隊列執行失敗后會自動執行該方法
    public function failed($data)
    {
        Log::error('消息隊列達到最大重復執行次數后失?。?#39; . json_encode($data));
    }

    // 消息隊列執行方法
    public function doJob($data)
    {
        // 具體執行業務
        
        
        $data = json_encode($data);
        echo '消息隊列：' . $data;
        // 這里的判斷條件以具體業務是否執行成功進行判斷
        if ($data) {
            echo "執行成功";
            return true;
        } else {
            echo "執行失敗";
            return false;
        }
    }}

接下來就是啟用隊列的監聽模式了，因為不可能每次一有任務加進來就去手動執行一次隊列。隊列的監聽模式有兩種

項目根目錄執行

php think queue:work --queue 隊列名

但是由于需要，我們還要將消費者掛起守護進程執行，以確保關掉終端還能夠啟動隊列。

nohup php think queue:listen --queue 隊列名 &

PS：shell中輸入exit來退出終端
PS：shell中輸入exit來退出終端
PS：shell中輸入exit來退出終端

因為在nohup執行成功后直接點關閉程序按鈕關閉終端時會斷掉該命令所對應的session，導致 nohup 對應的進程被通知需要一起關掉。

到此，相信大家對“ThinkPHP怎么使用think-queue實現redis消息隊列”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！