一文带你了解Laravel schedule调度的运行机制
时间:2022-03-02 19:55
本篇文章带大家聊聊Laravel 中 schedule 调度的运行机制,希望对大家有所帮助!
Laravel 的 console 命令行极大的方便了 PHP 定时任务的设置以及运行。以往通过 crontab
配置定时任务过程相对比较繁琐,并且通过 crontab
设置的定时任务很难防止任务的交叠运行。
所谓任务的交叠运行,是指由于定时任务运行时间较长,在 crontab 设置的运行周期不尽合理的情况下,已经启动的任务还没有结束运行,而系统又启动了新的任务去执行相同的操作。如果程序内部没有处理好数据一致性的问题,那么两个任务同时操作同一份数据,很可能会导致严重的后果。
⒈ runInBackground
和 withoutOverlapping
为了防止任务的交叠运行,Laravel 提供了 withoutOverlapping()
方法;为了能让多任务在后台并行执行,Laravel 提供了 runInBackground()
方法。
⑴ runInBackground()
方法
console 命令行中的每一个命令都代表一个 Event
,\App\Console\Kernel
中的 schedule()
方法的作用只是将这些命令行代表的 Event
注册到 Illuminate\Console\Scheduling\Schedule
的属性 $events
中。
// namespace \Illuminate\Console\Scheduling\Schedule public function command($command, array $parameters = []) { if (class_exists($command)) { $command = Container::getInstance()->make($command)->getName(); } return $this->exec( Application::formatCommandString($command), $parameters ); } public function exec($command, array $parameters = []) { if (count($parameters)) { $command .= ' '.$this->compileParameters($parameters); } $this->events[] = $event = new Event($this->eventMutex, $command, $this->timezone); return $event; }
Event
的运行方式有两种:Foreground
和 Background
。二者的区别就在于多个 Event
是否可以并行执行。Event
默认以 Foreground
的方式运行,在这种运行方式下,多个 Event
顺序执行,后面的 Event
需要等到前面的 Event
运行完成之后才能开始执行。
但在实际应用中,我们往往是希望多个 Event
可以并行执行,此时就需要调用 Event
的 runInBackground()
方法将其运行方式设置为 Background
。
Laravel 框架对这两种运行方式的处理区别在于命令行的组装方式和回调方法的调用方式。
// namespace \Illuminate\Console\Scheduling\Event protected function runCommandInForeground(Container $container) { $this->callBeforeCallbacks($container); $this->exitCode = Process::fromShellCommandline($this->buildCommand(), base_path(), null, null, null)->run(); $this->callAfterCallbacks($container); } protected function runCommandInBackground(Container $container) { $this->callBeforeCallbacks($container); Process::fromShellCommandline($this->buildCommand(), base_path(), null, null, null)->run(); } public function buildCommand() { return (new CommandBuilder)->buildCommand($this); } // namespace Illuminate\Console\Scheduling\CommandBuilder public function buildCommand(Event $event) { if ($event->runInBackground) { return $this->buildBackgroundCommand($event); } return $this->buildForegroundCommand($event); } protected function buildForegroundCommand(Event $event) { $output = ProcessUtils::escapeArgument($event->output); return $this->ensureCorrectUser( $event, $event->command.($event->shouldAppendOutput ? ' >> ' : ' > ').$output.' 2>&1' ); } protected function buildBackgroundCommand(Event $event) { $output = ProcessUtils::escapeArgument($event->output); $redirect = $event->shouldAppendOutput ? ' >> ' : ' > '; $finished = Application::formatCommandString('schedule:finish').' "'.$event->mutexName().'"'; if (windows_os()) { return 'start /b cmd /c "('.$event->command.' & '.$finished.' "%errorlevel%")'.$redirect.$output.' 2>&1"'; } return $this->ensureCorrectUser($event, '('.$event->command.$redirect.$output.' 2>&1 ; '.$finished.' "$?") > ' .ProcessUtils::escapeArgument($event->getDefaultOutput()).' 2>&1 &' ); }
从代码中可以看出,采用 Background
方式运行的 Event
,其命令行在组装的时候结尾会增加一个 &
符号,其作用是使命令行程序进入后台运行;另外,采用 Foreground
方式运行的 Event
,其回调方法是同步调用的,而采用 Background
方式运行的 Event
,其 after
回调则是通过 schedule:finish
命令行来执行的。
⑵ withoutOverlapping()
方法
在设置 Event
的运行周期时,由于应用场景的不断变化,很难避免某个特定的 Event
在某个时间段内需要运行较长的时间才能完成,甚至在下一个运行周期开始时还没有执行完成。如果不对这种情况进行处理,就会导致多个相同的 Event
同时运行,而如果这些 Event
当中涉及到对数据的操作并且程序中没有处理好幂等问题,很可能会造成严重后果。
为了避免出现上述的问题,Event
中提供了 withoutOverlapping()
方法,该方法通过将 Event
的 withoutOverlapping
属性设置为 TRUE
,在每次要执行 Event
时会检查当前是否存在正在执行的相同的 Event
,如果存在,则不执行新的 Event
任务。
// namespace Illuminate\Console\Scheduling\Event public function withoutOverlapping($expiresAt = 1440) { $this->withoutOverlapping = true; $this->expiresAt = $expiresAt; return $this->then(function () { $this->mutex->forget($this); })->skip(function () { return $this->mutex->exists($this); }); } public function run(Container $container) { if ($this->withoutOverlapping && ! $this->mutex->create($this)) { return; } $this->runInBackground ? $this->runCommandInBackground($container) : $this->runCommandInForeground($container); }
⒉ mutex
互斥锁
在调用 withoutOverlapping()
方法时,该方法还实现了另外两个功能:一个是设置超时时间,默认为 24 小时;另一个是设置 Event
的回调。
⑴ 超时时间
首先说超时时间,这个超时时间并不是 Event
的超时时间,而是 Event
的属性 mutex
的超时时间。在向 Illuminate\Console\Scheduling\Schedule
的属性 $events
中注册 Event
时,会调用 Schedule
中的 exec()
方法,在该方法中会新建 Event
对象,此时会向 Event
的构造方法中传入一个 eventMutex
,这就是 Event
对象中的属性 mutex
,超时时间就是为这个 mutex
设置的。而 Schedule
中的 eventMutex
则是通过实例化 CacheEventMutex
来创建的。
// namespace \Illuminate\Console\Scheduling\Schedule $this->eventMutex = $container->bound(EventMutex::class) ? $container->make(EventMutex::class) : $container->make(CacheEventMutex::class);
设置了 withoutOverlapping
的 Event
在执行之前,首先会尝试获取 mutex
互斥锁,如果无法成功获取到锁,那么 Event
就不会执行。获取互斥锁的操作通过调用 mutex
的 create()
方法完成。
CacheEventMutex
在实例化时需要传入一个 \Illuminate\Contracts\Cache\Factory
类型的实例,其最终传入的是一个 \Illuminate\Cache\CacheManager
实例。在调用 create()
方法获取互斥锁时,还需要通过调用 store()
方法设置存储引擎。
// namespace \Illuminate\Foundation\Console\Kernel protected function defineConsoleSchedule() { $this->app->singleton(Schedule::class, function ($app) { return tap(new Schedule($this->scheduleTimezone()), function ($schedule) { $this->schedule($schedule->useCache($this->scheduleCache())); }); }); } protected function scheduleCache() { return Env::get('SCHEDULE_CACHE_DRIVER'); } // namespace \Illuminate\Console\Scheduling\Schedule public function useCache($store) { if ($this->eventMutex instanceof CacheEventMutex) { $this->eventMutex->useStore($store); } /* ... ... */ return $this; } // namespace \Illuminate\Console\Scheduling\CacheEventMutex public function create(Event $event) { return $this->cache->store($this->store)->add( $event->mutexName(), true, $event->expiresAt * 60 ); } // namespace \Illuminate\Cache\CacheManager public function store($name = null) { $name = $name ?: $this->getDefaultDriver(); return $this->stores[$name] = $this->get($name); } public function getDefaultDriver() { return $this->app['config']['cache.default']; } protected function get($name) { return $this->stores[$name] ?? $this->resolve($name); } protected function resolve($name) { $config = $this->getConfig($name); if (is_null($config)) { throw new InvalidArgumentException("Cache store [{$name}] is not defined."); } if (isset($this->customCreators[$config['driver']])) { return $this->callCustomCreator($config); } else { $driverMethod = 'create'.ucfirst($config['driver']).'Driver'; if (method_exists($this, $driverMethod)) { return $this->{$driverMethod}($config); } else { throw new InvalidArgumentException("Driver [{$config['driver']}] is not supported."); } } } protected function getConfig($name) { return $this->app['config']["cache.stores.{$name}"]; } protected function createFileDriver(array $config) { return $this->repository(new FileStore($this->app['files'], $config['path'], $config['permission'] ?? null)); }
在初始化 Schedule
时会指定 eventMutex
的存储引擎,默认为环境变量中的配置项 SCHEDULE_CACHE_DRIVER
的值。但通常这一项配置在环境变量中并不存在,所以 useCache()
的参数值为空,进而 eventMutex
的 store
属性值也为空。这样,在 eventMutex
的 create()
方法中调用 store()
方法为其设置存储引擎时,store()
方法的参数值也为空。
当 store()
方法的传参为空时,会使用应用的默认存储引擎(如果不做任何修改,默认 cache
的存储引擎为 file
)。之后会取得默认存储引擎的配置信息(引擎、存储路径、连接信息等),然后实例化存储引擎。最终,file
存储引擎实例化的是 \Illuminate\Cache\FileStore
。
在设置完存储引擎之后,紧接着会调用 add()
方法获取互斥锁。由于 store()
方法返回的是 \Illuminate\Contracts\Cache\Repository
类型的实例,所以最终调用的是 Illuminate\Cache\Repository
中的 add()
方法。
// namespace \Illuminate\Cache\Repository public function add($key, $value, $ttl = null) { if ($ttl !== null) { if ($this->getSeconds($ttl) <= 0) { return false; } if (method_exists($this->store, 'add')) { $seconds = $this->getSeconds($ttl); return $this->store->add( $this->itemKey($key), $value, $seconds ); } } if (is_null($this->get($key))) { return $this->put($key, $value, $ttl); } return false; } public function get($key, $default = null) { if (is_array($key)) { return $this->many($key); } $value = $this->store->get($this->itemKey($key)); if (is_null($value)) { $this->event(new CacheMissed($key)); $value = value($default); } else { $this->event(new CacheHit($key, $value)); } return $value; } // namespace \Illuminate\Cache\FileStore public function get($key) { return $this->getPayload($key)['data'] ?? null; } protected function getPayload($key) { $path = $this->path($key); try { $expire = substr( $contents = $this->files->get($path, true), 0, 10 ); } catch (Exception $e) { return $this->emptyPayload(); } if ($this->currentTime() >= $expire) { $this->forget($key); return $this->emptyPayload(); } try { $data = unserialize(substr($contents, 10)); } catch (Exception $e) { $this->forget($key); return $this->emptyPayload(); } $time = $expire - $this->currentTime(); return compact('data', 'time'); }
这里需要说明,所谓互斥锁,其本质是写文件。如果文件不存在或文件内容为空或文件中存储的过期时间小于当前时间,则互斥锁可以顺利获得;否则无法获取到互斥锁。文件内容为固定格式:timestampb:1
。
所谓超时时间,与此处的 timestamp 的值有密切的联系。获取互斥锁时的时间戳,再加上超时时间的秒数,即是此处的 timestamp 的值。
由于 FileStore
中不存在 add()
方法,所以程序会直接尝试调用 get()
方法获取文件中的内容。如果 get()
返回的结果为 NULL
,说明获取互斥锁成功,之后会调用 FileStore
的 put()
方法写文件;否则,说明当前有相同的 Event
在运行,不会再运行新的 Event
。
在调用 put()
方法写文件时,首先需要根据传参计算 eventMutex
的超时时间的秒数,之后再调用 FileStore
中的 put()
方法,将数据写入文件中。
// namespace \Illuminate\Cache\Repository public function put($key, $value, $ttl = null) { /* ... ... */ $seconds = $this->getSeconds($ttl); if ($seconds <= 0) { return $this->forget($key); } $result = $this->store->put($this->itemKey($key), $value, $seconds); if ($result) { $this->event(new KeyWritten($key, $value, $seconds)); } return $result; } // namespace \Illuminate\Cache\FileStore public function put($key, $value, $seconds) { $this->ensureCacheDirectoryExists($path = $this->path($key)); $result = $this->files->put( $path, $this->expiration($seconds).serialize($value), true ); if ($result !== false && $result > 0) { $this->ensureFileHasCorrectPermissions($path); return true; } return false; } protected function path($key) { $parts = array_slice(str_split($hash = sha1($key), 2), 0, 2); return $this->directory.'/'.implode('/', $parts).'/'.$hash; } // namespace \Illuminate\Console\Scheduling\Schedule public function mutexName() { return 'framework'.DIRECTORY_SEPARATOR.'schedule-'.sha1($this->expression.$this->command); }
这里需要重点说明的是 $key
的生成方法以及文件路径的生成方法。$key
通过调用 Event
的 mutexName()
方法生成,其中需要用到 Event
的 $expression
和 $command
属性。其中 $command
为我们定义的命令行,在调用 $schedule->comand()
方法时传入,然后进行格式化,$expression
则为 Event
的运行周期。
以命令行 schedule:test
为例,格式化之后的命令行为 `/usr/local/php/bin/php` `artisan` schedule:test
,如果该命令行设置的运行周期为每分钟一次,即 * * * * *
,则最终计算得到的 $key
的值为 framework/schedule-768a42da74f005b3ac29ca0a88eb72d0ca2b84be
。文件路径则是将 $key
的值再次进行 sha1
计算之后,以两个字符为一组切分成数组,然后取数组的前两项组成一个二级目录,而配置文件中 file
引擎的默认存储路径为 storage/framework/cache/data
,所以最终的文件路径为 storage/framework/cache/data/eb/60/eb608bf555895f742e5bd57e186cbd97f9a6f432
。而文件中存储的内容则为 1642122685b:1
。
⑵ 回调方法
再来说设置的 Event
回调,调用 withoutOverlapping()
方法会为 Event
设置两个回调:一个是 Event
运行完成之后的回调,用于释放互斥锁,即清理缓存文件;另一个是在运行 Event
之前判断互斥锁是否被占用,即缓存文件是否已经存在。
无论 Event
是以 Foreground
的方式运行,还是以 Background
的方式运行,在运行完成之后都会调用 callAfterCallbacks()
方法执行 afterCallbacks
中的回调,其中就有一项回调用于释放互斥锁,删除缓存文件 $this->mutex->forget($this)
。区别就在于,以 Foreground
方式运行的 Event
是在运行完成之后显式的调用这些回调方法,而以 Background
方式运行的 Event
则需要借助 schedule:finish
来调用这些回调方法。
所有在 \App\Console\Kernel
中注册 Event
,都是通过命令行 schedule:run
来调度的。在调度之前,首先会判断当前时间点是否满足各个 Event
所配置的运行周期的要求。如果满足的话,接下来就是一些过滤条件的判断,这其中就包括判断互斥锁是否被占用。只有在互斥锁没有被占用的情况下,Event
才可以运行。
// namespace \Illuminate\Console\Scheduling\ScheduleRunCommand public function handle(Schedule $schedule, Dispatcher $dispatcher) { $this->schedule = $schedule; $this->dispatcher = $dispatcher; foreach ($this->schedule->dueEvents($this->laravel) as $event) { if (! $event->filtersPass($this->laravel)) { $this->dispatcher->dispatch(new ScheduledTaskSkipped($event)); continue; } if ($event->onOneServer) { $this->runSingleServerEvent($event); } else { $this->runEvent($event); } $this->eventsRan = true; } if (! $this->eventsRan) { $this->info('No scheduled commands are ready to run.'); } } // namespace \Illuminate\Console\Scheduling\Schedule public function dueEvents($app) { return collect($this->events)->filter->isDue($app); } // namespace \Illuminate\Console\Scheduling\Event public function isDue($app) { /* ... ... */ return $this->expressionPasses() && $this->runsInEnvironment($app->environment()); } protected function expressionPasses() { $date = Carbon::now(); /* ... ... */ return CronExpression::factory($this->expression)->isDue($date->toDateTimeString()); } // namespace \Cron\CronExpression public function isDue($currentTime = 'now', $timeZone = null) { /* ... ... */ try { return $this->getNextRunDate($currentTime, 0, true)->getTimestamp() === $currentTime->getTimestamp(); } catch (Exception $e) { return false; } } public function getNextRunDate($currentTime = 'now', $nth = 0, $allowCurrentDate = false, $timeZone = null) { return $this->getRunDate($currentTime, $nth, false, $allowCurrentDate, $timeZone); }
有时候,我们可能需要 kill 掉一些在后台运行的命令行,但紧接着我们会发现这些被 kill 掉的命令行在一段时间内无法按照设置的运行周期自动调度,其原因就在于手动 kill 掉的命令行没有调用 schedule:finish 清理缓存文件,释放互斥锁。这就导致在设置的过期时间到达之前,互斥锁会一直被占用,新的 Event 不会再次运行。
【相关推荐:laravel视频教程】
以上就是一文带你了解Laravel schedule调度的运行机制的详细内容,更多请关注gxlsystem.com其它相关文章!