qianniaoQueue/Jobs/Model/MProcess.Class.php

<?php
namespace Jobs\Model;

use Mall\Framework\Factory;

use Jobs\Model\MJobs;

class MProcess
{
    const CHILD_PROCESS_CAN_RESTART                    ='staticWorker'; //子进程可以重启,进程个数固定
    const CHILD_PROCESS_CAN_NOT_RESTART                ='dynamicWorker'; //子进程不可以重启，进程个数根据队列堵塞情况动态分配
    const STATUS_RUNNING                               ='runnning'; //主进程running状态
    const STATUS_WAIT                                  ='wait'; //主进程wait状态
    const STATUS_STOP                                  ='stop'; //主进程stop状态
    const APP_NAME                                     ='jobs'; //app name
    const STATUS_HSET_KEY_HASH                         ='status'; //status hash名

    public $processName                   = ':swooleProcessTopicQueueJob'; // 进程重命名, 方便 shell 脚本管理
    public $workers                       = [];

    private $version                      = '2.5';
    private $excuteTime                   =3600; //子进程最长执行时间,单位：秒
    private $queueMaxNum                  =10; //队列达到一定长度，启动动态子进程个数发和送消息提醒
    private $queueTickTimer               =1000 * 10; //一定时间间隔（毫秒）检查队列长度;默认10秒钟
    private $messageTickTimer             =1000 * 180; //一定时间间隔（毫秒）发送消息提醒;默认3分钟
    private $message                      =[]; //提醒消息内容
    private $workerNum                    =0; //固定分配的子进程个数
    private $dynamicWorkerNum             =[]; //动态（不能重启）子进程计数，最大数为每个topic配置workerMaxNum，它的个数是动态变化的
    private $workersInfo                  =[];
    private $ppid;
    private $config                       = [];
    private $pidFile                      = 'master.pid'; //pid存放文件
    private $pidInfoFile                  = 'master.info'; //pid 序列化信息
    private $pidStatusFile                = 'status.info'; //pid status信息
    private $status                       = '';
    private $logger                       = null;
    private $queue                        = null;
    private $topics                       = null;
    private $beginTime                    = '';
    private $logSaveFileWorker            = 'workers.log';

    public function __construct($config)
    {
        $this->config                    = $config;
        $this->logger                    = Factory::logs($this->config['logPath'] ?:'', $this->config['logSaveFileApp'] ?:'', $this->config['app_name'] ?:'');
        $this->topics                    = $this->config['job']['topics'] ?: [];
        $this->processName               = $this->config['processName'] ?: $this->processName;
        $this->excuteTime                = $this->config['excuteTime'] ?: $this->excuteTime;
        $this->queueMaxNum               = $this->config['queueMaxNum'] ?: $this->queueMaxNum;
        $this->queueTickTimer            = $this->config['queueTickTimer'] ?: $this->queueTickTimer;
        $this->messageTickTimer          = $this->config['messageTickTimer'] ?: $this->messageTickTimer;
        $this->logSaveFileWorker         = $this->config['logSaveFileWorker'] ?: $this->logSaveFileWorker;

        $this->beginTime=time();
        //该变量需要在多进程共享
        $this->status=self::STATUS_RUNNING;

        if (isset($this->config['pidPath']) && !empty($this->config['pidPath'])) {
            if (!is_dir($this->config['pidPath'])) {
                mkdir($this->config['pidPath'], '0777', true);
            }
            $this->pidFile      =$this->config['pidPath'] . '/' . $this->pidFile;
            $this->pidInfoFile  =$this->config['pidPath'] . '/' . $this->pidInfoFile;
            $this->pidStatusFile=$this->config['pidPath'] . '/' . $this->pidStatusFile;
        } else {
            die('config pidPath must be set!' . PHP_EOL);
        }

        /*
         * master.pid 文件记录 master 进程 pid, 方便之后进程管理
         * 请管理好此文件位置, 使用 systemd 管理进程时会用到此文件
         * 判断文件是否存在，并判断进程是否在运行
         */
        if (file_exists($this->pidFile)) {
            $pid=$this->getMasterData('pid');

            // $signo=0，可以检测进程是否存在，不会发送默认SIGTERM信号,终止进程
            if ($pid && @\Swoole\Process::kill($pid, 0)) {
                die('已有进程运行中,请先结束或重启' . PHP_EOL);
            }
        }

        // 使当前进程蜕变为一个守护进程。
        \Swoole\Process::daemon();
        $this->ppid    = getmypid();
        $data['pid']   =$this->ppid;
        $data['status']=$this->status;
        $this->saveMasterData($data);
        $this->setProcessName(self::APP_NAME . ' master ' . $this->ppid . $this->processName);
        //报错解决:must be forked outside the coroutine
        swoole_async_set(['enable_coroutine' => false]);//关闭内置协程
    }

