一起聊聊thinkphp6使用think-queue實現普通隊列和延遲隊列

composer require topthink/think-queue

return [     'default'     => 'redis',     'connections' => [         'sync'     => [             'type' => 'sync',         ],         'database' => [             'type'       => 'database',             'queue'      => 'default',             'table'      => 'jobs',             'connection' => null,         ],         'redis'    => [             'type'       => 'redis',             'queue'      => 'default',             'host'       => env('redis.host', '127.0.0.1'),             'port'       => env('redis.port', '6379'),             'password'   => env('redis.password','123456'),             'select'     => 0,             'timeout'    => 0,             'persistent' => false,         ],     ],     'failed'      => [         'type'  => 'none',         'table' => 'failed_jobs',     ], ];

<?phpnamespace appqueue;use thinkfacadeCache;use thinkqueueJob;use thinkfacadeLog;/**  * Class Queue 隊列消費基礎類  * @package appqueue  */abstract class Queue{     /**      * @describe:fire是消息隊列默認調用的方法      * @param thinkqueueJob $job      * @param $message      */     public function fire(Job $job, $data)     {         if (empty($data)) {             Log::error(sprintf('[%s][%s] 隊列無消息', __CLASS__, __FUNCTION__));             return ;         }          $jobId = $job->getJobId(); // 隊列的數據庫id或者redis key         // $jobClassName = $job->getName(); // 隊列對象類         // $queueName = $job->getQueue(); // 隊列名稱          // 如果已經執行中或者執行完成就不再執行了         if (!$this->checkJob($jobId, $data)) {             $job->delete();             Cache::store('redis')->delete($jobId);             return ;         }          // 執行業務處理         if ($this->execute($data)) {             Log::record(sprintf('[%s][%s] 隊列執行成功', __CLASS__, __FUNCTION__));             $job->delete(); // 任務執行成功后刪除             Cache::store('redis')->delete($jobId); // 刪除redis中的緩存         } else {             // 檢查任務重試次數             if ($job->attempts() > 3) {                 Log::error(sprintf('[%s][%s] 隊列執行重試次數超過3次，執行失敗', __CLASS__, __FUNCTION__));                  // 第1種處理方式：重新發布任務,該任務延遲10秒后再執行；也可以不指定秒數立即執行                 //$job->release(10);                  // 第2種處理方式：原任務的基礎上1分鐘執行一次并增加嘗試次數                 //$job->failed();                    // 第3種處理方式：刪除任務                 $job->delete(); // 任務執行后刪除                 Cache::store('redis')->delete($jobId); // 刪除redis中的緩存             }         }     }      /**      * 消息在到達消費者時可能已經不需要執行了      * @param  string  $jobId      * @param $message      * @return bool 任務執行的結果      * @throws PsrSimpleCacheInvalidArgumentException      */     protected function checkJob(string $jobId, $message): bool     {         // 查詢redis         $data = Cache::store('redis')->get($jobId);         if (!empty($data)) {             return false;         }         Cache::store('redis')->set($jobId, $message);         return true;     }      /**      * @describe: 根據消息中的數據進行實際的業務處理      * @param $data 數據      * @return bool 返回結果      */     abstract protected function execute($data): bool;}

<?phpnamespace appqueuetest;use appqueueQueue;class Test extends Queue{     protected function execute($data): bool     {        // 具體消費業務邏輯     }}

use thinkfacadeQueue;  // 普通隊列生成調用方式 Queue::push($job, $data, $queueName); // 例： Queue::push(Test::class, $data, $queueName);  // 延時隊列生成調用方式 Queue::later($delay, $job, $data, $queueName); // 例如使用延時隊列 10 秒后執行： Queue::later(10 , Test::class, $data, $queueName);

php think queue:listen php think queue:work