swoole_process를 배우도록 안내합니다.

추천(무료): swoole

swoole 프로세스 간 통신 방법에는 두 가지가 있는데, 하나는 메시지 큐(queue), 다른 하나는 파이프(pipe), swoole_process에 대한 연구가 나와 있습니다. 스울에서는 특히 중요합니다.

사전 지식

IO 다중화

Swoole의 IO 다중화는 C 언어의 epoll 함수로 표현되는 기본 epoll 프로세스 모델로 표현됩니다.

epoll 모델은 자신의 이름으로 소켓 설명자 fd를 계속 모니터링합니다.
소켓이 모니터링하는 이벤트가 트리거되면 epoll 함수는 응답하고 그 당시 수신 중인 모든 소켓의 컬렉션을 반환합니다.
epoll의 본질은 IO를 차단하기 위해 동시에 많은 수의 소켓 연결을 처리할 수 있다는 장점이 있습니다.
이벤트 루프 이벤트 루프

swoole은 epoll의 Reactor 스레드 모델 캡슐화를 구현하고 이벤트 읽기 및 쓰기 이벤트에 대한 모니터링 콜백 기능을 설정합니다. . (자세한 내용은 swoole_event_add 참조)

이벤트 루프는 epoll 인스턴스가 실행되는 Reactor 스레드입니다.
swoole_event_add를 통해 epoll 리스너에 소켓 설명자의 이벤트를 추가하면 이벤트 발생 시 콜백 함수가 실행됩니다.
fpm이 작업 종료 시 프로세스를 종료할 수 있으므로 fpm 환경에서는 사용할 수 없습니다.

swoole_process

C 언어로 캡슐화된 프로세스 관리 모듈, PHP 호출에 편리함
프로세스 간 통신을 용이하게 하는 내장 파이프 및 메시지 큐 인터페이스
php-fpm.conf 구성 파일에서 두 가지가 있음을 발견했습니다. 프로세스 풀 관리 설정.

정적 모드는 고정된 수의 프로세스를 초기화하는 것입니다. 요청이 오면 이를 처리하기 위해 하나의 프로세스가 선택됩니다.
동적 모드는 요청 볼륨이 너무 크고 프로세스 수가 최대 제한을 초과하지 않는 경우 새 스레드를 추가하여 요청을 처리합니다. 이것은 단지 swoole_process, 프로세스 간 통신을 이해하기 위한 것입니다. 타이머 등을 사용하려면 캡슐화된 swoole_server를 사용하여 작업 대기열 풀을 구현하는 것이 실제로 더 편리합니다.

예약 배송 작업 대기열이 있는 경우:

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

위 내용이 모두에게 도움이 되기를 바랍니다

위 내용은 swoole_process를 배우도록 안내합니다.의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!