    /**
     * 设置进程名
     * @param mixed $name
     */
    private function setProcessName($name)
    {
        $updateName = 0;
        // 低版本Linux内核和Mac OSX不支持进程重命名
        if (function_exists('cli_set_process_title') && PHP_OS != 'Darwin'){
            if(cli_set_process_title($name)){ // 按照官网文档优先使用内置函数，php >= 5.5
                $updateName = 1;
            }
        }else if (function_exists('swoole_set_process_name') && PHP_OS != 'Darwin' && !$updateName) {
            swoole_set_process_name($name); // swoole >= 1.6.3
        }
    }

    /**
     * 保存守护进程数据
     * @param array $data
     */
    private function saveMasterData($data=[])
    {
        isset($data['pid']) && file_put_contents($this->pidFile, $data['pid']);
        file_put_contents($this->pidInfoFile, serialize($data));
    }

    /**
     * 读取守护进程数据
     * @param string $key
     * @return mixed|null
     */
    private function getMasterData($key='')
    {
        $data=unserialize(file_get_contents($this->pidInfoFile));
        if ($key) {
            return $data[$key] ?: null;
        }

        return $data;
    }

    /**
     * 按照配置开启每个topic进程
     */
    public function start()
    {
        $topics = $this->topics;
        $this->logger->log('topics: ' . json_encode($topics));

        if ($topics) {
            //遍历topic任务列表
            foreach ((array) $topics as  $topic) {
                if (isset($topic['workerMinNum']) && isset($topic['name'])) {
                    //每个topic开启最少个进程消费队列
                    for ($i = 0; $i < $topic['workerMinNum']; ++$i) {
                        $this->reserveQueue($i, $topic['name'], self::CHILD_PROCESS_CAN_RESTART);
                    }
                }
            }
        }

        $this->registSignal();
        $this->registTimer();
    }

