基于Swoole 下的 Hyperf 框架构建爬虫平台-定时任务设置与执行
Nash
共 32427字，需浏览 65分钟
· 2023-10-10
上期内容，我们已经构建了一个可以执行的命令系统，同时我们规定了命令在平台上开放的参数和选项，今天我们将命令再次封装成为一个可执行的任务，简单来说就是将对应的命令选出来，针对其开放的参数和接口进行选值设定，并调整定时执行的模式和启动周期，在服务运行过程中保持检测和运行即可。
老规矩，直接上实操，下期再来填坑，文章内容全面升级，将采取分段方式讲解整个实操步骤。
1、创建数据库迁移文件，设置数据库结构
执行命令php bin/hyperf.php gen:migration create_spiders_table，创建爬虫任务数据库表的迁移文件，并修改内容：
<?php
use Hyperf\Database\Schema\Schema;use Hyperf\Database\Schema\Blueprint;use Hyperf\Database\Migrations\Migration;
class CreateSpidersTable extends Migration{/*** Run the migrations.*/public function up(): void {Schema::create('spiders', function (Blueprint $table) {$table->bigIncrements('id');$table->string('name', 32)->comment('爬虫名称');$table->unsignedBigInteger('command_id')->comment('命令ID')->index('idx_command_id');$table->json('arguments')->comment('指令列表');$table->json('options')->comment('参数列表');$table->string('crontab_mode', 64)->comment('定时模式');$table->enum('state', ['enabled', 'disabled'])->comment('状态');$table->softDeletesTz();$table->timestampsTz();});}
/*** Reverse the migrations.*/public function down(): void {Schema::dropIfExists('spiders');}}
执行命令 php bin/hyperf.php gen:migration create_spider_logs_table，创建爬虫任务数据库表的迁移文件，并修改内容：
<?php
use Hyperf\Database\Schema\Schema;use Hyperf\Database\Schema\Blueprint;use Hyperf\Database\Migrations\Migration;
class CreateSpiderLogsTable extends Migration{/*** Run the migrations.*/public function up(): void {Schema::create('spider_logs', function (Blueprint $table) {$table->bigIncrements('id');$table->unsignedBigInteger('spider_id')->comment('爬虫ID')->index('idx_spider_id');$table->enum('event', ['trigger', 'start', 'print', 'finish'])->comment('事件名称：trigger 任务被触发；start 任务启动；print 任务回显；finish 任务结束');$table->longText('message')->comment('事件信息');$table->softDeletesTz();$table->timestampsTz();});}
/*** Reverse the migrations.*/public function down(): void {Schema::dropIfExists('spider_logs');}}
很显然，我们使用 spiders 表来记录创建的任务所使用的命令和对应的参数，使用 spider_logs 来记录任务执行的过程
执行命令 php bin/hyperf.php migrate 执行数据库迁移；
执行命令 php bin/hyperf.php gen:model spiders 和 php bin/hyperf.php gen:model spider_logs 分别创建两个新的数据库表的模型类
2、完善 Spider 运行方法
执行命令：composer require dragonmantank/cron-expression 安装 crontab 解析工具，下面修改 App\Model\Spider 类，完善 spider 的执行流程
<?php
declare(strict_types=1);
namespace App\Model;
use App\Exception\CommandException;use Carbon\Carbon;use Cron\CronExpression;use Exception;use Hyperf\Context\ApplicationContext;use Hyperf\Redis\Redis;use Psr\Container\ContainerExceptionInterface;use Psr\Container\NotFoundExceptionInterface;use RedisException;use Symfony\Component\Process\Process;
/*** @property int $id* @property string $name* @property int $commandId* @property string $arguments* @property string $options* @property string $crontabMode* @property string $state* @property Carbon $deletedAt* @property Carbon $createdAt* @property Carbon $updatedAt*/class Spider extends Model{/*** The table associated with the model.*/protected ?string $table = 'spiders';
/*** The attributes that are mass assignable.*/protected array $fillable = ['id', 'name', 'command_id', 'arguments', 'options', 'crontab_mode', 'state', 'deleted_at', 'created_at', 'updated_at'];
/*** The attributes that should be cast to native types.*/protected array $casts = ['id' => 'integer', 'command_id' => 'integer', 'created_at' => 'datetime', 'updated_at' => 'datetime', 'deleted_at' => 'datetime'];
/*** @return self[]* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/public static function getDue(): array {return array_filter(array_map(fn($item)=>($spider=unserialize($item)) instanceof Spider && $spider->cronExpression->isDue()?$spider:null, ApplicationContext::getContainer()->get(Redis::class)->hGetAll('crontab')));}
/*** @return string|null*/public function exec(): ?string {SpiderLog::create(['spiderId' => $this->id,'event' => 'trigger','message' => json_encode(['time' => date('Y-m-d H:i:s')], JSON_UNESCAPED_UNICODE),]);/** @var Command $command */$command = Command::query()->find($this->commandId);$argsData = json_decode($command->argument, true);$args = json_decode($this->arguments, true);$params = [];foreach ($argsData as $argsItem)$params[] = $args[$argsItem['name']];$options = json_decode($this->options, true);SpiderLog::create(['spiderId' => $this->id,'event' => 'start','message' => json_encode(['arguments' => $params, 'options' => $options], JSON_UNESCAPED_UNICODE),]);$leave = '';$command->exec($params, $options, function ($type, $buffer) use (&$leave) {SpiderLog::create(['spiderId' => $this->id,'event' => 'print','message' => $buffer,]);$arrayData = explode("\n", $leave . $buffer);$leave = array_pop($arrayData);foreach ($arrayData as $item) {var_dump($item);// 提交到 MongoDB 中}});try {$result = $command->wait();} catch (CommandException $e) {$result = $e->getMessage();}SpiderLog::create(['spiderId' => $this->id,'event' => 'finish','message' => json_encode(['arguments' => $params, 'options' => $options], JSON_UNESCAPED_UNICODE),]);return $result;}
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/public static function cleanCache(): void {ApplicationContext::getContainer()->get(Redis::class)->del('crontab');}
/** @var CronExpression|null */protected ?CronExpression $cronExpression = null;
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/public function cache(): void {$this->cronExpression = new CronExpression($this->crontabMode);if ($this->state === 'enabled')ApplicationContext::getContainer()->get(Redis::class)->hSet('crontab', strval($this->id), serialize($this));elseApplicationContext::getContainer()->get(Redis::class)->hDel('crontab', strval($this->id));}}
这个类的使用其实也挺简单的：
cleanCache 静态方法则是清除全部 Redis 缓存
cache 方法可以将当前数据同步到 Redis 缓存中
getDue 静态方法可以将 Redis 缓存中达到执行时间的 Spider 对象返回
exec 方法构建当前对象中的参数、选项和 Command 对象，然后调用 Command 对象的 exec 方法开始执行对应的命令，在不同的时机会保存对应的 SpiderLog 记录
3、设置系统启动时全量构建 Spider 记录的缓存
修改 config\autoload\server.php 文件，在启动添加 start 回调：
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/
use App\Model\Spider;use Hyperf\Server\Event;use Hyperf\Server\Server;use Swoole\Constant;
return ['mode' => SWOOLE_PROCESS,'servers' => [['name' => 'http','type' => Server::SERVER_WEBSOCKET,'host' => '0.0.0.0','port' => 9501,'sock_type' => SWOOLE_SOCK_TCP,'callbacks' => [Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'],Event::ON_HAND_SHAKE => [Hyperf\WebSocketServer\Server::class, 'onHandShake'],Event::ON_MESSAGE => [Hyperf\WebSocketServer\Server::class, 'onMessage'],Event::ON_CLOSE => [Hyperf\WebSocketServer\Server::class, 'onClose'],],],],'settings' => [Constant::OPTION_ENABLE_COROUTINE => true,Constant::OPTION_WORKER_NUM => swoole_cpu_num(),Constant::OPTION_PID_FILE => BASE_PATH . '/runtime/hyperf.pid',Constant::OPTION_OPEN_TCP_NODELAY => true,Constant::OPTION_MAX_COROUTINE => 100000,Constant::OPTION_OPEN_HTTP2_PROTOCOL => true,Constant::OPTION_MAX_REQUEST => 100000,Constant::OPTION_SOCKET_BUFFER_SIZE => 2 * 1024 * 1024,Constant::OPTION_BUFFER_OUTPUT_SIZE => 2 * 1024 * 1024,
Constant::OPTION_TASK_WORKER_NUM => swoole_cpu_num() * 4,Constant::OPTION_TASK_ENABLE_COROUTINE => false,],'callbacks' => [Event::ON_WORKER_START => [Hyperf\Framework\Bootstrap\WorkerStartCallback::class, 'onWorkerStart'],Event::ON_PIPE_MESSAGE => [Hyperf\Framework\Bootstrap\PipeMessageCallback::class, 'onPipeMessage'],Event::ON_WORKER_EXIT => [Hyperf\Framework\Bootstrap\WorkerExitCallback::class, 'onWorkerExit'],
Event::ON_TASK => [Hyperf\Framework\Bootstrap\TaskCallback::class, 'onTask'],Event::ON_FINISH => [Hyperf\Framework\Bootstrap\FinishCallback::class, 'onFinish'],Event::ON_START => function(){Spider::cleanCache();$spiders = Spider::query()->where('state', 'enabled')->get();foreach ($spiders as $spider)$spider->cache();var_dump('cache success!!!');}],];
如上例代码展示，添加的 ON_START 回调，在系统启动时，清除全部 redis 的缓存，然后将处于 enabled 状态的 Spider 记录进行缓存。
4、设置定时器，定时增量和全量更新 Spider 的缓存
执行命令：composer require hyperf\crontab 为系统添加定时任务执行的能力。
修改配置文件：config\autoload\processes.php 文件
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/
return [Hyperf\Crontab\Process\CrontabDispatcherProcess::class,];
新增或修改配置文件：config\autoload\crontab.php 文件
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/return ['enable' => true,];
创建工具类：App\Utils\CrontabExec
<?php
namespace App\Utils;
use App\Model\Spider;use App\Process\CrontabDispatcherProcess;use Carbon\Carbon;use Hyperf\Crontab\Annotation\Crontab;use Psr\Container\ContainerExceptionInterface;use Psr\Container\NotFoundExceptionInterface;use RedisException;
class CrontabExec{/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/#[Crontab(rule: '* * * * *', name: 'crontab-cache', singleton: true, memo: '定时任务缓存更新命令')]public function cacheSpider(): void {$spiders = Spider::query()->where('state', 'enabled')->where('updated_at', '>', (new Carbon())->subSeconds(90))->get();foreach ($spiders as $spider)$spider->cache();}
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/#[Crontab(rule: '0 0 * * *', name: 'crontab-cache-all', singleton: true, memo: '定时任务缓存全部命令')]public function cacheAllSpider(): void {$spiders = Spider::query()->get();foreach ($spiders as $spider)$spider->cache();}}
以上代码也比较简单，为系统添加了 crontab 任务执行进程，并在进程中设置了两个定时任务，分别为每分钟执行一次的增量缓存更新和每天执行一次的全量缓存更新
5、设置定时器，定时执行 Spider 任务
添加类：App\Process\CrontabDispatcherProcess
<?php
namespace App\Process;
use Hyperf\Context\ApplicationContext;use Hyperf\Contract\ConfigInterface;use Hyperf\Coroutine\Concurrent;use Hyperf\Coroutine\Coroutine;use Hyperf\Engine\Contract\ChannelInterface;use Hyperf\Process\ProcessManager;use Psr\Container\ContainerExceptionInterface;use Psr\Container\NotFoundExceptionInterface;use Throwable;
class CrontabDispatcherProcess extends \Hyperf\Crontab\Process\CrontabDispatcherProcess{/** @var self|null */private static ?self $_instance;
/** @var ?ChannelInterface */private ?ChannelInterface $channel;
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface*/public function handle(): void {self::$_instance = $this;$this->channel = ApplicationContext::getContainer()->get(ChannelInterface::class);Coroutine::create(function () {$concurrent = new Concurrent(intval(ApplicationContext::getContainer()->get(ConfigInterface::class)->get('crontab.pool')));while (ProcessManager::isRunning()) {try {$concurrent->create($this->channel->pop());} catch (Throwable) {continue;}}});parent::handle();}
/*** @param callable $callback* @return bool*/public static function pushExec(callable $callback): bool {return self::$_instance->channel->push($callback);}}
修改配置文件：config\autoload\processes.php 文件（这个文件在上一步操作已经调整过，再次调整，使用自定义的定时任务执行进程来替代原本的定时任务执行进程）
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/
return [App\Process\CrontabDispatcherProcess::class,];
修改配置文件：config\autoload\crontab.php 文件（添加一个 pool 配置）
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/return ['enable' => true,'pool' => 128,];
修改配置文件：config\autoload\dependencies.php 文件
<?php
declare(strict_types=1);/*** This file is part of Hyperf.** @link https://www.hyperf.io* @document https://hyperf.wiki* @contact group@hyperf.io* @license https://github.com/hyperf/hyperf/blob/master/LICENSE*/
return [Hyperf\Engine\Contract\ChannelInterface::class => Hyperf\Engine\Channel::class,Hyperf\Crontab\Strategy\StrategyInterface::class => Hyperf\Crontab\Strategy\CoroutineStrategy::class,];
修改 App\Utils\CrontabExec 类，添加方法 execute
<?php
namespace App\Utils;
use App\Model\Spider;use App\Process\CrontabDispatcherProcess;use Carbon\Carbon;use Hyperf\Crontab\Annotation\Crontab;use Psr\Container\ContainerExceptionInterface;use Psr\Container\NotFoundExceptionInterface;use RedisException;
class CrontabExec{/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/#[Crontab(rule: '* * * * *', name: 'crontab-exec', singleton: false, memo: '定时任务调起执行脚本')]public function execute(): void {array_map(fn(Spider $spider)=>CrontabDispatcherProcess::pushExec(fn()=>$spider->exec()),Spider::getDue());}
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/#[Crontab(rule: '* * * * *', name: 'crontab-cache', singleton: true, memo: '定时任务缓存更新命令')]public function cacheSpider(): void {$spiders = Spider::query()->where('state', 'enabled')->where('updated_at', '>', (new Carbon())->subSeconds(90))->get();foreach ($spiders as $spider)$spider->cache();}
/*** @return void* @throws ContainerExceptionInterface* @throws NotFoundExceptionInterface* @throws RedisException*/#[Crontab(rule: '0 0 * * *', name: 'crontab-cache-all', singleton: true, memo: '定时任务缓存全部命令')]public function cacheAllSpider(): void {$spiders = Spider::query()->get();foreach ($spiders as $spider)$spider->cache();}}
上例代码中完成的基本流程是：
自定义一个 CrontabDispatcherProcess 进程类继承并替代系统定义的CrontabDispatcherProcess 进程类，通过 Channel 来推入和推出callback， 并将 callback 投放到协程池中执行
修改 CrontabExec 类，设置一个每分钟执行一次的定时器，将符合执行时间的 Spider 对象的 exec 方法作为回调丢入到 Channel 中
最后：功能验证
新建文件 bin\test.php
(function ($url) {
function getDetailInformation(string $url): void {$curl = curl_init();curl_setopt($curl, CURLOPT_URL, $url);curl_setopt($curl, CURLOPT_HEADER, false);curl_setopt($curl, CURLOPT_RETURNTRANSFER, true);curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, false);curl_setopt($curl, CURLOPT_SSL_VERIFYHOST, false);if (!($content = curl_exec($curl))) {throw new Exception(curl_error($curl), curl_errno($curl));}$infos = explode('<div class="book-mid-info">', $content);foreach ($infos as $index => $info) {if ($index <= 0)continue;preg_match('/\\<a href="(\\/\\/www\\.qidian\\.com\\/book\\/\\d+\\/)" target="\\_blank" data\\-eid="qd\\_B58" data\\-bid="\d+" title="[\Wa-zA-Z0-9]+"\\>([\Wa-zA-Z0-9]+)\\<\\/a\\>/', $info, $matches);preg_match('/\\<a class="name" href="\\/\\/my\\.qidian\\.com\\/author\\/\\d+\\/" data\\-eid="qd\\_B59" target="\\_blank"\\>([\Wa-zA-Z0-9]+)\\<\\/a\\>/', $info, $author);echo json_encode(['url' => 'https:' . $matches[1], 'title' => $matches[2], 'author' => $author[1]], JSON_UNESCAPED_UNICODE), PHP_EOL;}curl_close($curl);}
getDetailInformation($url);
})('https://www.qidian.com/free/all/action1-/');
这是一个很简单的爬虫脚本，获取某点小说网站免费小说第一页的书目列表，并输出到标准输出中
修改 commands 表的 php 命令的参数和选项：
这里将不安全的 -r 选项删除，并添加 script 参数作为执行脚本文件
这里设置爬虫任务调用 php 命令，并将 script 参数设置为 test.php 的目录，并将 -d 选项设置为 memory_limit=1G ，定时器模式为：* * * * *，既每分钟执行一次，启动项目之后，观察命令行输出：
查看 spider_logs 记录信息：
由此可得，我们的定时任务已经可以按照预期的行为开始工作了。
今天的内容就是这些了，后续本人考虑将功能代码或一些参考文档开放出来，希望有需要的朋友可以关注我，一起参考、学习和交流。
----------
正文结束，以下是恰饭内容，关注公众号，超想要美了优选，享受外卖、出行、加油、用车、酒店优惠，更有超多福利等你发现，感谢您的支持
评论
真高！比亚迪员工爆料比亚迪在越南的薪资水平：基本工资480万，全勤奖35万，交通补助20万，餐补110万，每周6天，每天10小时
上一篇：某大公司为逼迫员工离职，竟然把他的工位安排到厕所旁，没想到他直接开始记录领导的如厕时间，还发到公司大群...对此，你怎么看？--完--PS：欢迎在留言区留下你的观点，一起讨论提高。如果今天的文章让你有新的启发，欢迎转发分享给更多人。全文完，感谢你的耐心阅读。如果你还想看到我的文章，请一定给本
开发者全社区
0
英伟达Blackwell平台网络配置分析
本文来自“英伟达Blachwell平台网络配置详解”。GTC大会英伟达展示了全新的 Blackwell 平台系列产品，包括 HGX B100 服务器、NVLINK Switch、GB200Superchip Computer Node、Quantum X800 交换机和 CX8 网卡（InfiniB
架构师技术联盟
0
某大公司为逼迫员工离职，竟然把他的工位安排到厕所旁，没想到他直接开始记录领导的如厕时间，还发到公司大群...
上一篇：字节的跳动职级与薪资（2024年）我们与公司间的合作，宛如两艘船只在茫茫大海上相互依靠，共同抵御风浪，携手驶向成功的彼岸。然而，当航向开始产生分歧，或是波涛汹涌的风浪改变了我们的初衷，我们或许应当冷静地选择和平分手，而非在风雨中硬撑。最近，一位网友的遭遇引起了广大职场人的关注和热议。这位网友
开发者全社区
0
金融研究 | 使用Python测量关键审计事项的「信息含量」
Tips: 公众号推送后内容只能更改一次，且只能改20字符。如果内容出问题，或者想更新内容， 只能重复推送。为了更好的阅读体验，建议阅读本文博客版， 链接地址https://textdata.cn/blog/2023-01-13-information-content-of-critical-aud
大邓和他的Python
0
我看阿里的年终奖总算发了！
到4月底了，这两天看朋友圈，发现阿里的年终奖终于发了，问了问老同学，也从网上检索了不少信息，基本搞清楚了阿里今年的年终奖情况。近来来阿里一些集团对绩效等级做了较大的调整，以前的旧绩效系统中，绩效分为3.25、3.5、3.75、4和5五个等级，其中4和5是较高绩效等级，较少见。而且之前3.5绩效内部划
公子龙
0
CVPR 2024｜大视觉模型的开山之作！无需任何语言数据即可打造大视觉模型
↑ 点击蓝字 关注极市平台作者丨科技猛兽编辑丨极市平台极市导读 本文提出一种序列建模 (sequential modeling) 的方法，不使用任何语言数据，训练大视觉模型。>>加入极市CV技术交流群，走在计算机视觉的最前沿本文目录1 序列建模打造大视觉模型(来自 U
极市平台
1
金融研究(更新) | 使用Python构建关键审计事项的「信息含量」
Tips: 公众号推送后内容只能更改一次，且只能改20字符。如果内容出问题，或者想更新内容， 只能重复推送。为了更好的阅读体验，建议阅读本文博客版， 链接地址https://textdata.cn/blog/2023-01-13-information-content-of-critical-aud
大邓和他的Python
0
字节的跳动职级与薪资（2024年）
上一篇：阿里公布年终奖，P7, 3.5+，22W年终奖，还有35W长期现金激励，真香字节跳动自2012年3月成立以来，已经迅速成长为一个全球性的科技公司。其产品和服务已经遍布全球150多个国家与地区，并且支持超过75种不同的语言。在字节跳动的官方网站上，列出了一系列引人注目的产品和服务，包括但不限于
开发者全社区
0