PHP多进程用例--swoole和pcntl

<?php

class MyMulProcess {
/**
* 进程列表
* @var array
*/
private $_works = array();

/**
* 对应CPU核心
* @var int
*/
private $_cpuNums = 24;

/**
* 进程名字
* @var string
*/
private $_name = "MyProcess";

/**
* php 命令路径
* @var string
*/
private $_phpbin = "/usr/bin/php";

/**
* 实例对象
* @var null
*/
private static $_instance = null;

/**
* 默认是否存在swoole模块
* @var bool
*/
private static $_isSwoole = false;

/**
* 私有构造函数
* MyMulProcess constructor.
*/
private function __construct() {
}

/**
* 返回实例对象
* @return MyMulProcess|null
*/
public static function getInst() {
if ( self::$_instance ) {
return self::$_instance;
}

//模块载入
if ( extension_loaded("swoole") ) {
self::$_isSwoole = true;
}

self::$_instance = new self();
return self::$_instance;
}

/**
* 设置CPU核数
* @param $nums
* @return $this
*/
public function setCpuNums($nums) {
$this->_cpuNums = $nums;
return $this;
}

/**
* 设置主进程
* @param $name
* @return $this
*/
public function setProcessName($name) {
$this->_name = $name;

return $this;
}

/**
* 设置php可执行的路径
* @param $phpbin
* @return $this
*/
public function setPhpBin($phpbin) {
$this->_phpbin = $phpbin;

return $this;
}

public function start ($total, $pageSize, $phpfile, $extras = array()) {
if (self::$_isSwoole) {
swoole_set_process_name($this->_name);
} else if (extension_loaded("cli_set_process_title")) {
cli_set_process_title($this->_name);
}

$name = $this->_name;
$phpbin = $this->_phpbin;
$count = 0;
$pages = ceil($total/$pageSize);
while (true) {
for ($i = $count; $i<$pages; $i++) {
if ( count($this->_works) >= $this->_cpuNums ) {
break;
}

if (self::$_isSwoole) {
$process = new swoole_process(function (swoole_process $work) use
($name, $phpbin, $phpfile, $i, $pageSize,$pages, $extras) {
array_unshift($extras, $pages);
array_unshift($extras, $pageSize);
array_unshift($extras, $i);
array_unshift($extras, $name);
array_unshift($extras, $phpfile);
$work->exec($phpfile, $extras);

$work->exit(0);
}, false, true);

$pid = $process->start();

if (empty($pid)) {
echo swoole_errno().''.swoole_strerror();
continue;
}

$this->_works[$pid] = $process;
} else {
$pid = pcntl_fork();
if ($pid == 0) { //子进程
array_unshift($extras, $pages);
array_unshift($extras, $pageSize);
array_unshift($extras, $i);
array_unshift($extras, $name);
array_unshift($extras, $phpfile);
pcntl_exec($phpbin, $extras);

exit(0);
} else if ($pid > 0) {
$this->_works[$pid] = $pid;
} else {
continue;
}
}
$count++;
}

foreach ($this->_works as $work) {
if (self::$_isSwoole) {
if ($ret = $work->wait(false)) {
unset($this->_works[$work->pid]);
}
} else {
$status = 0;
$pid = pcntl_waitpid($work, $status, WNOHANG);
if($pid == -1 || $pid > 0) {
unset($this->_works[$pid]);
}
}
}
if (empty($this->_works) && $count >= $pages) {
echo "empty works" . " ";
break;
}
// 防止该进程抢占CPU
sleep(1);
}
}
}

//index.php 接受参数
/**
* 命令行 接收参数定义说明
* -------------------------------
* 第零个 参数 phpdiamante文件完整路径
* 第一个 参数 进程名字 $name
* 第二个 参数 页数从0开始$i, 从数据库拿数据的话,需要[$i*$limit]位置开始
* 第三个 参数 每页大小$limit
* 第四个 参数 总页数$pages,如果需要判断($i+1 >= $pages)最后一个可以跑所有的数据(要不然会丢失一些数据)
* 第五个 参数 自己传入
* ........
* 第N个 参数 这个N不能太大,要不然命令直接挂掉
*/
$name = $_SERVER['argv'][1];
$i = $_SERVER['argv'][2];
$limit = $_SERVER['argv'][3];
$pages = $_SERVER['argv'][4];

//设置进程名字,主要是好管理而已
if (extension_loaded("swoole")) {
swoole_set_process_name($name.'-child-'.$i);
} else if(extension_loaded("cli_set_process_title")) {
cli_set_process_title($name.'-child-'.$i);
}

//测试代码
// 测试
MyMulProcess::getInst()
->setCpuNums(5) // CPU数
->setProcessName("TTTT") // 进程名称
->setPhpBin("/usr/bin/php7") // PHP 可执行路径
->start($total, $pageSize, "/index.php", array('bbb', 'ccc'));