示例代码 - 多进程爬虫
优质
小牛编辑
143浏览
2023-12-01
EasySwoole利用redis队列+定时器+task进程实现的一个多进程爬虫。直接上代码
添加Redis配置信息
修改配置文件,添加Redis配置
"REDIS"=>array(
"HOST"=>'',
"PORT"=>6379,
"AUTH"=>""
)
封装Redis
namespace AppUtilityDb;
use ConfConfig;
class Redis
{
private $con;
protected static $instance;
protected $tryConnectTimes = 0;
protected $maxTryConnectTimes = 3;
function __construct()
{
$this->connect();
}
function connect(){
$this->tryConnectTimes++;
$conf = Config::getInstance()->getConf("REDIS");
$this->con = new Redis();
$this->con->connect($conf['HOST'], $conf['PORT'],2);
$this->con->auth($conf['AUTH']);
if(!$this->ping()){
if($this->tryConnectTimes <= $this->maxTryConnectTimes){
return $this->connect();
}else{
trigger_error("redis connect fail");
return null;
}
}
$this->con->setOption(Redis::OPT_SERIALIZER,Redis::SERIALIZER_PHP);
}
static function getInstance(){
if(is_object(self::$instance)){
return self::$instance;
}else{
self::$instance = new Redis();
return self::$instance;
}
}
function rPush($key,$val){
try{
return $this->con->rpush($key,$val);
// return $ret;
}catch(Exception $e){
$this->connect();
if($this->tryConnectTimes <= $this->maxTryConnectTimes){
return $this->rPush($key,$val);
}else{
return false;
}
}
}
function lPop($key){
try{
return $this->con->lPop($key);
}catch(Exception $e){
$this->connect();
if($this->tryConnectTimes <= $this->maxTryConnectTimes){
return $this->lPop($key);
}else{
return false;
}
}
}
function lSize($key){
try{
$ret = $this->con->lSize($key);
return $ret;
}catch(Exception $e){
$this->connect();
if($this->tryConnectTimes <= $this->maxTryConnectTimes){
return $this->lSize($key);
}else{
return false;
}
}
}
function getRedisConnect(){
return $this->con;
}
function ping(){
try{
$ret = $this->con->ping();
if(!empty($ret)){
$this->tryConnectTimes = 0;
return true;
}else{
return false;
}
}catch(Exception $e){
return false;
}
}
}
定义SysConst
namespace AppUtility;
class SysConst extends CoreComponentSysConst
{
const TASK_RUNNING_NUM = 'TASK_RUNNING_NUM';
}
封装队列
namespace AppModel;
use AppUtilityDbRedis;
class Queue
{
const QUEUE_NAME = 'task_list';
static function set(TaskBean $taskBean){
return Redis::getInstance()->rPush(self::QUEUE_NAME,$taskBean);
}
static function pop(){
return Redis::getInstance()->lPop(self::QUEUE_NAME);
}
}
封装TaskBean
namespace AppModel;
use CoreComponentSplSplBean;
class TaskBean extends SplBean
{
/*
* 仅仅做示例,curl opt 选项请自己写
*/
protected $url;
/**
* @return mixed
*/
public function getUrl()
{
return $this->url;
}
/**
* @param mixed $url
*/
public function setUrl($url)
{
$this->url = $url;
}
protected function initialize()
{
// TODO: Implement initialize() method.
}
}
封装异步执行模型
namespace AppModel;
use AppUtilitySysConst;
use CoreAbstractInterfaceAbstractAsyncTask;
use CoreComponentLogger;
use CoreComponentShareMemory;
use CoreUtilityCurlRequest;
class Runner extends AbstractAsyncTask
{
function handler(swoole_server $server, $taskId, $fromId)
{
// TODO: Implement handler() method.
//记录处于运行状态的task数量
$share = ShareMemory::getInstance();
$share->startTransaction();
$share->set(SysConst::TASK_RUNNING_NUM,$share->get(SysConst::TASK_RUNNING_NUM)+1);
$share->commit();
//while其实为危险操作,while会剥夺进程控制权
while (true){
$task = Queue::pop();
if($task instanceof TaskBean){
$req = new Request($task->getUrl());
$ret = $req->exec()->getBody();
Logger::getInstance("curl")->console("finish url:".$task->getUrl());
}else{
break;
}
}
// Logger::getInstance()->console("async task exit");
$share->startTransaction();
$share->set(SysConst::TASK_RUNNING_NUM,$share->get(SysConst::TASK_RUNNING_NUM)-1);
$share->commit();
}
function finishCallBack(swoole_server $server, $task_id, $resultData)
{
// TODO: Implement finishCallBack() method.
}
}
注册事件
在Conf/Event.php中
- 在启动前先清理共享内存信息。
function frameInitialized()
{
// TODO: Implement frameInitialized() method.
ShareMemory::getInstance()->clear();
}
- 注册定时器,定时去获取任务
function onWorkerStart(swoole_server $server, $workerId)
{
// TODO: Implement onWorkerStart() method.
//为第一个进程添加唤起任务执行的定时器;
if($workerId == 0){
Timer::loop(1000,function (){
$share = ShareMemory::getInstance();
//请勿使得所有worker全部处于繁忙状态 危险操作
if($share->get(SysConst::TASK_RUNNING_NUM) < 2){
AsyncTaskManager::getInstance()->add(Runner::class);
}
});
}
}
任务投递控制器
任意建立一个控制器,添加以下两个方法。function addTask(){
$url = $this->request()->getRequestParam("url");
if(empty($url)){
$url = 'http://wiki.swoole.com/';
}
$bean = new TaskBean();
$bean->setUrl($url);
//做异步投递
AsyncTaskManager::getInstance()->add(function ()use($bean){
Queue::set($bean);
});
$this->response()->writeJson(200,null,"任务投递成功");
}
function status(){
$num = ShareMemory::getInstance()->get(SysConst::TASK_RUNNING_NUM);
$this->response()->writeJson(200,array(
"taskRuningNum"=>$num
));
}
执行
启动EasySwoole,访问addTask方法往Redis队列中添加任务,并等待执行结果。
本例子仅供参考,未做详尽错误处理。