您好,登录后才能下订单哦!
'connections' => [ .... 'database' => [ 'driver' => 'database', 'table' => 'jobs', 'queue' => 'default', 'expire' => 60, ], 'redis' => [ 'driver' => 'redis', 'connection' => 'default', 'queue' => 'default', 'expire' => 180, ], .... ],
Laravel5.2队列驱动config/queue.php配置文件,“database”和“redis”有个expire参数,手册上解释是“队列任务过期时间(秒)”,默认为60秒。
(注:5.2和之后的配置文件发生了变化,改为'retry_after' 参数,具体见手册)
网上搜了一下这个配置,没有太多说明,但是实际使用的过程中,发现对于执行时间超过expire设置时间的队列进程,还有使用队列进行分布式程序部署,这个参数和这种设计模式是个大坑。。。
发现这个问题是想使用分布式程序部署处理队列,两台服务器部署Laravel框架artisan脚本,连接一个MYSQL数据库,使用一张jobs队列表。
部署的后,分别启动两台服务器的脚本,发现后执行的脚本,在队列驱动中取数据,如MYSQL的jobs表,遇到先执行的脚本队列数据时不会跳过,而是把这条数据视为Failed,储存一条新数据到failed_jobs表(Laravel队列失败时会将队列数据储存到failed_jobs表),造成数据重复。
之前在一台服务器启动3个进程执行脚本,并不会发生这种错误,后执行的脚本不会取得前一个进程的队列数据,更不会判断成Failed,多服务处理时是什么原因造成队列驱动中的数据错误呢?
根据队列执行的流程,程序执行时,队列到队列驱动中取任务,获得任务的过程队列驱动应该做事物处理,这样第二个进程取任务会跳过正在执行的队列数据。
查了一些资料,了解了Laravel队列的原理,最后还得看Queue的源码。
Laravel的Queue的源码都在Illuminate\Queue目录下。
先分析以MYSQL为驱动的jobs表:
CREATE TABLE `jobs` ( `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT, `queue` varchar(255) COLLATE utf8_unicode_ci NOT NULL, `payload` longtext COLLATE utf8_unicode_ci NOT NULL, `attempts` tinyint(3) unsigned NOT NULL, `reserved` tinyint(3) unsigned NOT NULL, `reserved_at` int(10) unsigned DEFAULT NULL, `available_at` int(10) unsigned NOT NULL, `created_at` int(10) unsigned NOT NULL, PRIMARY KEY (`id`), KEY `jobs_queue_reserved_reserved_at_index` (`queue`,`reserved`,`reserved_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;
手册中主要介绍了队列任务的保存,payload字段储存的是序列化后的任务,Laravel队列可以将数据模型序列化,执行时候队列系统会自动从数据库中获取整个模型实例,具体说明见手册。
但是其他几个状态和时间字段才是保证队列事物处理的关键字段。
“attempts”执行次数,“reserved”执行的状态,“reserved_at”执行时间,‘available_at’预订执行时间,‘created_at’是队列创建时间。
监听事件的脚本有Listener.php和Worker.php两个脚本,看源码说明Listener可以处理指定的队列,connection参数,但是实际上最后都是通过work来处理队列的。Laravel5.4已经取消了queue:listen参数,都用queue:work来执行。不过我这里说的是Laravel5.2的问题,不知道是不是下面的原因使Laravel优化去掉了listen。
继续分析队列处理的Worker类的源码,取队列数据时用pop方法,这个方法会根据传递的驱动类型如database或redis,调用该驱动的pop方法。
$connection = $this->manager->connection($connectionName); $job = $this->getNextJob($connection, $queue); // If we're able to pull a job off of the stack, we will process it and // then immediately return back out. If there is no job on the queue // we will "sleep" the worker for the specified number of seconds. if (! is_null($job)) { return $this->process( $this->manager->getName($connectionName), $job, $maxTries, $delay ); }
下面是DatabaseQueue.php的pop方法。
/** * Pop the next job off of the queue. * * @param string $queue * @return \Illuminate\Contracts\Queue\Job|null */ public function pop($queue = null) { $queue = $this->getQueue($queue); $this->database->beginTransaction(); if ($job = $this->getNextAvailableJob($queue)) { $job = $this->markJobAsReserved($job); $this->database->commit(); return new DatabaseJob( $this->container, $this, $job, $queue ); } $this->database->commit(); }
取数据的过程事物处理已经打开。
取队列数据的核心还是$this->getNextAvailableJob($queue)。
打开sql日志,看看队列数据是如何查询出来的。
/** * Get the next available job for the queue. * * @param string|null $queue * @return \StdClass|null */ protected function getNextAvailableJob($queue) { $this->database->enableQueryLog(); $job = $this->database->table($this->table) ->lockForUpdate() ->where('queue', $this->getQueue($queue)) ->where(function ($query) { $this->isAvailable($query); $this->isReservedButExpired($query); }) ->orderBy('id', 'asc') ->first(); var_dump($this->database->getQueryLog()); return $job ? (object) $job : null; }
array(1) { [0] => array(3) { 'query' => string(165) "select * from `jobs` where `queue` = ? and ((`reserved` = ? and `available_at` <= ?) or (`reserved` = ? and `reserved_at` <= ?)) order by `id` asc limit 1 for update" 'bindings' => array(5) { [0] => string(7) "default" [1] => int(0) [2] => int(1493634233) [3] => int(1) [4] => int(1493634173) } 'time' => double(1.55) }
从sql语句中可以看出,取队列数据有两个条件
reserved为0时,available_at时间小于当前时间,这个条件是待执行的队列;reserved为1时,reserved_at执行开始时间小于计算出的时间($this->isReservedButExpired),即当前时间减去超时秒Carbon::now()->subSeconds($this->expire)->getTimestamp(),这个条件是判断队列任务是否过期。
整个select过程是 “for update”的,有排他锁。
取得符合条件的队列后
/** * Mark the given job ID as reserved. * * @param \stdClass $job * @return \stdClass */ protected function markJobAsReserved($job) { $job->reserved = 1; $job->attempts = $job->attempts + 1; $job->reserved_at = $this->getTime(); $this->database->table($this->table)->where('id', $job->id)->update([ 'reserved' => $job->reserved, 'reserved_at' => $job->reserved_at, 'attempts' => $job->attempts, ]); return $job; }
程序会更新该条数据,并且更新完后即commit。
同一服务器,第二个进程取数据时候遇到悲观锁,需要等第一个进程取数据更新reserved和时间后执行。也就是说Laravel队列使用database时,并发的进程并不是同时取多条数据,而是取同一条数据等待其中一个进程update数据状态和超时时间,队列取得数据成功后第一个操作就是更新,所以第二个进程不会取到第一进程的同样数据,除非是队列过期。
在DatabaseQueue.php的pop方法中,取得队列数据后,“$this->database->commit(); ”前 sleep(10),会很明显的看到第二队列没有获取其他队列数据,说明“for update”只是update级排他锁,不会排斥select。
Laravel使用database队列有时候会有阻塞现象,不知道是不是这个原因造成的。
如果执行时间过长,超过‘expire’参数设置时间,第二队列会取得第一个队列数据,判断超时,这时候就会根据设置的最大执行次数tries来判断是插入新队列数据继续尝试执行,还是插入到错误队列“failed_jobs”表判断队列执行失败。
以上就是Laravel使用mysql执行队列的逻辑,之前提到的两台服务器部署Laravel框架执行artisan脚本,一个jobs表队列Failed的问题就是服务器时间不一致的原因,后一台服务器执行时候将前一队列数据判断为超时而插入到“failed_jobs”一条新数据,已经达到最大失败次数的情况,否则还会插入新的数据继续尝试。
所以queue:listen的执行时间参数 --timeout=60,一定要设置小于队列任务过期时间expire参数!
还有,Laravel5.2的queue:work并没有--timeout=60这个参数。。。。。
最后是队列执行完的处理逻辑。
如果队列执行成功会删除jobs的数据,这没什么问题。如果失败,包括超时、异常等,会根据设置的最大失败次数判断是否插入一条新数据,或者插入一条Failed数据到“failed_jobs”表。
出现错误时,handleJobException的异常处理调用DatabaseQueue.php的release方法,$job->release($delay),最终是pushToDatabase实现。
插入新数据时候,attempts是失败次数,reserved为0,available_at为当前时间戳加上超时配置时间,
这样整个队列处理就形成了完整的数据操作。
Laravel5.4对队列功能进行了很大的修改,手册中的提示
任务执行时间
在配置文件 config/queue.php
中,每个连接都定义了 retry_after
项。该配置项的目的是定义任务在执行以后多少秒后释放回队列。如果retry_after
设定的值为 90
, 任务在运行 90
秒后还未完成,那么将被释放回队列而不是删除掉。毫无疑问,你需要把 retry_after
的值设定为任务执行时间的最大可能值。
Laravel5.4去掉了queue的listen命令,work也增加了超时参数。Laravel5.5出来的时候应该必须直接升级上去。
附录:Laravel5.2测试的脚本,之前网上搜出来的都比较早,还是把job写成命令的方式,其实5.2以后job使用非常简单。
jobs下定义job任务,handle可以增加一些测试方案,比如我这种抛出异常,直接Failed的
class MyJob extends Job implements ShouldQueue { use InteractsWithQueue, SerializesModels; private $key; private $value; /** * Create a new job instance. * * @return void */ public function __construct($key, $value) { $this->key = $key; $this->value = $value; } /** * Execute the job. * * @return void */ public function handle() { for($i=0;$i<20;$i++){ echo "$i\n"; sleep(1); } echo "sss\t".$this->key."\t".date("Y-m-d H:i:s")."\n"; throw new \Exception("测试\n"); // Redis::hset('queue.test', $this->key, $this->value); } public function failed() { dump('failed'); } }
控制器访问设置任务队列,key和value之前用了测试redis插入的,可以按自己的测试方案设置job参数。
for($i = 0; $i < 5; $i ++) { echo "$i"; $job = (new MyJob($i, $i))->delay(20); $this->dispatch($job); }
我的例子设置了5个队列,开启多个shell并发执行artisan测试吧。
本来想将redis队列代码读完,一起发出来的,最近事情太多,redis代码也没怎么看。
redis驱动可以参考 http://www.cnblogs.com/z1298703836/p/5346728.html 这篇文章对Laravel队列redis驱动逻辑介绍的很详细了,redis驱动使用的list和zset结构储存队列,执行过程会移除转存队列,没有数据库的“for update” 操作,所以应该不是存在队列阻塞的情况。
BUT队列任务过期时间设置和数据库驱动是一样的,所以同样
queue:listen的执行时间参数 --timeout=60,一定要设置小于队列任务过期时间expire参数!
终于写完了。。。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。