搜尋

首頁 >後端開發 >php教程 >PHP模擬supervisor的流程管理

PHP模擬supervisor的流程管理

藏色散人
藏色散人轉載
2021-01-08 13:53:434109瀏覽

推薦：《PHP影片教學》

#前言

模擬supervisor進程管理DEMO（簡易實作）

沒錯，是造輪子！目的在於學習！

截圖：
PHP模擬supervisor的流程管理

在圖中自己實作了一個Copy子程序的功能。如果用在AMQP增減消費者時，我覺得應該會很有用。

實作

1、在主行程循環內啟動子程序執行指令
2、在web輸入127.0.0.1:7865 取得子程序狀態
3、socket接收請求訊息，並且執行對應操作，返回web頁面
4、回收子進程，防止稱為殭屍進程

#不足：無法持續監聽錯誤頁面。由於socket得到的回應是透過include函數載入的，所以在載入的頁面內不能出現tail -f指令，否則stream就會掉入了死循環了~。我想應該有方案解決（寫了socket 多進程模式，模仿fpm在接收到請求之後就啟動一個子進程去處理的模式，但是執行有問題。因此將程式碼貼出來希望得到大家的指點）。
延伸：由於對流程可以很好的管理（期望如此），那麼就可以客製化自己的一些需求，例如：（1）客製化AMQP的消費者進程管理服務。 （2）模擬crontab定時服務。

知識點

程式碼實作的過程中，有很多的細節是值得學習的。
1、在while()迴圈中，啟用了stream的非阻塞模式。所以不能在迴圈中使用sleep(1)，而是用stream_select($read, $write, $except, 1)讓stream內部阻塞。
關於阻塞非阻塞模式，可以參考這裡
2、能夠執行外部程式的函數很多，但是都稍有不同。這裡採用的是proc_open，是一個很強大的函數。在這之前我曾用pcntl_exec執行過外部程序，但是需要先pcntl_fork。而用其他的如exec,shell_exec無法對子程序進行管理。
3、重啟或停止等操作子進程時，只是先更改主進程中該子進程在記憶體中的狀態，並不是真正的對子進程操作。在統一處init()處理子程序。如此才能防止因為子行程啟動時的上下文所導致的一些怪異的現象。

程式碼

由於程式碼過多，所以如果你對我的方案有更好的建議可以在github這裡看。

主進程代碼：Process.php