    //增加定时器，检查队列积压情况；
    public function registTimer()
    {
        \Swoole\Timer::tick($this->queueTickTimer, function ($timerId) {
            $topics = $this->topics;
            $this->status  =$this->getMasterData('status');
            if (empty($this->workers) && self::STATUS_WAIT == $this->status) {
                $this->exitMaster();
            }
            $this->queue   = MQueue::getQueue($this->config['job']['queue'], $this->logger);
            if (empty($this->queue)) {
                $this->logger->log('queue connection is lost', 'info', $this->logSaveFileWorker);

                return;
            }
            $this->queue->setTopics($topics);

            if ($topics && self::STATUS_RUNNING == $this->status) {
                //遍历topic任务列表
                foreach ((array) $topics as  $topic) {
                    if (empty($topic['name'])) {
                        continue;
                    }
                    $this->dynamicWorkerNum[$topic['name']]=$this->dynamicWorkerNum[$topic['name']] ?? 0;
                    $topic['workerMaxNum']                       =$topic['workerMaxNum'] ?? 0;
                    try {
                        $len=$this->queue->len($topic['name']);
                        $this->logger->log('topic: ' . $topic['name'] . ' ' . $this->status . ' len: ' . $len, 'info', $this->logSaveFileWorker);
                    } catch (\Throwable $e) {
                        $this->logger->log('queueError' . $e->getMessage(), 'error', $this->logSaveFileWorker);
                    } catch (\Exception $e) {
                        $this->logger->log('queueError: ' . $e->getMessage(), 'error', $this->logSaveFileWorker);
                    }
                    $this->status=$this->getMasterData('status');
                    //消息提醒：消息体收集
                    if ($len > $this->queueMaxNum && count($this->message) <= count($topics)) {
                        $this->message[]= strtr('Time:{time} Pid:{pid} ProName:{pname} Topic:{topic} Message:{message}', [
                            '{time}'   => date('Y-m-d H:i:s'),
                            '{pid}'    => $this->ppid,
                            '{pname}'  => $this->processName,
                            '{topic}'  => $topic['name'],
                            '{message}'=> '积压消息个数:' . $len . PHP_EOL,
                        ]);
                    }

                    if ($topic['workerMaxNum'] > $topic['workerMinNum'] && self::STATUS_RUNNING == $this->status && $len > $this->queueMaxNum && $this->dynamicWorkerNum[$topic['name']] < $topic['workerMaxNum']) {
                        $max=$topic['workerMaxNum'] - $this->dynamicWorkerNum[$topic['name']];
                        for ($i=0; $i < $max; ++$i) {
                            //队列堆积达到一定数据，拉起一次性子进程,这类进程不会自动重启[没必要]
                            $this->reserveQueue($this->dynamicWorkerNum[$topic['name']], $topic['name'], self::CHILD_PROCESS_CAN_NOT_RESTART);
                            ++$this->dynamicWorkerNum[$topic['name']];
                            $this->logger->log('topic: ' . $topic['name'] . ' ' . $this->status . ' len: ' . $len . ' for: ' . $i . ' ' . $max, 'info', $this->logSaveFileWorker);
                        }
                    }
                }
            }
            $this->queue->close();
        });
        //积压队列提醒
        \Swoole\Timer::tick($this->messageTickTimer, function ($timerId) {
            !empty($this->message) && $this->logger->log('Warning Message: ' . implode('', $this->message), 'warning', $this->logSaveFileWorker);
            if ($this->message && isset($this->config['message'])) {
                //$message =Message::getMessage($this->config['message']);
                //$message->send(implode('', $this->message), $this->config['message']['token']);
            }
            //重置message，防止message不断变长
            $this->message=[];
        });
    }

    //退出主进程
    private function exitMaster()
    {
        @unlink($this->pidFile);
        @unlink($this->pidInfoFile);
        $this->logger->log('Time: ' . microtime(true) . '主进程' . $this->ppid . '退出', 'info', $this->logSaveFileWorker);

        sleep(1);
        exit();
    }

