低调大师

您现在的位置是: 首页 > PHP使用QPM实现多进程并行任务处理程序

文章详情

PHP使用QPM实现多进程并行任务处理程序

2015-3-10 21:32 6已围观 收藏 加入我们

考虑用PHP实现以下场景: 有一个抓站的URL列表保存在队列里,后台程序读取这个队列,然后转交给子进程去抓取HTML存放到文件里。 为了提高效率,允许多任务并行执行,但为了避免机器负载过高,限制了最大的并行任务数(为了测试方便,我们把这个数设为3),当队列中取到 END标记时,程序结束运行。

这个场景用QPM的Supervisor::taskFactoryMode()实现,非常简单。

QPM全名是 Quick Process Management Module for PHP. PHP 是强大的web开发语言,以至于大家常常忘记PHP 可以用来开发健壮的命令行(CLI)程序以至于daemon程序。 而编写daemon程序免不了与各种进程管理打交道。QPM正式为简化进程管理而开发的类库。QPM的项目地址是:https://github.com/Comos/qpm

为了,简化测试环境,我们可以用一个文本文件来模拟队列的数据。完整的例子文件看这里:spider_task_factory_data.txt

http://ent.ifeng.com/
http://news.sina.com.cn/
http://news.ifeng.com/
http://news.163.com/
http://news.sohu.com/
http://ent.sina.com.cn/
http://ent.ifeng.com/
...
END

使用QPM的taskFactoryMode之前,我们需要准备一个TaskFactory类。 我们将其命名为 SpiderTaskFactory,SpdierTaskFactory 的工厂方法fetchTask 正常返回 Runnable的子类的实例。当碰到END或文件结束,则throw StopSignal,这样程序就会终止。

以下是组装 Supervisor 并执行的代码片段。完整的例子见:spider_task_factory.php

//如果没有从参数指定输入,把spider_task_factory_data.txt作为数据源

$input = isset($argv[1]) ? $argv[1] : __DIR__.'/spider_task_factory_data.txt';
$spiderTaskFactory = new SpiderTaskFactory($input);$config = [
    //指定taskFactory对象和工厂方法
    'factoryMethod'=>[$spiderTaskFactory, 'fetchTask'],
    //指定最大并发数量为3
    'quantity' => 3,
];
//启动Supervisor
qpm\supervisor\Supervisor::taskFactoryMode($config)->start();

SpiderTaskFactory 的实现如下:

/**
 * 任务工厂,必须实现 fetchTask方法。
 * 该方法正常返回
 * 
*/class SpiderTaskFactory {
    private $_fh;
    public function __construct($input) {
        $this->_input = $input;
        $this->_fh = fopen($input, 'r');
        if ($this->_fh === false) {
            throw new Exception('fopen failed:'.$input);
        }
    }
    public function fetchTask() {
      while (true) {
        if (feof($this->_fh)) {
                throw new qpm\supervisor\StopSignal();
        }
        $line = trim(fgets($this->_fh));
        if ($line == 'END') {
            throw new qpm\supervisor\StopSignal();
        }
        if (empty($line)) {
            continue;
        }
        break;
      }
      return new SpiderTask($line);
    }
}

SpiderTask 的实现如下:

/**
 * 在子进程中执行任务的类
 * 必须实现 qpm\process\Runnable 接口
 */
 class SpiderTask implements qpm\process\Runnable {
    private $_target;
    public function __construct($target) {
      $this->_target = $target;
    }
    
    //在子进程中执行的部分
    public function run() {
      $r = @file_get_contents($this->_target);    
      if ($r===false) {        
        throw new Exception('fail to crawl url:'.$this->_target);
      }
      file_put_contents($this->getLocalFilename(), $r);   
    }
    private function getLocalFilename() {
      $filename = str_replace('/', '~', $this->_target);    
      $filename = str_replace(':', '_', $filename);    
      $filename = $filename.'-'.date('YmdHis');    
      return __DIR__.'/_spider/'.$filename.'.html';
    }
}

 真实的生产环境,用队列替换文件输入,即可实现持久运行的生产者/消费者模型的程序。

关于 QPM的使用,可以参考:

QPM介绍

安装和使用PHP进程管理框架 QPM
使用qpm创建daemon程序
使用QPM编写PHP 多进程程序
 PHP使用QPM实现多进程并行任务处理程序


文章转载至:https://my.oschina.net/bigbigant/blog/385021
收藏 (0)

文章评论

共有0条评论来说两句吧...