<?php require_once __DIR__ . &#39;/Consumer.php&#39;;require_once __DIR__ . &#39;/StreamConnection.php&#39;;require_once __DIR__ . &#39;/Http.php&#39;;class Process{
    /** 
     * 待启动的消费者数组
     */
    protected $consumers = array();
    protected $childPids = array();

    const PPID_FILE = __DIR__ . &#39;/process&#39;;
    protected $serializerConsumer;

    public function __construct()
    {
        $this->consumers = $this->getConsumers();
    }

    // 这里是个DEMO，实际可以用读取配置文件的方式。
    public function getConsumers()
    {
        $consumer = new Consumer([
            'program' => 'test',
            'command' => '/usr/bin/php test.php',
            'directory' => __DIR__,
            'logfile' => __DIR__ . '/test.log',
            'uniqid' => uniqid(),
            'auto_restart' => false,
        ]);
        return [
            $consumer->uniqid => $consumer,
        ];
    }

    public function run()
    {
        if (empty($this->consumers)) {
            // consumer empty
            return;
        }
        if ($this->_notifyMaster()) {
            // master alive
            return;
        }

        $pid = pcntl_fork();
        if ($pid  0) {
            exit;
        }
        if (!posix_setsid()) {
            exit;
        }

        $stream = new StreamConnection('tcp://0.0.0.0:7865');
        @cli_set_process_title('AMQP Master Process');
        // 将主进程ID写入文件
        file_put_contents(self::PPID_FILE, getmypid());
        // master进程继续
        while (true) {
            $this->init();
            pcntl_signal_dispatch();
            $this->waitpid();
            // 如果子进程被全部回收，则主进程退出
            // if (empty($this->childPids)) {
            //     $stream->close($stream->getSocket());
            //     break;
            // }
            $stream->accept(function ($uniqid, $action) {
                $this->handle($uniqid, $action);
                return $this->display();
            });
        }
    }

    protected function init()
    {
        foreach ($this->consumers as &$c) {
            switch ($c->state) {
                case Consumer::RUNNING:
                case Consumer::STOP:
                    break;
                case Consumer::NOMINAL:
                case Consumer::STARTING:
                    $this->fork($c);
                    break;
                case Consumer::STOPING:
                    if ($c->pid && posix_kill($c->pid, SIGTERM)) {
                        $this->reset($c, Consumer::STOP);
                    }
                    break;
                case Consumer::RESTART:
                    if (empty($c->pid)) {
                        $this->fork($c);
                        break;
                    }
                    if (posix_kill($c->pid, SIGTERM)) {
                        $this->reset($c, Consumer::STOP);
                        $this->fork($c);
                    }
                    break;
                default:
                    break;
            }
        }
    }

    protected function reset(Consumer $c, $state)
    {
        $c->pid = '';
        $c->uptime = '';
        $c->state = $state;
        $c->process = null;
    }

    protected function waitpid()
    {
        foreach ($this->childPids as $uniqid => $pid) {
            $result = pcntl_waitpid($pid, $status, WNOHANG);
            if ($result == $pid || $result == -1) {
                unset($this->childPids[$uniqid]);
                $c = &$this->consumers[$uniqid];
                $state = pcntl_wifexited($status) ? Consumer::EXITED : Consumer::STOP;
                $this->reset($c, $state);
            }
        }
    }


    /**
     * 父进程存活情况下，只会通知父进程信息，否则可能产生多个守护进程
     */
    private function _notifyMaster()
    {
        $ppid = file_get_contents(self::PPID_FILE );
        $isAlive = $this->checkProcessAlive($ppid);
        if (!$isAlive) return false;
        return true;
    }

    public function checkProcessAlive($pid)
    {
        if (empty($pid)) return false;
        $pidinfo = `ps co pid {$pid} | xargs`;
        $pidinfo = trim($pidinfo);
        $pattern = "/.*?PID.*?(\d+).*?/";
        preg_match($pattern, $pidinfo, $matches);
        return empty($matches) ? false : ($matches[1] == $pid ? true : false);
    }

    /**
     * fork一个新的子进程
     */
    protected function fork(Consumer $c)
    {
        $descriptorspec = [2 => ['file', $c->logfile, 'a'],];
        $process = proc_open('exec ' . $c->command, $descriptorspec, $pipes, $c->directory);
        if ($process) {
            $ret = proc_get_status($process);
            if ($ret['running']) {
                $c->state = Consumer::RUNNING;
                $c->pid = $ret['pid'];
                $c->process = $process;
                $c->uptime = date('m-d H:i');
                $this->childPids[$c->uniqid] = $ret['pid'];
            } else {
                $c->state = Consumer::EXITED;
                proc_close($process);
            }
        } else {
            $c->state = Consumer::ERROR;
        }
        return $c;
    }

    public function display()
    {
        $location = 'http://127.0.0.1:7865';
        $basePath = Http::$basePath;
        $scriptName = isset($_SERVER['SCRIPT_NAME']) &&
            !empty($_SERVER['SCRIPT_NAME']) &&
            $_SERVER['SCRIPT_NAME'] != '/' ? $_SERVER['SCRIPT_NAME'] : '/index.php';
        if ($scriptName == '/index.html') {
            return Http::status_301($location);
        }

        $sourcePath = $basePath . $scriptName;
        if (!is_file($sourcePath)) {
            return Http::status_404();
        }

        ob_start();
        include $sourcePath;
        $response = ob_get_contents();
        ob_clean();

        return Http::status_200($response);
    }

    public function handle($uniqid, $action)
    {
        if (!empty($uniqid) && !isset($this->consumers[$uniqid])) {
            return;
        }
        switch ($action) {
            case 'refresh':
                break;
            case 'restartall':
                $this->killall(true);
                break;
            case 'stopall':
                $this->killall();
                break;
            case 'stop':
                $c = &$this->consumers[$uniqid];
                if ($c->state != Consumer::RUNNING) break;
                $c->state = Consumer::STOPING;
                break;
            case 'start':
                $c = &$this->consumers[$uniqid];
                if ($c->state == Consumer::RUNNING) break;
                $c->state = Consumer::STARTING;
                break;
            case 'restart':
                $c = &$this->consumers[$uniqid];
                $c->state = Consumer::RESTART;
                break;
            case 'copy':
                $c = $this->consumers[$uniqid];
                $newC = clone $c;
                $newC->uniqid = uniqid('C');
                $newC->state = Consumer::NOMINAL;
                $newC->pid = '';
                $this->consumers[$newC->uniqid] = $newC;
                break;
            default:
                break;
        }
    }

    protected function killall($restart = false)
    {
        foreach ($this->consumers as &$c) {
            $c->state = $restart ? Consumer::RESTART : Consumer::STOPING;
        }
    }}$cli = new Process();$cli->run();