    /**
     * 注册信号
     */
    public function registSignal()
    {
        //终止进程信号
        \Swoole\Process::signal(SIGTERM, function ($signo) {
            $this->killWorkersAndExitMaster();
        });
        //
        \Swoole\Process::signal(SIGKILL, function ($signo) {
            $this->killWorkersAndExitMaster();
        });
        //平滑退出
        \Swoole\Process::signal(SIGUSR1, function ($signo) {
            $this->waitWorkers();
        });
        //记录进程状态
        \Swoole\Process::signal(SIGUSR2, function ($signo) {
            $this->logger->log('[master pid: ' . $this->ppid . '] has been received  signal' . $signo);
            $result=$this->showStatus();
            $this->saveSwooleJobsStatus($result);
            //echo $result;
        });

        // 在一个进程终止或者停止时，将SIGCHLD信号发送给其父进程
        \Swoole\Process::signal(SIGCHLD, function ($signo) {
            while (true) {
                try {
                    $ret = \Swoole\Process::wait(false);
                } catch (\Exception $e) {
                    $this->logger->log('signoError: ' . $signo . $e->getMessage(), 'error', $this->logSaveFileWorker);
                }
                if ($ret) {
                    $pid           = $ret['pid'];
                    $childProcess = $this->workers[$pid];
                    $topic = $this->workersInfo[$pid]['topic'] ?: '';
                    $this->status=$this->getMasterData('status');
                    if(empty($this->dynamicWorkerNum) || empty($this->dynamicWorkerNum[$topic])){
                        $topicCanNotRestartNum = null;
                    }else{
                        $topicCanNotRestartNum =  $this->dynamicWorkerNum[$topic];
                    }
                    // $topicCanNotRestartNum =  $this->dynamicWorkerNum[$topic] ?: 'null';
                    $this->logger->log(self::CHILD_PROCESS_CAN_RESTART . '---' . $topic . '***' . $topicCanNotRestartNum . '***' . $this->status . '***' . $this->workersInfo[$pid]['type'] . '***' . $pid, 'info', $this->logSaveFileWorker);
                    $this->logger->log($pid . ',' . $this->status . ',' . self::STATUS_RUNNING . ',' . $this->workersInfo[$pid]['type'] . ',' . self::CHILD_PROCESS_CAN_RESTART, 'info', $this->logSaveFileWorker);

                    //主进程状态为running并且该子进程是可以重启的
                    if (self::STATUS_RUNNING == $this->status && $this->workersInfo[$pid]['type'] == self::CHILD_PROCESS_CAN_RESTART) {
                        try {
                            //子进程重启可能失败，必须启动成功之后，再往下执行;最多尝试30次
                            for ($i=0; $i < 30; ++$i) {
                                var_dump("aaaa");
                                $newPid = $childProcess->start();
                                var_dump("xxxxxx".$newPid);
                                if ($newPid > 0) {
                                    break;
                                }
                                sleep(1);
                            }
                            if (!$newPid) {
                                $this->logger->log('静态子进程重启失败，问题有点严重，平滑退出子进程，主进程会跟着退出', 'error', $this->logSaveFileWorker);
                                $this->waitWorkers();
                                //$this->reserveQueue(0, $topic);
                                continue;
                            }

                            $this->workers[$newPid] = $childProcess;
                            $this->workersInfo[$newPid]['type'] = self::CHILD_PROCESS_CAN_RESTART;
                            $this->workersInfo[$newPid]['topic'] = $topic;
                            ++$this->workerNum;
                            $this->logger->log("Worker Restart, kill_signal={$ret['signal']} PID=" . $newPid, 'info', $this->logSaveFileWorker);
                        } catch (\Throwable $e) {
                            $this->logger->log('restartErrorThrow' . $e->getMessage(), 'error', $this->logSaveFileWorker);
                        } catch (\Exception $e) {
                            $this->logger->log('restartError: ' . $e->getMessage(), 'error', $this->logSaveFileWorker);
                        }
                    }
                    //某个topic动态变化的子进程，退出之后个数减少一个
                    if ($this->workersInfo[$pid]['type'] == self::CHILD_PROCESS_CAN_NOT_RESTART) {
                        --$this->dynamicWorkerNum[$topic];
                    }
                    $this->logger->log("Worker Exit, kill_signal={$ret['signal']} PID=" . $pid, 'info', $this->logSaveFileWorker);
                    unset($this->workers[$pid], $this->workersInfo[$pid]);
                    --$this->workerNum;

                    $this->logger->log('Worker count: ' . count($this->workers) . '==' . $this->workerNum, 'info', $this->logSaveFileWorker);
                    //如果$this->workers为空，且主进程状态为wait，说明所有子进程安全退出，这个时候主进程退出
                    if (empty($this->workers) && self::STATUS_WAIT == $this->status) {
                        $this->logger->log('主进程收到所有信号子进程的退出信号，子进程安全退出完成', 'info', $this->logSaveFileWorker);
                        $this->exitMaster();
                    }
                } else {
                    break;
                }
            }
        });
    }

    private function saveSwooleJobsStatus($data)
    {
        file_put_contents($this->pidStatusFile, $data);
    }

