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

kution

考虑用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://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,
　　];//启动Supervisorqpm\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';
　　}
　　}
　　
　　
真实的生产环境，用队列替换文件输入，即可实现持久运行的生产者／消费者模型的程序。