Laravel 隊列使用

觸發

任務的觸發,主要的實現是在IlluminateFoundationBusDispatchesJobs這個trait中實現的,其只包含兩個方法

protected function dispatch($job){
    return app(Dispatcher::class)->dispatch($job);
}

protected function dispatchNow($job){
    return app(Dispatcher::class)->dispatchNow($job);
}

這兩個方法的區別就在於時間上,一個是推送到隊列,另外一個是推送並立即執行,依據前面的traitQueueable可知,我們可以在對任務進行設置,例如

protected function hpJob(){
    $job = (new HappybirdJob(5))->onConnection("other")
        ->onQueue("happybird")
        ->delay(5);
}

以上例子表示,我們使用other的連接在happybird隊列中設置一個延遲時間為5s的HappybirdJob任務

由於dispatch, dispatchNow都是使用的是容器中綁定的Dsipatch類,接下來看一下這個類 這個類位於IlluminateBusDispatcher中

Dispatch

該類主要是用於隊列任務的分發以及設置,其中可以看到

    public function dispatch($command)
    {
        if ($this->queueResolver && $this->commandShouldBeQueued($command)) {
            return $this->dispatchToQueue($command);
        } else {
            return $this->dispatchNow($command);
        }
    }

當使用dispatch類觸發任務時,將需要判斷是否實現了ShouldQueue,這個就是之前在討論異步隊列和同步隊列時兩個不同的列所實現的差異

而對於立即執行的操作來說,則是通過管道來執行

    public function dispatchNow($command)
    {
        return $this->pipeline->send($command)
            ->through($this->pipes)
            ->then(function ($command) {
                return $this->container->call([$command, 'handle']);
        });
    }

對於管道的說明和解釋,可以參考 Laravel Pipeline 組件的實現, Understanding Laravel Pipelines這兩篇文章

執行

對於推送到隊列里面的任務,可以通過artisan命令及其參數來進行控制,通過如下命令查看詳情

$ php artisan list | grep queue

將可以得到如下的結果:

  queue:failed        List all of the failed queue jobs
  queue:failed-table  Create a migration for the failed queue jobs database table
  queue:flush         Flush all of the failed queue jobs
  queue:forget        Delete a failed queue job
  queue:listen        Listen to a given queue
  queue:restart       Restart queue worker daemons after their current job
  queue:retry         Retry a failed queue job
  queue:table  大專欄  Laravel 隊列使用        Create a migration for the queue jobs database table
  queue:work          Process the next job on a queue

這里需要特別注意的是,queue:work, queue:listen這兩個命令,其余的都沒有額外的參數

queue:listern

--queue[=QUEUE]      The queue to listen on
--delay[=DELAY]      Amount of time to delay failed jobs [default: 0]
--memory[=MEMORY]    The memory limit in megabytes [default: 128]
--timeout[=TIMEOUT]  Seconds a job may run before timing out [default: 60]
--sleep[=SLEEP]      Seconds to wait before checking queue for jobs [default: 3]
--tries[=TRIES]      Number of times to attempt a job before logging it failed [default: 0]

queue:work

--queue[=QUEUE]    The queue to listen on
--daemon           Run the worker in daemon mode
--delay[=DELAY]    Amount of time to delay failed jobs [default: 0]
--force            Force the worker to run even in maintenance mode
--memory[=MEMORY]  The memory limit in megabytes [default: 128]
--sleep[=SLEEP]    Number of seconds to sleep when no job is available [default: 3]
--tries[=TRIES]    Number of times to attempt a job before logging it failed [default: 0]

相同:

• queue: 當包含多個隊列時,指定其執行的優先級,優先級順序為:越往前優先級越高
• delay: 失敗隊列的延遲執行時間
• memory: 任務執行的最大使用內存,默認為128M
• tries: 任務失敗后,嘗試運行的次數,默認為不啟動

差異: 對於listen來說,其包含一個--timeout參數,用來置頂執行的時間,超過該時間,將會造成任務失敗,程序異常,隊列停止

而--sleep選項對於listen來說,表示暫時未有任務執行時,重新執行任務需要的時間,對於queue:work來說,則表示當沒有任務運行時,重新檢測任務的時間間隔

queue:work還有一項特別的參數就是用來指定work運行的模式,如果啟用,則會強制隊列服務器持續處理任務,無需重啟框架

守護

官方推薦使用Supervisor來守護隊列的進程,其推薦的配置如下:

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /home/forge/app.com/artisan queue:work sqs --sleep=3 --tries=3 --daemon
autostart=true
autorestart=true
user=forge
numprocs=8
redirect_stderr=true
stdout_logfile=/home/forge/app.com/worker.log

• command:用來設置運行的命令,這個需要替換成自己的環境配置
• numprocs: 監控的進程數量,需要注意的是,進程的數量不能設置過大