    private function showStatus()
    {
        $statusStr  ='-------------------------------------' . $this->processName . ' status--------------------------------------------' . PHP_EOL;
        $statusStr .= 'PHP version:' . PHP_VERSION . '      swoole-jobs version: ' . $this->version . PHP_EOL;
        $statusStr .= 'start time : ' . date('Y-m-d H:i:s', $this->beginTime) . '   run ' . floor((time() - $this->beginTime) / (24 * 60 * 60)) . ' days ' . floor(((time() - $this->beginTime) % (24 * 60 * 60)) / (60 * 60)) . ' hours   ' . PHP_EOL;
        //$statusStr .= Utils::getSysLoadAvg() . '   memory use:' . Utils::getServerMemoryUsage() . PHP_EOL;
        $statusStr .= '|-- Master pid ' . $this->ppid . ' status: ' . $this->status . ' Worker num: ' . count($this->workers) . PHP_EOL;
        if ($this->workers) {
            foreach ($this->workers as $pid => $value) {
                $type =$this->workersInfo[$pid]['type'];
                $topic=$this->workersInfo[$pid]['topic'];

                $statusStr .= '    |---- Worker pid:  ' . $pid . ' ' . $type . ' ' . $topic . PHP_EOL;
            }
        }

        return $statusStr;
    }

    //平滑等待子进程退出之后，再退出主进程
    private function waitWorkers()
    {
        $data['pid']   =$this->ppid;
        $data['status']=self::STATUS_WAIT;
        $this->saveMasterData($data);
        $this->status = self::STATUS_WAIT;
        $this->logger->log('master status: ' . $this->status, 'info', $this->logSaveFileWorker);
    }

    //强制杀死子进程并退出主进程
    private function killWorkersAndExitMaster()
    {
        //修改主进程状态为stop
        $this->status   =self::STATUS_STOP;
        if ($this->workers) {
            foreach ($this->workers as $pid => $worker) {
                //强制杀workers子进程
                \Swoole\Process::kill($pid);
                unset($this->workers[$pid]);
                $this->logger->log('主进程收到退出信号,[' . $pid . ']子进程跟着退出', 'info', $this->logSaveFileWorker);
                $this->logger->log('Worker count: ' . count($this->workers), 'info', $this->logSaveFileWorker);
            }
        }
        $this->exitMaster();
    }

    /**
     * fork子进程消费队列.
     *
     * @param [type] $num   子进程编号
     * @param [type] $topic topic名称
     * @param string $type  是否会重启 canRestart|unRestart
     */
    public function reserveQueue($num, $topic, $type=self::CHILD_PROCESS_CAN_RESTART)
    {
        $reserveProcess = new \Swoole\Process(function ($worker) use ($num, $topic, $type) {
            $this->checkMpid($worker);
            $beginTime=microtime(true);
            try {
                //设置进程名字
                $this->setProcessName($type . ' ' . $topic . ' job ' . $num . ' ' . $this->processName);
                $jobs  = new MJobs($this->pidInfoFile, $this->config);
                do {
                    $jobs->run($topic);
                    $this->status=$this->getMasterData('status');
                    $where = (self::STATUS_RUNNING == $this->status) && (self::CHILD_PROCESS_CAN_RESTART == $type ? time() < ($beginTime + $this->excuteTime) : false);
                } while ($where);
            } catch (\Throwable $e) {
                catchError($this->logger, $e);
            } catch (\Exception $e) {
                catchError($this->logger, $e);
            }

            $endTime=microtime(true);
            $this->logger->log($topic . ' worker id: ' . $num . ' is done!!! Timing: ' . ($endTime - $beginTime), 'info', $this->logSaveFileWorker);
            unset($num, $topic, $type);
        });
        $pid                                        = $reserveProcess->start();
        $this->workers[$pid]                        = $reserveProcess;
        $this->workersInfo[$pid]['type']            = $type;
        $this->workersInfo[$pid]['topic']           = $topic;
        $this->logger->log('topic: ' . $topic . ' ' . $type . ' worker id: ' . $num . ' pid: ' . $pid . ' is start...', 'info', $this->logSaveFileWorker);
        ++$this->workerNum;
    }

    /**
     * 主进程如果不存在了，子进程退出
     **/
    private function checkMpid(&$worker)
    {
        if (!@\Swoole\Process::kill($this->ppid, 0)) {
            $worker->exit();
            $this->logger->log("Master process exited, I [{$worker['pid']}] also quit", 'info', $this->logSaveFileWorker);
        }
    }

}