这篇文章主要介绍了swoole_process实现进程池的方法示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考，一起跟随小编过来看看吧。

swoole —— 重新定义PHP

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任务队列池会更方便。

假如有个定时投递的任务队列：

1. <?php 
2.  
3. /** 
4.  * 动态进程池，类似fpm 
5.  * 动态新建进程 
6.  * 有初始进程数，最小进程数，进程不够处理时候新建进程，不超过最大进程数 
7.  */ 
8.  
9. // 一个进程定时投递任务 
10.  
11. /** 
12.  * 1. tick 
13.  * 2. process及其管道通讯 
14.  * 3. event loop 事件循环 
15.  */ 
16. class processPool 
17. { 
18.   private $pool; 
19.  
20.   /** 
21.    * @var swoole_process[] 记录所有worker的process对象 
22.    */ 
23.   private $workers = []; 
24.  
25.   /** 
26.    * @var array 记录worker工作状态 
27.    */ 
28.   private $used_workers = []; 
29.  
30.   /** 
31.    * @var int 最小进程数 
32.    */ 
33.   private $min_woker_num = 5; 
34.  
35.   /** 
36.    * @var int 初始进程数 
37.    */ 
38.   private $start_worker_num = 10; 
39.  
40.   /** 
41.    * @var int 最大进程数 
42.    */ 
43.   private $max_woker_num = 20; 
44.  
45.   /** 
46.    * 进程闲置销毁秒数 
47.    * @var int 
48.    */ 
49.   private $idle_seconds = 5; 
50.  
51.   /** 
52.    * @var int 当前进程数 
53.    */ 
54.   private $curr_num; 
55.  
56.   /** 
57.    * 闲置进程时间戳 
58.    * @var array 
59.    */ 
60.   private $active_time = []; 
61.  
62.   public function __construct() 
63.   { 
64.     $this->pool = new swoole_process(function () { 
65.       // 循环建立worker进程 
66.       for ($i = 0; $i < $this->start_worker_num; $i++) { 
67.         $this->createWorker(); 
68.       } 
69.       echo '初始化进程数：' . $this->curr_num . PHP_EOL; 
70.       // 每秒定时往闲置的worker的管道中投递任务 
71.       swoole_timer_tick(1000, function ($timer_id) { 
72.         static $count = 0; 
73.         $count++; 
74.         $need_create = true; 
75.         foreach ($this->used_workers as $pid => $used) { 
76.           if ($used == 0) { 
77.             $need_create = false; 
78.             $this->workers[$pid]->write($count . ' job'); 
79.             // 标记使用中 
80.             $this->used_workers[$pid] = 1; 
81.             $this->active_time[$pid] = time(); 
82.             break; 
83.           } 
84.         } 
85.         foreach ($this->used_workers as $pid => $used) 
86.           // 如果所有worker队列都没有闲置的，则新建一个worker来处理 
87.           if ($need_create && $this->curr_num < $this->max_woker_num) { 
88.             $new_pid = $this->createWorker(); 
89.             $this->workers[$new_pid]->write($count . ' job'); 
90.             $this->used_workers[$new_pid] = 1; 
91.             $this->active_time[$new_pid] = time(); 
92.           } 
93.  
94.         // 闲置超过一段时间则销毁进程 
95.         foreach ($this->active_time as $pid => $timestamp) { 
96.           if ((time() - $timestamp) > $this->idle_seconds && $this->curr_num > $this->min_woker_num) { 
97.             // 销毁该进程 
98.             if (isset($this->workers[$pid]) && $this->workers[$pid] instanceof swoole_process) { 
99.               $this->workers[$pid]->write('exit'); 
100.               unset($this->workers[$pid]); 
101.               $this->curr_num = count($this->workers); 
102.               unset($this->used_workers[$pid]); 
103.               unset($this->active_time[$pid]); 
104.               echo "{$pid} destroyed\n"; 
105.               break; 
106.             } 
107.           } 
108.         } 
109.  
110.         echo "任务{$count}/{$this->curr_num}\n"; 
111.  
112.         if ($count == 20) { 
113.           foreach ($this->workers as $pid => $worker) { 
114.             $worker->write('exit'); 
115.           } 
116.           // 关闭定时器 
117.           swoole_timer_clear($timer_id); 
118.           // 退出进程池 
119.           $this->pool->exit(0); 
120.           exit(); 
121.         } 
122.       }); 
123.  
124.     }); 
125.  
126.     $master_pid = $this->pool->start(); 
127.     echo "Master $master_pid start\n"; 
128.  
129.     while ($ret = swoole_process::wait()) { 
130.       $pid = $ret['pid']; 
131.       echo "process {$pid} existed\n"; 
132.     } 
133.   } 
134.  
135.   /** 
136.    * 创建一个新进程 
137.    * @return int 新进程的pid 
138.    */ 
139.   public function createWorker() 
140.   { 
141.     $worker_process = new swoole_process(function (swoole_process $worker) { 
142.       // 给子进程管道绑定事件 
143.       swoole_event_add($worker->pipe, function ($pipe) use ($worker) { 
144.         $data = trim($worker->read()); 
145.         if ($data == 'exit') { 
146.           $worker->exit(0); 
147.           exit(); 
148.         } 
149.         echo "{$worker->pid} 正在处理 {$data}\n"; 
150.         sleep(5); 
151.         // 返回结果，表示空闲 
152.         $worker->write("complete"); 
153.       }); 
154.     }); 
155.  
156.     $worker_pid = $worker_process->start(); 
157.  
158.     // 给父进程管道绑定事件 
159.     swoole_event_add($worker_process->pipe, function ($pipe) use ($worker_process) { 
160.       $data = trim($worker_process->read()); 
161.       if ($data == 'complete') { 
162.         // 标记为空闲 
163. //        echo "{$worker_process->pid} 空闲了\n"; 
164.         $this->used_workers[$worker_process->pid] = 0; 
165.       } 
166.     }); 
167.  
168.     // 保存process对象 
169.     $this->workers[$worker_pid] = $worker_process; 
170.     // 标记为空闲 
171.     $this->used_workers[$worker_pid] = 0; 
172.     $this->active_time[$worker_pid] = time(); 
173.     $this->curr_num = count($this->workers); 
174.     return $worker_pid; 
175.   } 
176.  
177. } 
178.  
179. new processPool();

Tags: swoole_process php进程池

分享到：