【Laravel-海贼王系列】第十一章，Job&队列消费端实现
启动指令
php artisan queue:workphp
启动文件
namespace Illuminate\Queue\Console;
use Illuminate\Queue\Worker;
use Illuminate\Support\Carbon;
use Illuminate\Console\Command;
use Illuminate\Contracts\Queue\Job;
use Illuminate\Queue\WorkerOptions;
use Illuminate\Queue\Events\JobFailed;
use Illuminate\Queue\Events\JobProcessed;
use Illuminate\Queue\Events\JobProcessing;
class WorkCommand extends Command
{
...
/**
* @var \Illuminate\Queue\Worker
*/
protected $worker;
public function __construct(Worker $worker)
{
parent::__construct();
$this->worker = $worker;
}
public function handle()
{
if ($this->downForMaintenance() && $this->option('once')) {
return $this->worker->sleep($this->option('sleep'));
}
$this->listenForEvents();
$connection = $this->argument('connection')
?: $this->laravel['config']['queue.default'];
$queue = $this->getQueue($connection);
$this->runWorker(
$connection, $queue
);
}
...
}
复制代码
咱们先从构造函数和 handle() 方法开始分析，这是入口。html
片断一：判断是否维护模式或者
--force强制启动laravel
if ($this->downForMaintenance() && $this->option('once')) {
return $this->worker->sleep($this->option('sleep'));
}
复制代码
片断二：经过事件绑定在控制台输出信息redis
$this->listenForEvents();
protected function listenForEvents()
{
$this->laravel['events']->listen(JobProcessing::class, function ($event) {
$this->writeOutput($event->job, 'starting');
});
$this->laravel['events']->listen(JobProcessed::class, function ($event) {
$this->writeOutput($event->job, 'success');
});
$this->laravel['events']->listen(JobFailed::class, function ($event) {
$this->writeOutput($event->job, 'failed');
$this->logFailedJob($event);
});
}
复制代码
片断三：经过配置文件中配置的驱动获取对应驱动的队列名,若是没有则返回
defaultjson
$connection = $this->argument('connection') ?: $this->laravel['config']['queue.default'];
$queue = $this->getQueue($connection);
protected function getQueue($connection)
{
return $this->option('queue') ?: $this->laravel['config']->get(
"queue.connections.{$connection}.queue", 'default'
);
}
复制代码
片断四：传入链接驱动和队列名称到
runWorker方法运行任务。bash
$this->runWorker(
$connection, $queue
);
复制代码
这里是启动的重点，咱们传入的 $connection = 'redis' $queue = 'default'，继续分析app
protected function runWorker($connection, $queue)
{
// "这里的 $this->laravel['cache'] 是 Illuminate\Cache\CacheManager 类的实例。 （是在 app.providers.Illuminate\Cache\CacheServiceProvider::class 注册的） $this->laravel['cache']->driver() 返回 Illuminate\Cache\Repository 类的实例。"
// "框架经过 CacheManager 对不少存储管理进行了统一。 能够经过修改 app.config.cache.default 和 `app.config.cache.stores 中的值来修改存储驱动。"
// "将获取的驱动赋值给 workder 的 cache成员"
$this->worker->setCache($this->laravel['cache']->driver());
// "当 worker 对象拥有了cache对象以后便拥有了操做对应数据的能力 ！"
return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
$connection, $queue, $this->gatherWorkerOptions()
);
}
复制代码
继续运行框架
return $this->worker->{$this->option('once') ? 'runNextJob' : 'daemon'}(
$connection, $queue, $this->gatherWorkerOptions()
);
复制代码
这里传入的参数分别是，能够看出都是对队列消费的一些基本设置。异步
当运行模式非 --once 的状况下就会以 daemon 的方式运行。async
咱们看 \Illuminate\Queue\Worker 对象的 daemon 方法便可
守护进程模式
public function daemon($connectionName, $queue, WorkerOptions $options)
{
if ($this->supportsAsyncSignals()) {
$this->listenForSignals();
}
$lastRestart = $this->getTimestampOfLastQueueRestart();
while (true) {
if (! $this->daemonShouldRun($options, $connectionName, $queue)) {
$this->pauseWorker($options, $lastRestart);
continue;
}
$job = $this->getNextJob(
$this->manager->connection($connectionName), $queue
);
if ($this->supportsAsyncSignals()) {
$this->registerTimeoutHandler($job, $options);
}
if ($job) {
$this->runJob($job, $connectionName, $options);
} else {
$this->sleep($options->sleep);
}
$this->stopIfNecessary($options, $lastRestart, $job);
}
}
复制代码
进程参数设定
先设置进程的一些管理参数
if ($this->supportsAsyncSignals()) { // extension_loaded('pcntl'); 是否支持 'pcntl' 拓展，支持多进程的拓展。
$this->listenForSignals();
}
protected function listenForSignals()
{
// "PHP7.1信号新特性 -- 开启异步信号处理"
pcntl_async_signals(true);
// "安装信号处理器，后面能够传入相应的信号来终止或其余操做"
pcntl_signal(SIGTERM, function () {
// "SIGTERM 终止进程 软件终止信号"
$this->shouldQuit = true;
});
pcntl_signal(SIGUSR2, function () {
// "SIGUSR2 终止进程 用户定义信号2"
$this->paused = true;
});
pcntl_signal(SIGCONT, function () {
// "SIGCONT 忽略信号 继续执行一个中止的进程"
$this->paused = false;
});
}
复制代码
关于
pcntl的用法能够参考 PCNTL
信号能够参考对照表
接着看，从 cache 中获取上一次重启的时间戳
$lastRestart = $this->getTimestampOfLastQueueRestart();
复制代码
循环任务执行
判断是否终止运行
if (! $this->daemonShouldRun($options, $connectionName,$queue)) {
// "$opions 就是 调用artisan 传入的参数 $connectionName 我用了redis驱动，全部就是 'redis' $queue 这里没有传入队列则是 'default'"
$this->pauseWorker($options, $lastRestart);
continue;
}
复制代码
下面代码一共三个判断：
1.是不是关站模式而且非强制运行。
2.是否有外部传入的暂停信号
3.是否有绑定 Looping 事件执行并返回结果
若是符合条件则暂停或者发送终止信号。
主要功能是为了控制是否继续执行任务。
protected function daemonShouldRun(WorkerOptions $options, $connectionName, $queue)
{
return ! (($this->manager->isDownForMaintenance() && ! $options->force) ||
$this->paused ||
$this->events->until(new Events\Looping($connectionName, $queue)) === false);
}
复制代码
获取待运行的 Job
// "$this->manager->connection($connectionName) 是 Illuminate\Queue\RedisQueue 对象 $queue ： 'default'"
$job = $this->getNextJob(
$this->manager->connection($connectionName), $queue
);
复制代码
继续看 getNextJob
protected function getNextJob($connection, $queue)
{
try {
foreach (explode(',', $queue) as $queue) {
if (! is_null($job = $connection->pop($queue))) {
return $job;
}
}
} catch (Exception $e) {
// "异常处理主要是报告异常"
// "设置 '$this->shouldQuit = true;' 后续就会终止"
$this->exceptions->report($e);
$this->stopWorkerIfLostConnection($e);
$this->sleep(1);
} catch (Throwable $e) {
$this->exceptions->report($e = new FatalThrowableError($e));
$this->stopWorkerIfLostConnection($e);
$this->sleep(1);
}
}
复制代码
上面分析过了 $connection 是 RedisQueue 对象，全部展开 RedisQueue 的 pop 方法，获取要执行的任务对象。
public function pop($queue = null)
{
$this->migrate($prefixed = $this->getQueue($queue));
if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
return;
}
[$job, $reserved] = $nextJob;
if ($reserved) {
return new RedisJob(
$this->container, $this, $job,
$reserved, $this->connectionName, $queue ?: $this->default
);
}
}
复制代码
迁移延迟队列
在 pop 的过程当中首先迁移延迟队列的相关数据
protected function migrate($queue)
{
// "这里是否是很熟悉了，上一章存储端分析的时候延迟"
// "队列就是用的这个key来存的"
// "将延迟的队列迁移到主队列"
$this->migrateExpiredJobs($queue.':delayed', $queue);
// "将过时队列迁移到主队列"
if (! is_null($this->retryAfter)) {
$this->migrateExpiredJobs($queue.':reserved', $queue);
}
}
复制代码
继续看如何迁移到主队列的
public function migrateExpiredJobs($from, $to)
{
return $this->getConnection()->eval(
LuaScripts::migrateExpiredJobs(),
2,
$from,
$to,
$this->currentTime()
);
}
public static function migrateExpiredJobs()
{
return <<<'LUA'
if(next(val) ~= nil) then
redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)
for i = 1, #val, 100 do
redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
end
end
return val
LUA;
}
复制代码
最终经过 eval 命令使用 Lua 解释器执行脚本。 请看 Redis Eval
真香，这仅仅是把延迟任务切回主队列，继续！
检索数据
从队列检索下一个 Job
if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
return; // 没有数据就返回
}
复制代码
展开检索代码
protected function retrieveNextJob($queue)
{
// "默认值是 null"
if (! is_null($this->blockFor)) {
return $this->blockingPop($queue);
}
// "这段是直接经过 lua 从 redis lpop出对象,"
// "在lua中完成封装,执行逻辑和 blockingPop 类似"
return $this->getConnection()->eval(
LuaScripts::pop(), 2, $queue, $queue.':reserved',
$this->availableAt($this->retryAfter)
);
}
复制代码
咱们主要看 blockingPop 的代码
protected function blockingPop($queue)
{
// "以阻塞的方式弹出队列的第一个元素"
$rawBody = $this->getConnection()->blpop($queue, $this->blockFor);
// "解析获取的数据，同时再封装一个重试对象并写入有序集合。"
if (! empty($rawBody)) {
$payload = json_decode($rawBody[1], true);
$payload['attempts']++;
$reserved = json_encode($payload);
$this->getConnection()->zadd($queue.':reserved', [
$reserved => $this->availableAt($this->retryAfter),
]);
return [$rawBody[1], $reserved];
}
return [null, null];
}
复制代码
检索完成以后回到 pop 中继续执行
public function pop($queue = null)
{
$this->migrate($prefixed = $this->getQueue($queue));
if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
return;
}
// "到这里了！"
[$job, $reserved] = $nextJob;
if ($reserved) {
return new RedisJob(
$this->container, $this, $job,
$reserved, $this->connectionName, $queue ?: $this->default
);
}
}
复制代码
咱们来看看 $nextJob 是什么
最后调用
return new RedisJob(
$this->container, $this, $job,
$reserved, $this->connectionName, $queue ?: $this->default
);
复制代码
看看 Illuminate\Queue\Jobs\RedisJob 的构造函数
public function __construct(Container $container, RedisQueue $redis, $job, $reserved, $connectionName, $queue)
{
$this->job = $job;
$this->redis = $redis;
$this->queue = $queue;
$this->reserved = $reserved;
$this->container = $container;
$this->connectionName = $connectionName;
$this->decoded = $this->payload();
}
复制代码
这应该是最后一层封装,最后要返回给最外层的任务对象。
运行 Job
回到 Worker 对象中
...
$job = $this->getNextJob(
$this->manager->connection($connectionName), $queue
);
// "刚刚咱们从 redis 中拿到了封装好的 $job 对象，继续执行"
// "$job 就是 Illuminate\Queue\Jobs\RedisJob 对象"
// "是否支持 pcntl 拓展，异步模式传递信号"
if ($this->supportsAsyncSignals()) {
// "设置超时信号处理"
$this->registerTimeoutHandler($job, $options);
}
复制代码
继续注册超时信号控制
protected function registerTimeoutHandler($job, WorkerOptions $options)
{
pcntl_signal(SIGALRM, function () {
$this->kill(1);
});
pcntl_alarm(
max($this->timeoutForJob($job, $options), 0)
);
}
复制代码
总算要到运行 Job 的部分了
if ($job) {
$this->runJob($job, $connectionName, $options);
} else {
// "不存在 $job 则睡眠,最低睡眠1秒"
$this->sleep($options->sleep);
}
复制代码
解析 runJob
到这一步咱们已经拿到了全部的对象,接下来就是把 对象用起来！
protected function runJob($job, $connectionName, WorkerOptions $options)
{
try {
return $this->process($connectionName, $job, $options);
} catch (Exception $e) {
// "异常处理和上部分的同样，"
// "设定中止信号，在循环的结尾会检测信号"
// "所以咱们不须要分析这段"
$this->exceptions->report($e);
$this->stopWorkerIfLostConnection($e);
} catch (Throwable $e) {
$this->exceptions->report($e = new FatalThrowableError($e));
$this->stopWorkerIfLostConnection($e);
}
}
复制代码
展开
$this->process($connectionName, $job, $options);
复制代码
继续展开
public function process($connectionName, $job, WorkerOptions $options)
{
try {
// "触发任务执行前的绑定事件，从队列删除任务"
$this->raiseBeforeJobEvent($connectionName, $job);
// "标记超过最大重试次数的任务"
$this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
$connectionName, $job, (int) $options->maxTries
);
$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)
);
}
}
复制代码
$job->fire()
$job => Illuminate\Queue\Jobs\RedisJob 继承了 Illuminate\Queue\Jobs\Job 因此调用了抽象父类的 fire() 方法
public function fire()
{
$payload = $this->payload();
[$class, $method] = JobName::parse($payload['job']);
($this->instance = $this->resolve($class))->{$method}($this, $payload['data']);
}
复制代码
咱们看看 $payload 的结构实际就是 json_decode($job, true)
转换后的[$class, $method] 分别是 Illuminate\Queue\CallQueuedHandler 和 call
最后就是从容器中解析出 Illuminate\Queue\CallQueuedHandler 对象而且调用 call 方法，展开方法
public function call(Job $job, array $data)
{
try {
$command = $this->setJobInstanceIfNecessary(
$job, unserialize($data['command'])
);
} catch (ModelNotFoundException $e) {
return $this->handleModelNotFound($job, $e);
}
$this->dispatcher->dispatchNow(
$command, $this->resolveHandler($job, $command)
);
if (! $job->hasFailed() && ! $job->isReleased()) {
$this->ensureNextJobInChainIsDispatched($command);
}
if (! $job->isDeletedOrReleased()) {
$job->delete();
}
}
复制代码
先看看 $command 获取的是什么
protected function setJobInstanceIfNecessary(Job $job, $instance)
{
if (in_array(InteractsWithQueue::class, class_uses_recursive($instance))) {
$instance->setJob($job);
}
return $instance;
}
复制代码
打印 class_uses_recursive($instance)
接着就调用了 $instance->setJob($job);
这里的 $instance 就是对应咱们本身编写的任务对象。
执行完以后最终 $command 返回的就是本身编写的类
将 RedisJob 和 $command 传给 dispatchNow 方法 $this->dispatcher 是 Illuminate\Bus\Dispatcher 对象
$this->dispatcher->dispatchNow(
$command, $this->resolveHandler($job, $command)
);
复制代码
最后的真像
public function dispatchNow($command, $handler = null)
{
if ($handler || $handler = $this->getCommandHandler($command)) {
$callback = function ($command) use ($handler) {
// "划重点，要考!"
return $handler->handle($command);
};
} else {
$callback = function ($command) {
return $this->container->call([$command, 'handle']);
};
}
return $this->pipeline->send($command)->through($this->pipes)->then($callback);
}
复制代码
其实费了那么大的力气,最后就是调用 $command->handle 回头看看 job 的定义
就像烟火事后同样,消失于无形。
最后
总体分析下来感受使用 pcntl 拓展来作异步信号控制和进程中断来实现终止循环是一个亮点！
至此完成了任务队列消费端的分析,后续有机会分析 Horizon 是如何消费队列的哈~
相关文章
- 1. 【Laravel-海贼王系列】第十章，Job&队列存储端实现
- 2. 【Laravel-海贼王系列】第十六章，Builder 解析
- 3. 【Laravel-海贼王系列】第十二章，Facade 模式解析
- 4. 【Laravel-海贼王系列】第十五章，Model Observer 解析
- 5. 【Laravel-海贼王系列】第十四章，Session 解析
- 6. 【Laravel-海贼王系列】第一章，从index.php开始
- 7. 【Laravel-海贼王系列】第四章，Kernel 类解析
- 8. 【Laravel-海贼王系列】第五章，Request 类解析
- 9. 【Laravel-海贼王系列】第七章，Pipeline 类解析
- 10. 【Laravel-海贼王系列】第八章， Provider 功能解析
- 更多相关文章...
- • C# 队列（Queue） - C#教程
- • Scala List(列表) - Scala教程
- • Docker容器实战(一) - 封神Server端技术
- • ☆基于Java Instrument的Agent实现
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. vs2019运行opencv图片显示代码时，窗口乱码
- 2. app自动化 - 元素定位不到？别慌，看完你就能解决
- 3. 在Win8下用cisco ××× Client连接时报Reason 422错误的解决方法
- 4. eclipse快速补全代码
- 5. Eclipse中Java/Html/Css/Jsp/JavaScript等代码的格式化
- 6. idea+spring boot +mabitys(wanglezapin)+mysql (1)
- 7. 勒索病毒发生变种 新文件名将带有“.UIWIX”后缀
- 8. 【原创】Python 源文件编码解读
- 9. iOS9企业部署分发问题深入了解与解决
- 10. 安装pytorch报错CondaHTTPError：******
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 【Laravel-海贼王系列】第十章，Job&队列存储端实现
- 2. 【Laravel-海贼王系列】第十六章，Builder 解析
- 3. 【Laravel-海贼王系列】第十二章，Facade 模式解析
- 4. 【Laravel-海贼王系列】第十五章，Model Observer 解析
- 5. 【Laravel-海贼王系列】第十四章，Session 解析
- 6. 【Laravel-海贼王系列】第一章，从index.php开始
- 7. 【Laravel-海贼王系列】第四章，Kernel 类解析
- 8. 【Laravel-海贼王系列】第五章，Request 类解析
- 9. 【Laravel-海贼王系列】第七章，Pipeline 类解析
- 10. 【Laravel-海贼王系列】第八章， Provider 功能解析