简单多进程任务处理程序

<?php /** * 多进程任务处理辅助类 */ class TaskHelper{ /** * worker进程最大数量, 至少两个 */ protected $maxProcess; /** * 动态参数，设置为 0 表示不使用自适应进程方式 */ protected $dynamicParam; /** * 任务的实际处理者，对象, 必须有 runWorker 方法 */ protected $worker; public function __construct($worker, $maxProcess = 4, $dynamicParam = 0) { $this->worker = $worker;
        $this->maxProcess = max(2, (int)$maxProcess);
        $this->dynamicParam = max(0, (int)$dynamicParam);
    }
    
    /**
     * fork子进程处理数据
     * @param Array $data 需要处理的数据，必须是数组
     */
    public function run(&$data) {
        $count = count($data);
        
        // 需要开启的子进程数
        $num = $this->dynamicParam ? min( $this->maxProcess, ceil($count / $this->dynamicParam) ) : $this->maxProcess;
        
        // 每个进程处理的数据量
        $n = ceil($count / $num); 
        
        $childs = array();
        
        for($i = 0; $i < $count; $i += $n) { $pid = pcntl_fork(); if($pid == -1) { echo "Fork worker failed!"; return false; } if($pid) { echo "Fork worker success! pid:", $pid, "\n"; $childs[] = $pid; } else { $sliceData = array_slice($data, $i, $n); $this->worker->runWorker($sliceData);
                exit();
            }
        }
        
        $this->check($childs);
    }
    
    /**
     * 检测子进程状态，监控子进程是否退出，并防止僵尸进程
     */
    protected function check($childs) {
        while(true) {
            foreach($childs as $index => $pid) {
                $pid && $res = pcntl_waitpid($pid, $status, WNOHANG);
                if(!$pid || $res == -1) {
                    echo "End worker: $pid \n";
                    unset($childs[$index]);
                }
            }
            
            if(empty($childs)) break;
            sleep(1);
        }
    }
}

/**
 * 使用示例
 */
class Test {
    public function run() {
        $data = array_fill(0, 800, 1);
        
        // 开8个进程将 $data 分成8份，交由下面的 runWorker 方法处理
        $task = new TaskHelper($this, 8);
        
        $task->run($data); // 如果前面连接了数据库、redis等，最好在这之前关闭掉
    }
    
    /**
     * 这里编写代相应码来处理数据
     */
    public function runWorker($data) {
        // do something...
    }
}

$obj = new Test;
$obj->run();