crazy_god_API/lib/mp_work.php

<?php
$actList 	  = array("start","stop","reload");//命令合集
$pidFile 	  = dirname(__FILE__).'/../deBUG/work/work.pid';//正在运行的进程-记录
$worklogFile  =dirname(__FILE__).'/../deBUG/work/work.log';//运行日志信息
$process_num  = 2;
$children_list = [];
$masterPid 	  = 0;
$chiledNum    = 0;
//命令提示
function useageExit(){
	global $argv,$actList;
	echo "may be you can try:{$argv[0]} [ ".implode(' | ', $actList)." ]\n";
	exit(1);
}
/**
Note:杀死进程
@var $type 类型stop reload
*/
function killPids($type){
	global $masterPid,$pidFile,$worklogFile;
	if(!isset($type)) exit(0);
	$pidStr = file_get_contents($pidFile);
	$pidlist = $pids = explode(" ", $pidStr);
	foreach ($pids as $k=>$pid) {
		if($k==0){//reload跳过master进程
			if($type=='reload'||$type=='restart'){
				$masterPid = $pid;
				continue;				
			}
		}
		exec("kill -9 $pid");
		file_put_contents($worklogFile,date('Y-m-d H:i:s')."##{$type}--kill {$pid} \n", FILE_APPEND | LOCK_EX);	
		unset($pidlist[$k]);			
	}
	unlink($pidFile);//删除进程记录文件
}
//创建多子进程
function createWorkProcess($process_num) {
    global $children_list,$chiledNum;
    for($i=0;$i<$process_num;$i++){
    	$pid = pcntl_fork();
    	if($pid == -1){
			exit("err:fork children fail");
		}
		$chiledNum++;
		if($pid == 0){//业务功能逻辑部分
			$tpid = posix_getpid();//当前子进程pid			
			cli_set_process_title("mp_work_{$chiledNum}");//设置当前进程名
			doWork2($chiledNum);
			exit(0);
		}
		array_push($children_list, $pid);
    }
    return $process_num==1?$pid:$children_list;
}

//======工作业务逻辑========
function doWork($chiledNum){
    include(dirname(__FILE__)."/../crontab/crontab.inc.php");
    include(dirname(__FILE__)."/../com.php");
    $memoryLogFile = dirname(__FILE__)."/../deBUG/work/memoryLog_{$chiledNum}.log";
    $thisWPid = posix_getpid();
    while (true) {
        sleep(2);
        $res = oo::async()->do_async($chiledNum);
        if(!$res){
            oo::commonOprDb('common')->close();
        }
    }
}
function doWork2($chiledNum){
    include(dirname(__FILE__)."/../crontab/crontab.inc.php");
	include(dirname(__FILE__)."/../com.php");
	$memoryLogFile = dirname(__FILE__)."/../deBUG/work/memoryLog_{$chiledNum}.log";
    $thisWPid = posix_getpid();
	while (true) {
		sleep(5);
        file_put_contents($memoryLogFile,date('Y-m-d H:i:s')."##PID：{$thisWPid} -chiled:{$chiledNum} -do:ok \n");
		try{
//            $memoryOne = memory_get_usage()/1024;
		    $res = oo::async()->do_async($chiledNum);
//		    $memoryTwe = memory_get_usage()/1024;
//		    file_put_contents($memoryLogFile, date('Y-m-d H:i:s')."##PID：{$thisWPid} - startMemory:{$memoryOne} - endMemory:{$memoryTwe}\n",FILE_APPEND | LOCK_EX);
            if(!$res){
                $err = error_get_last();
                throw new Exception('任务异常-'.json_encode($err));
            }

        }catch (Exception $e){
		    $msg = $e->getMessage();
            file_put_contents($memoryLogFile,date('Y-m-d H:i:s')."##PID：{$thisWPid} - error:{$msg} \n");
        }
		//大于100M内存退出程序，防止内存泄漏被系统杀死导致任务终端
		if(memory_get_usage()>100*1024*1024){
		    exit('内存溢出');
		}
	}
}
//==========================

if(!isset($argv[1])||!in_array($argv[1],$actList)){
	useageExit();
}
//结束进程
if($argv[1] == 'stop'){
	killPids('stop');
	file_put_contents($worklogFile, date('Y-m-d H:i:s')."##work is stop <<===\n=======\n", FILE_APPEND | LOCK_EX);
	exit(0);
}

//验证扩展
if (!extension_loaded("pcntl")) {
    die("require pcntl extension loaded");
}
//验证运行模式
if (substr(php_sapi_name(), 0, 3) !== 'cli') {
    die("This Programe can only be run in CLI mode");
}

//重载子进程(尚未完整)
if($argv[1] == 'reload'){
	exit(0);
	killPids('reload');
	createWorkProcess($process_num);//创建子进程
	array_unshift($children_list,$masterPid);
	file_put_contents($pidFile,implode(" ", $children_list));
	file_put_contents($worklogFile, date('Y-m-d H:i:s')."##work is reload \n",FILE_APPEND | LOCK_EX);
}

//启动进程
if($argv[1]!= 'start')
{
    usageExit();
}
if(is_file($pidFile))
{
    die("mp_work is run. \n");
}
$pid = pcntl_fork();//分出子进程下行
// 创建子进程失败
if ( $pid == -1 ) {
    die("err: start server fail. can not create process.\n");
} elseif ( $pid > 0 ) {
    exit(0);
}

//创建新会话，防止工作进程关联控制终端
if(-1 ==posix_setsid()){
	die("err: create session id fail");
}
$masterPid = posix_getpid();//当前进程pid
cli_set_process_title('mp_work_master');//设置当前进程名
createWorkProcess($process_num);//创建子进程
$childstr = json_encode($children_list);
echo "work is start===master:{$masterPid}===children_list:{$childstr}"."\r\n";
array_unshift($children_list,$masterPid);//master进程入进程组
file_put_contents($pidFile,implode(" ", $children_list));
file_put_contents($worklogFile, date("Y-m-d H:i:s")."##work is start ===>>\n", FILE_APPEND | LOCK_EX);


//监控
while (true) {
    $pid = pcntl_wait($status/*, WNOHANG*/);
    if ( $pid > 0 && in_array($pid,$children_list) ) {
		sleep(2);
		unset($children_list[array_search($pid, $children_list)]);//移除进程组中的被kill pid
        $newpid = 0;
//		$newpid = createWorkProcess(1);//创建新子进程
		unlink($pidFile);
		file_put_contents($pidFile,implode(" ", $children_list));
		file_put_contents($worklogFile, date('Y-m-d H:i:s')."##work is restart (out {$pid} join {$newpid})\n",FILE_APPEND | LOCK_EX);
    }
}