Consumer消費者物件

<?php require_once __DIR__ . &#39;/BaseObject.php&#39;;class Consumer extends BaseObject{
    /** 开启多少个消费者 */
    public $numprocs = 1;
    /** 当前配置的唯一标志 */
    public $program;
    /** 执行的命令 */
    public $command;
    /** 当前工作的目录 */
    public $directory;

    /** 通过 $qos $queueName $duplicate 生成的 $queue */
    public $queue;
    /** 程序执行日志记录 */
    public $logfile = &#39;&#39;;
    /** 消费进程的唯一ID */
    public $uniqid;
    /** 进程IDpid */
    public $pid;
    /** 进程状态 */
    public $state = self::NOMINAL;
    /** 自启动 */
    public $auto_restart = false;

    public $process;
    /** 启动时间 */
    public $uptime;

    const RUNNING = &#39;running&#39;;
    const STOP = &#39;stoped&#39;;
    const NOMINAL = &#39;nominal&#39;;
    const RESTART = &#39;restart&#39;;
    const STOPING = &#39;stoping&#39;;
    const STARTING = &#39;stating&#39;;
    const ERROR = &#39;error&#39;;
    const BLOCKED = &#39;blocked&#39;;
    const EXITED = &#39;exited&#39;;
    const FATEL = &#39;fatel&#39;;}

stream相關程式碼：StreamConnection.php

<?php class StreamConnection{
    protected $socket;
    protected $timeout = 2; //s
    protected $client;

    public function __construct($host)
    {
        $this->socket = $this->connect($host);
    }

    public function connect($host)
    {
        $socket = stream_socket_server($host, $errno, $errstr);
        if (!$socket) {
            exit('stream error');
        }
        stream_set_timeout($socket, $this->timeout);
        stream_set_chunk_size($socket, 1024);
        stream_set_blocking($socket, false);
        $this->client = [$socket];
        return $socket;
    }

    public function accept(Closure $callback)
    {
        $read = $this->client;
        if (stream_select($read, $write, $except, 1) socket, $read)) {
            $cs = stream_socket_accept($this->socket);
            $this->client[] = $cs;
        }
        foreach ($read as $s) {
            if ($s == $this->socket) continue;
            $header = fread($s, 1024);
            if (empty($header)) {
                $index = array_search($s, $this->client);
                if ($index)
                    unset($this->client[$index]);
                $this->close($s);
                continue;
            }
            Http::parse_http($header);
            $uniqid = isset($_GET['uniqid']) ? $_GET['uniqid'] : '';
            $action = isset($_GET['action']) ? $_GET['action'] : '';
            $response = $callback($uniqid, $action);
            $this->write($s, $response);
            $index = array_search($s, $this->client);
            if ($index)
                unset($this->client[$index]);
            $this->close($s);
        }
    }

    public function write($socket, $response)
    {
        $ret = fwrite($socket, $response, strlen($response));
    }

    public function close($socket)
    {
        $flag = fclose($socket);
    }

    public function getSocket()
    {
        return $this->socket;
    }}

Http回應程式碼：Http.php

