推薦(免費):swoole
#swoole 的進程之間有兩種通訊方式,一種是訊息佇列(queue),另一種是管道(pipe),對swoole_process
的研究在swoole中顯得尤為重要。
預備知識
IO多路復用
swoole 中的io多路復用表現為底層的 epoll進程模型,在C語言中表現為 epoll 函數。
epoll 模型下會持續監聽自己名下的素有socket 描述子fd
當觸發了socket 監聽的事件時,epoll 函數才會回應,並傳回所有監聽該時間的socket 集合
epoll 的本質是阻塞IO,它的優點在於能同事處理大量socket連接
Event loop 事件循環
swoole 對epoll 實作了一個Reactor線程模型封裝,設定了read事件和write事件的監聽回調函數。 (詳見swoole_event_add)
Event loop 是一個Reactor線程,其中運行了一個epoll實例。
透過swoole_event_add將socket描述符的事件加入epoll監聽中,事件發生時將執行回呼函數
不可用於fpm環境下,因為fpm在任務結束時可能會關掉進程。
swoole_process
基於C語言封裝的進程管理模組,方便php來呼叫
內建管道、訊息佇列接口,方便實現進程間通訊
我們在php-fpm. conf設定檔中發現,php-fpm中有兩種行程池管理設定。
靜態模式 即初始化固定的進程數,當來了一個請求時,從中選取一個進程來處理。
動態模式指定最小、最大進程數,當請求量過大,進程數不超過最大限制時,新增線程去處理請求
接下來用swoole程式碼來實現,這裡只是為理解swoole_process、進程間通訊、定時器等使用,實際情況使用封裝好的swoole_server來實作task任務佇列池會比較方便。
假如有個定時投遞的任務佇列:
<?php/** * 动态进程池,类似fpm * 动态新建进程 * 有初始进程数,最小进程数,进程不够处理时候新建进程,不超过最大进程数 */// 一个进程定时投递任务/** * 1. tick * 2. process及其管道通讯 * 3. event loop 事件循环 */class processPool{ private $pool; /** * @var swoole_process[] 记录所有worker的process对象 */ private $workers = []; /** * @var array 记录worker工作状态 */ private $used_workers = []; /** * @var int 最小进程数 */ private $min_woker_num = 5; /** * @var int 初始进程数 */ private $start_worker_num = 10; /** * @var int 最大进程数 */ private $max_woker_num = 20; /** * 进程闲置销毁秒数 * @var int */ private $idle_seconds = 5; /** * @var int 当前进程数 */ private $curr_num; /** * 闲置进程时间戳 * @var array */ private $active_time = []; public function __construct() { $this->pool = new swoole_process(function () { // 循环建立worker进程 for ($i = 0; $i < $this->start_worker_num; $i++) { $this->createWorker(); } echo '初始化进程数:' . $this->curr_num . PHP_EOL; // 每秒定时往闲置的worker的管道中投递任务 swoole_timer_tick(1000, function ($timer_id) { static $count = 0; $count++; $need_create = true; foreach ($this->used_workers as $pid => $used) { if ($used == 0) { $need_create = false; $this->workers[$pid]->write($count . ' job'); // 标记使用中 $this->used_workers[$pid] = 1; $this->active_time[$pid] = time(); break; } } foreach ($this->used_workers as $pid => $used) // 如果所有worker队列都没有闲置的,则新建一个worker来处理 if ($need_create && $this->curr_num < $this->max_woker_num) { $new_pid = $this->createWorker(); $this->workers[$new_pid]->write($count . ' job'); $this->used_workers[$new_pid] = 1; $this->active_time[$new_pid] = time(); } // 闲置超过一段时间则销毁进程 foreach ($this->active_time as $pid => $timestamp) { if ((time() - $timestamp) > $this->idle_seconds && $this->curr_num > $this->min_woker_num) { // 销毁该进程 if (isset($this->workers[$pid]) && $this->workers[$pid] instanceof swoole_process) { $this->workers[$pid]->write('exit'); unset($this->workers[$pid]); $this->curr_num = count($this->workers); unset($this->used_workers[$pid]); unset($this->active_time[$pid]); echo "{$pid} destroyed\n"; break; } } } echo "任务{$count}/{$this->curr_num}\n"; if ($count == 20) { foreach ($this->workers as $pid => $worker) { $worker->write('exit'); } // 关闭定时器 swoole_timer_clear($timer_id); // 退出进程池 $this->pool->exit(0); exit(); } }); }); $master_pid = $this->pool->start(); echo "Master $master_pid start\n"; while ($ret = swoole_process::wait()) { $pid = $ret['pid']; echo "process {$pid} existed\n"; } } /** * 创建一个新进程 * @return int 新进程的pid */ public function createWorker() { $worker_process = new swoole_process(function (swoole_process $worker) { // 给子进程管道绑定事件 swoole_event_add($worker->pipe, function ($pipe) use ($worker) { $data = trim($worker->read()); if ($data == 'exit') { $worker->exit(0); exit(); } echo "{$worker->pid} 正在处理 {$data}\n"; sleep(5); // 返回结果,表示空闲 $worker->write("complete"); }); }); $worker_pid = $worker_process->start(); // 给父进程管道绑定事件 swoole_event_add($worker_process->pipe, function ($pipe) use ($worker_process) { $data = trim($worker_process->read()); if ($data == 'complete') { // 标记为空闲// echo "{$worker_process->pid} 空闲了\n"; $this->used_workers[$worker_process->pid] = 0; } }); // 保存process对象 $this->workers[$worker_pid] = $worker_process; // 标记为空闲 $this->used_workers[$worker_pid] = 0; $this->active_time[$worker_pid] = time(); $this->curr_num = count($this->workers); return $worker_pid; }}new processPool();
以上內容希望幫助到大家
以上是帶你學習swoole_process的詳細內容。更多資訊請關注PHP中文網其他相關文章!