laravel Queue——监听和消费，优雅的设计模式，知其然并知其所以然
设计模式 监听 优雅 laravel 消费 Queue
2023-09-27 14:25:41 时间
入口：
Illuminate\Queue\Console\WorkCommand类运行handle方法，监听失败处理，获取connection参数，获取queue，最后运行自身的runWorkder方法
public function handle()
{
if ($this->downForMaintenance() && $this->option('once')) {
return $this->worker->sleep($this->option('sleep'));
}
// We'll listen to the processed and failed events so we can write information
// to the console as jobs are processed, which will let the developer watch
// which jobs are coming through a queue and be informed on its progress.
$this->listenForEvents();
$connection = $this->argument('connection')
?: $this->laravel['config']['queue.default'];
// We need to get the right queue for the connection which is set in the queue
// configuration file for the application. We will pull it based on the set
// connection being run for the queue operation currently being executed.
$queue = $this->getQueue($connection);
$this->runWorker(
$connection, $queue
);
}runWorkder：
根据参数once选择是以守护进程daemon形式监听队列，还是只处理下一个job
/**
* Run the worker instance.
*
* @param string $connection
* @param string $queue
* @return array
*/
protected function runWorker($connection, $queue)
{
$this->worker->setCache($this->laravel['cache']->driver());
return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
$connection, $queue, $this->gatherWorkerOptions()
);
}
runNextJob很简单，就是简单得从队列里pop出一个job，通过runJob方法调用process处理即可，重点看daemon形式。
public function daemon($connectionName, $queue, WorkerOptions $options)
{
$this->listenForSignals();
$lastRestart = $this->getTimestampOfLastQueueRestart();
while (true) {
// Before reserving any jobs, we will make sure this queue is not paused and
// if it is we will just pause this worker for a given amount of time and
// make sure we do not need to kill this worker process off completely.
if (! $this->daemonShouldRun($options, $connectionName, $queue)) {
$this->pauseWorker($options, $lastRestart);
continue;
}
// First, we will attempt to get the next job off of the queue. We will also
// register the timeout handler and reset the alarm for this job so it is
// not stuck in a frozen state forever. Then, we can fire off this job.
$job = $this->getNextJob(
$this->manager->connection($connectionName), $queue
);
$this->registerTimeoutHandler($job, $options);
// If the daemon should run (not in maintenance mode, etc.), then we can run
// fire off this job for processing. Otherwise, we will need to sleep the
// worker so no more jobs are processed until they should be processed.
if ($job) {
$this->runJob($job, $connectionName, $options);
} else {
$this->sleep($options->sleep);
}
// Finally, we will check to see if we have exceeded our memory limits or if
// the queue should restart based on other indications. If so, we'll stop
// this worker and let whatever is "monitoring" it restart the process.
$this->stopIfNecessary($options, $lastRestart);
}
}首先需要看的是循环外边的 listenForSignals 方法，这里是通过unix底层的信号量机制，监听当前进程，分别注册接收3个信号，SIGTERM接收进程结束信号、GIGUSR2接收用户的暂停信号、SIGCONT接收用户恢复进程信号，三个信号分别标记shouldQuit和paused属性，控制进程的结束和暂停。
/**
* Enable async signals for the process.
*
* @return void
*/
protected function listenForSignals()
{
if ($this->supportsAsyncSignals()) {
pcntl_async_signals(true);
pcntl_signal(SIGTERM, function () {
$this->shouldQuit = true;
});
pcntl_signal(SIGUSR2, function () {
$this->paused = true;
});
pcntl_signal(SIGCONT, function () {
$this->paused = false;
});
}
}
while体中：
1、通过daemonShouldRun判断如果Application处于维护状态，则调用pauseWorker，然后，调用stopIfNecessary方法，通过shouldQuit标志判断是否需要重启新监听进程并kill掉当前监听进程；通过memoryExceeded()方法判断内存是否超出memory参数限制、上次重启的时间戳不相等（判断是否是被重启过），判断是否需要停止当前监听并重启新监听。
/**
* Pause the worker for the current loop.
*
* @param \Illuminate\Queue\WorkerOptions $options
* @param int $lastRestart
* @return void
*/
protected function pauseWorker(WorkerOptions $options, $lastRestart)
{
$this->sleep($options->sleep > 0 ? $options->sleep : 1);
$this->stopIfNecessary($options, $lastRestart);
}
/**
* Stop the process if necessary.
*
* @param \Illuminate\Queue\WorkerOptions $options
* @param int $lastRestart
*/
protected function stopIfNecessary(WorkerOptions $options, $lastRestart)
{
if ($this->shouldQuit) {
$this->kill();
}
if ($this->memoryExceeded($options->memory)) {
$this->stop(12);
} elseif ($this->queueShouldRestart($lastRestart)) {
$this->stop();
}
}
/**
* Determine if the memory limit has been exceeded.
*
* @param int $memoryLimit
* @return bool
*/
public function memoryExceeded($memoryLimit)
{
return (memory_get_usage(true) / 1024 / 1024) >= $memoryLimit;
}2、使用getNextJob方法获得下一个job
/**
* Get the next job from the queue connection.
*
* @param \Illuminate\Contracts\Queue\Queue $connection
* @param string $queue
* @return \Illuminate\Contracts\Queue\Job|null
*/
protected function getNextJob($connection, $queue)
{
try {
foreach (explode(',', $queue) as $queue) {
if (! is_null($job = $connection->pop($queue))) {
return $job;
}
}
} catch (Exception $e) {
$this->exceptions->report($e);
$this->stopWorkerIfLostConnection($e);
} catch (Throwable $e) {
$this->exceptions->report($e = new FatalThrowableError($e));
$this->stopWorkerIfLostConnection($e);
}
}3、laravel在处理每一个job之前，调用registerTimeoutHandler方法，通过timeout设置每个Job的最大处理时间，并注册监听SIGALRM信号，处理超时之后，结束相应进程，监听方法如下：
/**
* Register the worker timeout handler (PHP 7.1+).
*
* @param \Illuminate\Contracts\Queue\Job|null $job
* @param \Illuminate\Queue\WorkerOptions $options
* @return void
*/
protected function registerTimeoutHandler($job, WorkerOptions $options)
{
if ($this->supportsAsyncSignals()) {
// We will register a signal handler for the alarm signal so that we can kill this
// process if it is running too long because it has frozen. This uses the async
// signals supported in recent versions of PHP to accomplish it conveniently.
pcntl_signal(SIGALRM, function () {
$this->kill(1);
});
pcntl_alarm(
max($this->timeoutForJob($job, $options), 0)
);
}
}4、如果job存在，调用runJob处理任务否则根据sleep参数休眠等待
处理任务分为4步：调用before事件，判断最大重试次数、调用fire执行、调用after事件
/**
* Process the given job from the queue.
*
* @param string $connectionName
* @param \Illuminate\Contracts\Queue\Job $job
* @param \Illuminate\Queue\WorkerOptions $options
* @return void
*
* @throws \Throwable
*/
public function process($connectionName, $job, WorkerOptions $options)
{
try {
// First we will raise the before job event and determine if the job has already ran
// over its maximum attempt limits, which could primarily happen when this job is
// continually timing out and not actually throwing any exceptions from itself.
$this->raiseBeforeJobEvent($connectionName, $job);
$this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
$connectionName, $job, (int) $options->maxTries
);
// Here we will fire off the job and let it process. We will catch any exceptions so
// they can be reported to the developers logs, etc. Once the job is finished the
// proper events will be fired to let any listeners know this job has finished.
$job->fire();
$this->raiseAfterJobEvent($connectionName, $job);
} catch (Exception $e) {
$this->handleJobException($connectionName, $job, $options, $e);
} catch (Throwable $e) {
$this->handleJobException(
$connectionName, $job, $options, new FatalThrowableError($e)
);
}
}5、fire执行过程
调用Illuminate\Queue\Jobs\Job类中的fire方法，通过调用JobName::parse解析job，做生成job的反操作，然后调用响应的类的方法。具体生成和消费的参考另一篇文章
/**
* Fire the job.
*
* @return void
*/
public function fire()
{
$payload = $this->payload();
Log::info($payload['job']);
list($class, $method) = JobName::parse($payload['job']);
Log::info($class);
Log::info($method);
Log::info(json_encode($payload['data']));
($this->instance = $this->resolve($class))->{$method}($this, $payload['data']);
}Log代码是我临时加的，用来追踪
关于如何生产job不再赘述，网上文档一抓一大片，直接上我的生产者任务：
<?php
namespace App\Console\Commands;
use App\Jobs\ExampleJob;
use Illuminate\Console\Command;
use Illuminate\Queue\Jobs\JobName;
/**
* @brief content
* @author weihaoyu
* @createdAt date
* @updatedAt date
* @version 1.0
*/
class WhyTest extends Command
{
/**
* The name and signature of the console command.
*
* @var string
*/
protected $name = 'why:test';
/**
* The console command description.
*
* @var string
*/
protected $description = 'why测试用';
/**
* Create a new command instance.
*
* @return void
*/
public function __construct()
{
parent::__construct();
}
/**
* Execute the console command.
*
* @return mixed
*/
public function handle()
{
echo dispatch((new ExampleJob())->onQueue('why_test')->delay(10)); //输出job的id
try{
}catch (\Exception $e){
echo $e->getCode() . "\n";
echo $e->getMessage() . "\n";
}
}
}
<?php
namespace App\Jobs;
use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Support\Facades\Log;
class ExampleJob implements \Illuminate\Contracts\Queue\ShouldQueue
{
use InteractsWithQueue, Queueable, SerializesModels;
public $tries = 5;
/**
* Create a new job instance.
*
* @return void
*/
public function __construct()
{
//
}
/**
* Execute the job.
*
* @return void
*/
public function handle()
{
Log::info(333333);
}
}
查看log可以看到每个job存储的信息，
[2019-11-12 16:05:04] lumen.INFO: Illuminate\Queue\CallQueuedHandler@call
[2019-11-12 16:05:04] lumen.INFO: Illuminate\Queue\CallQueuedHandler
[2019-11-12 16:05:04] lumen.INFO: call
[2019-11-12 16:05:04] lumen.INFO: {"commandName":"App\\Jobs\\ExampleJob","command":"O:19:\"App\\Jobs\\ExampleJob\":8:{s:5:\"tries\";i:5;s:6:\"\u0000*\u0000job\";N;s:10:\"connection\";N;s:5:\"queue\";s:8:\"why_test\";s:15:\"chainConnection\";N;s:10:\"chainQueue\";N;s:5:\"delay\";i:10;s:7:\"chained\";a:0:{}}"}
[2019-11-12 16:05:04] lumen.INFO: 333333 同时，因为自己定义的Job类中use了InteractsWithQueue特性，同时InteractsWithQueue通过setJob方法注入了实例化Illuminate\Contracts\Queue\Job接口的类（依赖反转原则，参考另一篇设计模式相关文章，这种设计模式还方便扩展，不管用什么消息队列做驱动，只要相应的类implements接口，就可以重写此接口提供提供的任务相关操作方法），从而在自己的Job类中可以调用删除、重放等等一系列Job相关操作。
可以参考laravel的BeanstalkdJob类
<?php
namespace Illuminate\Queue\Jobs;
use Pheanstalk\Pheanstalk;
use Illuminate\Container\Container;
use Pheanstalk\Job as PheanstalkJob;
use Illuminate\Contracts\Queue\Job as JobContract;
class BeanstalkdJob extends Job implements JobContract
{
/**
* The Pheanstalk instance.
*
* @var \Pheanstalk\Pheanstalk
*/
protected $pheanstalk;
/**
* The Pheanstalk job instance.
*
* @var \Pheanstalk\Job
*/
protected $job;
/**
* Create a new job instance.
*
* @param \Illuminate\Container\Container $container
* @param \Pheanstalk\Pheanstalk $pheanstalk
* @param \Pheanstalk\Job $job
* @param string $connectionName
* @param string $queue
* @return void
*/
public function __construct(Container $container, Pheanstalk $pheanstalk, PheanstalkJob $job, $connectionName, $queue)
{
$this->job = $job;
$this->queue = $queue;
$this->container = $container;
$this->pheanstalk = $pheanstalk;
$this->connectionName = $connectionName;
}
/**
* Release the job back into the queue.
*
* @param int $delay
* @return void
*/
public function release($delay = 0)
{
parent::release($delay);
$priority = Pheanstalk::DEFAULT_PRIORITY;
$this->pheanstalk->release($this->job, $priority, $delay);
}
/**
* Bury the job in the queue.
*
* @return void
*/
public function bury()
{
parent::release();
$this->pheanstalk->bury($this->job);
}
/**
* Delete the job from the queue.
*
* @return void
*/
public function delete()
{
parent::delete();
$this->pheanstalk->delete($this->job);
}
/**
* Get the number of times the job has been attempted.
*
* @return int
*/
public function attempts()
{
$stats = $this->pheanstalk->statsJob($this->job);
return (int) $stats->reserves;
}
/**
* Get the job identifier.
*
* @return int
*/
public function getJobId()
{
return $this->job->getId();
}
/**
* Get the raw body string for the job.
*
* @return string
*/
public function getRawBody()
{
return $this->job->getData();
}
/**
* Get the underlying Pheanstalk instance.
*
* @return \Pheanstalk\Pheanstalk
*/
public function getPheanstalk()
{
return $this->pheanstalk;
}
/**
* Get the underlying Pheanstalk job.
*
* @return \Pheanstalk\Job
*/
public function getPheanstalkJob()
{
return $this->job;
}
}
6、循环最后一步通过stopIfNecessary方法检测是否需要kill掉
根据对消费的理解，自己用信号量实现了基于Redis的轻型消费者进程，失败记录，控制消费超时：https://blog.csdn.net/why444216978/article/details/102614361