<?php class Http{

    public static $basePath = __DIR__ . &#39;/views&#39;;
    public static $max_age = 120; //秒

    /*
    *  函数:     parse_http
    *  描述:     解析http协议
    */
    public static function parse_http($http)
    {
        // 初始化
        $_POST = $_GET = $_COOKIE = $_REQUEST = $_SESSION = $_FILES =  array();
        $GLOBALS[&#39;HTTP_RAW_POST_DATA&#39;] = &#39;&#39;;
        // 需要设置的变量名
        $_SERVER = array(
            &#39;QUERY_STRING&#39; => '',
            'REQUEST_METHOD' => '',
            'REQUEST_URI' => '',
            'SERVER_PROTOCOL' => '',
            'SERVER_SOFTWARE' => '',
            'SERVER_NAME' => '',
            'HTTP_HOST' => '',
            'HTTP_USER_AGENT' => '',
            'HTTP_ACCEPT' => '',
            'HTTP_ACCEPT_LANGUAGE' => '',
            'HTTP_ACCEPT_ENCODING' => '',
            'HTTP_COOKIE' => '',
            'HTTP_CONNECTION' => '',
            'REMOTE_ADDR' => '',
            'REMOTE_PORT' => '0',
            'SCRIPT_NAME' => '',
            'HTTP_REFERER' => '',
            'CONTENT_TYPE' => '',
            'HTTP_IF_NONE_MATCH' => '',
        );

        // 将header分割成数组
        list($http_header, $http_body) = explode("\r\n\r\n", $http, 2);
        $header_data = explode("\r\n", $http_header);

        list($_SERVER['REQUEST_METHOD'], $_SERVER['REQUEST_URI'], $_SERVER['SERVER_PROTOCOL']) = explode(' ', $header_data[0]);

        unset($header_data[0]);
        foreach ($header_data as $content) {
            // \r\n\r\n
            if (empty($content)) {
                continue;
            }
            list($key, $value) = explode(':', $content, 2);
            $key = strtolower($key);
            $value = trim($value);
            switch ($key) {
                case 'host':
                    $_SERVER['HTTP_HOST'] = $value;
                    $tmp = explode(':', $value);
                    $_SERVER['SERVER_NAME'] = $tmp[0];
                    if (isset($tmp[1])) {
                        $_SERVER['SERVER_PORT'] = $tmp[1];
                    }
                    break;
                case 'cookie':
                    $_SERVER['HTTP_COOKIE'] = $value;
                    parse_str(str_replace('; ', '&', $_SERVER['HTTP_COOKIE']), $_COOKIE);
                    break;
                case 'user-agent':
                    $_SERVER['HTTP_USER_AGENT'] = $value;
                    break;
                case 'accept':
                    $_SERVER['HTTP_ACCEPT'] = $value;
                    break;
                case 'accept-language':
                    $_SERVER['HTTP_ACCEPT_LANGUAGE'] = $value;
                    break;
                case 'accept-encoding':
                    $_SERVER['HTTP_ACCEPT_ENCODING'] = $value;
                    break;
                case 'connection':
                    $_SERVER['HTTP_CONNECTION'] = $value;
                    break;
                case 'referer':
                    $_SERVER['HTTP_REFERER'] = $value;
                    break;
                case 'if-modified-since':
                    $_SERVER['HTTP_IF_MODIFIED_SINCE'] = $value;
                    break;
                case 'if-none-match':
                    $_SERVER['HTTP_IF_NONE_MATCH'] = $value;
                    break;
                case 'content-type':
                    if (!preg_match('/boundary="?(\S+)"?/', $value, $match)) {
                        $_SERVER['CONTENT_TYPE'] = $value;
                    } else {
                        $_SERVER['CONTENT_TYPE'] = 'multipart/form-data';
                        $http_post_boundary = '--' . $match[1];
                    }
                    break;
            }
        }

        // script_name
        $_SERVER['SCRIPT_NAME'] = parse_url($_SERVER['REQUEST_URI'], PHP_URL_PATH);

        // QUERY_STRING
        $_SERVER['QUERY_STRING'] = parse_url($_SERVER['REQUEST_URI'], PHP_URL_QUERY);
        if ($_SERVER['QUERY_STRING']) {
            // $GET
            parse_str($_SERVER['QUERY_STRING'], $_GET);
        } else {
            $_SERVER['QUERY_STRING'] = '';
        }

        // REQUEST
        $_REQUEST = array_merge($_GET, $_POST);

        return array('get' => $_GET, 'post' => $_POST, 'cookie' => $_COOKIE, 'server' => $_SERVER, 'files' => $_FILES);
    }

    public static function status_404()
    {
        return <p>待執行的腳本：test.php</p><pre class="brush:php;toolbar:false"><?php while(true) {
    file_put_contents(__DIR__  .  &#39;/test.log&#39;, date(&#39;Y-m-d H:i:s&#39;));
    sleep(1);}

在目前目錄下的檢視頁面：
|- Process.php
|- Http. php
|- StreamConnection.php
|- Consumer.php
|- BaseObject.php
|- views/

更多程式設計相關知識，請造訪：程式教學！ ！

以上是PHP模擬supervisor的流程管理的詳細內容。更多資訊請關注PHP中文網其他相關文章！

php while include 循环 copy 对象 github http
陳述：
本文轉載於：learnku.com。如有侵權，請聯絡admin@php.cn刪除
上一篇：5種PHP定義數組的方法下一篇：5種PHP定義數組的方法

相關文章

看更多