PHP はスーパーバイザーのプロセス管理をシミュレートします
- 藏色散人転載
- 2021-01-08 13:53:434053ブラウズ
推奨: 「PHP ビデオ チュートリアル 」
序文
スーパーバイザー プロセス管理のシミュレーション デモ (簡単な実装)
そう、ホイールづくりです！目的は学ぶことです！
スクリーンショット: 
画像の
Copyサブプロセスの関数を実装しました。 AMQP でコンシューマを追加または削除するときに使用すると非常に便利だと思います。
実装
1. メイン プロセス ループでサブプロセスを開始して、コマンド
2. に 127.0.0.1:7865 を入力します。 Web からサブプロセスのステータスを取得します
3. ソケットはリクエスト メッセージを受信し、対応する操作を実行し、Web ページに戻ります
4. 子プロセスがゾンビと呼ばれることを防ぐためにリサイクルします
不十分: エラー ページの監視を続行できません。ソケットからの応答は include 関数を通じてロードされるため、ロードされたページに tail -f コマンドを含めることはできません。そうしないと、ストリームが無限ループに陥ってしまいます。解決策があるはずだと思います（fpmがリクエストを受信した後にサブプロセスを開始して処理するモードを模倣するためにソケットマルチプロセスモードを作成しましたが、実行に問題があります。そこでコードを投稿しました）皆さんのアドバイスが得られることを願っています）。
拡張: プロセスは適切に管理できる (予想される) ため、次のような独自のニーズをカスタマイズできます。 (1) AMQP コンシューマー プロセス管理サービスをカスタマイズします。 (2) crontab タイミング サービスをシミュレートします。
知識ポイント
コード実装のプロセスでは、学ぶ価値のある詳細がたくさんあります。
1. while() ループでは、ストリームのノンブロッキング モードが有効になります。したがって、ループ内で sleep(1) を使用することはできませんが、stream_select($read, $write, $excel, 1) を使用してストリームを内部的にブロックします。
ブロッキング モードとノンブロッキング モードについては、こちらを参照してください
2. 外部プログラムを実行できる関数は多数ありますが、どれも若干異なります。ここでは proc_open が使用されています。これは非常に強力な関数です。これまでは、pcntl_exec を使用して外部プログラムを実行していましたが、最初に pcntl_fork を使用する必要がありました。ただし、exec や shell_exec などの他のメソッドは子プロセスを管理できません。
3. 子プロセスを再起動または停止する場合、最初にメイン プロセスのメモリ内で子プロセスの状態が変更されるだけで、子プロセスに対して実際の操作は行われません。同じ場所で init() が子プロセスを処理します。これにより、子プロセスの開始時にコンテキストによって引き起こされる奇妙な現象を防ぐことができます。
コード
コードが多すぎるため、私のソリューションに対するより良い提案がある場合は、github で確認できます。
メインプロセスコード: Process.php
<?php require_once __DIR__ . '/Consumer.php';require_once __DIR__ . '/StreamConnection.php';require_once __DIR__ . '/Http.php';class Process{
/**
* 待启动的消费者数组
*/
protected $consumers = array();
protected $childPids = array();
const PPID_FILE = __DIR__ . '/process';
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();
コンシューマコンシューマオブジェクト
<?php require_once __DIR__ . '/BaseObject.php';class Consumer extends BaseObject{
/** 开启多少个消费者 */
public $numprocs = 1;
/** 当前配置的唯一标志 */
public $program;
/** 执行的命令 */
public $command;
/** 当前工作的目录 */
public $directory;
/** 通过 $qos $queueName $duplicate 生成的 $queue */
public $queue;
/** 程序执行日志记录 */
public $logfile = '';
/** 消费进程的唯一ID */
public $uniqid;
/** 进程IDpid */
public $pid;
/** 进程状态 */
public $state = self::NOMINAL;
/** 自启动 */
public $auto_restart = false;
public $process;
/** 启动时间 */
public $uptime;
const RUNNING = 'running';
const STOP = 'stoped';
const NOMINAL = 'nominal';
const RESTART = 'restart';
const STOPING = 'stoping';
const STARTING = 'stating';
const ERROR = 'error';
const BLOCKED = 'blocked';
const EXITED = 'exited';
const FATEL = 'fatel';}ストリーム関連コード: 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__ . '/views';
public static $max_age = 120; //秒
/*
* 函数: parse_http
* 描述: 解析http协议
*/
public static function parse_http($http)
{
// 初始化
$_POST = $_GET = $_COOKIE = $_REQUEST = $_SESSION = $_FILES = array();
$GLOBALS['HTTP_RAW_POST_DATA'] = '';
// 需要设置的变量名
$_SERVER = array(
'QUERY_STRING' => '',
'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__ . '/test.log', date('Y-m-d H:i:s'));
sleep(1);}現在のディレクトリのページを表示:
|- Process.php
|- Http.php
|- StreamConnection.php
|- Consumer.php
|- BaseObject.php
|- views/
プログラミング関連の知識の詳細については、次を参照してください: プログラミング教えます! ！
以上がPHP はスーパーバイザーのプロセス管理をシミュレートしますの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。
声明：
この記事はlearnku.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。
前の記事：PHP で配列を定義する 5 つの方法次の記事：PHP で配列を定義する 5 つの方法