<?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();