Nehmen Sie mit, um swoole_process zu lernen

Empfehlung (kostenlos): swoole

swoole Es gibt zwei Kommunikationsmethoden zwischen Prozessen, eine ist die Nachrichtenwarteschlange (Queue), die andere ist die Pipe (Pipe), die Forschung zu swoole_process ist darin enthalten ist besonders wichtig in Swoole.

Vorkenntnisse

IO-Multiplexing

Das IO-Multiplexing in Swoole wird durch das zugrunde liegende Epoll-Prozessmodell dargestellt, das durch die Epoll-Funktion in der C-Sprache dargestellt wird.

Das Epoll-Modell überwacht weiterhin den Socket-Deskriptor fd unter seinem eigenen Namen.
Wenn ein vom Socket überwachtes Ereignis ausgelöst wird, antwortet die Epoll-Funktion und gibt eine Sammlung aller zu diesem Zeitpunkt empfangsbereiten Sockets zurück. Das Wesentliche von Epoll ist Um E/A zu blockieren, besteht der Vorteil darin, dass eine große Anzahl von Socket-Verbindungen gleichzeitig verarbeitet werden kann.
Ereignisschleife Ereignisschleife

swoole implementiert eine Reactor-Thread-Modellkapselung von Epoll und richtet Überwachungsrückruffunktionen für Lese- und Schreibereignisse ein . (Einzelheiten finden Sie unter swoole_event_add)

Event-Schleife ist ein Reactor-Thread, in dem eine Epoll-Instanz ausgeführt wird.

Fügen Sie über swoole_event_add ein Ereignis des Socket-Deskriptors zum Epoll-Listener hinzu, und die Rückruffunktion wird ausgeführt, wenn das Ereignis auftritt.
Sie kann nicht in der FPM-Umgebung verwendet werden, da FPM den Prozess am Ende der Aufgabe möglicherweise herunterfährt.

swoole_process

Ein in C-Sprache gekapseltes Prozessmanagementmodul, das für PHP bequem aufzurufen ist

Integrierte Pipes und Nachrichtenwarteschlangenschnittstellen zur Erleichterung der Kommunikation zwischen Prozessen
Das haben wir in der Konfigurationsdatei php-fpm.conf gefunden sind zwei Einstellungen für die Prozesspoolverwaltung.

Im statischen Modus wird eine feste Anzahl von Prozessen initialisiert. Wenn eine Anfrage eingeht, wird ein Prozess ausgewählt, der diese bearbeitet.

Der dynamische Modus gibt die minimale und maximale Anzahl von Prozessen an. Wenn das Anforderungsvolumen zu groß ist und die Anzahl der Prozesse das maximale Limit nicht überschreitet, wird ein neuer Thread hinzugefügt, um die Anforderung zu verarbeiten Dies dient nur dem Verständnis von swoole_process und der Kommunikation zwischen Prozessen. Für die Verwendung von Timern usw. ist es tatsächlich bequemer, den gekapselten swoole_server zum Implementieren des Aufgabenwarteschlangenpools zu verwenden.

Wenn es eine Aufgabenwarteschlange für die geplante Lieferung gibt:

<?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 &#39;初始化进程数：&#39; . $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 . &#39; job&#39;);
            // 标记使用中
            $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 . &#39; job&#39;);
            $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(&#39;exit&#39;);
              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(&#39;exit&#39;);
          }
          // 关闭定时器
          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[&#39;pid&#39;];
      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 == &#39;exit&#39;) {
          $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 == &#39;complete&#39;) {
        // 标记为空闲//        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();

Ich hoffe, dass der obige Inhalt allen hilft

Das obige ist der detaillierte Inhalt vonNehmen Sie mit, um swoole_process zu lernen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!