php 进程池设计与实现，phper必学！

phper 为什么要学习进程池

在php开发过程中经常使用的 php-fpm 使用的进程模型就是进程池，学习进程池知识能让我们更好理解php-fpm 的运行模式，进程池也是php中主流的并发服务器解决方案

包含我们的 Workerman 也是用的是进程池，编写一个简单的进程池可以帮助我们更好学习Workerman 源码，了解Workerman 为何如此设计

池的概念

池是一组资源的集合，这组资源在服务器启动之初就完全被创建并初始化，这称为静态资源分配。当服务器进入正式运行阶段，即开始处理客户请求的时候，如果它需要相关的资源，就可以直接从池中获取，无需动态分配

很明显，直接从池中取得所需资源比动态分配资源的速度要快得多，因为分配系统资源的系统调用都是很耗时的。当服务器处理完一个客户连接后，可以把相关的资源放回池中，无需执行系统调用来释放资源。从最终效果来看，池相当于服务器管理系统资源的应用设施，它避免了服务器对内核的频繁访问。

进程池技术的应用至少由以下两部分组成：

资源进程：预先创建好的空闲进程，管理进程会把工作分发到空闲进程来处理。

管理进程：管理进程负责创建资源进程，把工作交给空闲资源进程处理，回收已经处理完工作的资源进程。

为什么要有进程池?

动态创建进程缺点

操作系统繁忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。
那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？

首先，动态创建进程（或线程）是比较耗费时间的，这将导致较慢的客
户响应

即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。

动态创建的子进程（或子线程）通常只用来为一个客户服务（除非我们做特殊的处理），这将导致系统上产生大量的细微进程（或线程）。进程（或线程）间的切换将消耗大量CPU时间。

动态创建的子进程是当前进程的完整映像。当前进程必须谨慎地管理其分配的文件描述符和堆内存等系统资源，否则子进程可能复制这些资源，从而使系统的可用资源急剧下降，进而影响服务器的性能

因此我们不能无限制的根据任务去开启或者结束进程。

进程池的优点

进程池是由服务器预先创建的一组子进程，这些子进程的数目在3～10个之间（当然，这只是典型情况）。具体看服务器配置

进程池中的所有子进程都运行着相同的代码，并具有相同的属性，比如优先级、PGID等。因为进程池在服务器启动之初就创建好了，所以每个子进程都相对“干净”，即它们没有打开不必要的文件描述符（从父进程继承而来），也不会错误地使用大块的堆内存（从父进程复制得到）。

选择子进程为新任务服务的方式

当有新的任务到来时，主进程将通过某种方式选择进程池中的某一个子进程来为之服务。相比于动态创建子进程，选择一个已经存在的子进程的代价显然要小得多。至于主进程选择哪个子进程来为新任务服务，则有两种方式：

第一种：主进程使用某种算法来主动选择子进程。最简单、最常用的算法是随机算法和Round Robin（轮流选取）算法，但更优秀、更智能的算法将使任务在各个工作进程中更均匀地分配，从而减轻服务器的整体压力。

第二种：主进程和所有子进程通过一个共享的工作队列来同步，子进程都睡眠在该工作队列上。当有新的任务到来时，主进程将任务添加到工作队列中。这将唤醒正在等待任务的子进程，不过只有一个子进程将获得新任务的“接管权”，它可以从工作队列中取出任务并执行之，而其他子进程将继续睡眠在工作队列上。

本篇文章使用第一种实现任务调度
当选择好子进程后，主进程还需要使用某种通知机制来告诉目标子进程有新任务需要处理，并传递必要的数据。这边我使用的是ipc消息队列进行父子进程间的通信

进程池模型

服务端

namespace php_pool;
class Process
{
    public $_pid;//进程pid
    public $_msqid;//进程通信消息队列msqid
}

class Server
{
    public $_sockFile = "pool.sock";// 定义一个sock 文件名 用于unix域通信
    public $_processNum = 3;//默认进程池启动 worker 进程
    public $_keyFile = "pool.php";

    public $idx;//方便子进程获取进程对象

    public $_process = []; // 进程池进程对象数组
    public $_sockfd; // 存放当前进程sock 实例
    public $_run = true; //运行开关

    public $_roll = 0; //轮询算法参数

    public $exitpid = []; //程序退出，回收子进程

	/**
	* 信号处理函数
	*/
    public function sigExitHandler($signo)
    {
        $this->_run = false;
    }

    public function __construct($num = 3)
    {
        $this->_processNum = $num;
        // 安装 SIGINT 信号  Ctrl+c 触发该信号
        pcntl_signal(SIGINT, [$this, "sigExitHandler"]);
        $this->forkWorker();// 创建一些 worker 进程
        cli_set_process_title('master'); // 设置主进程名 方便查看
        $this->Listen(); // 监听请求
        // 回收 worker 进程，避免僵尸进程
        while (1) {
            $pid = pcntl_wait($status);
            if ($pid > 0) {
                $this->exitpid[] = $pid;
            }
            if (count($this->exitpid) == $this->_processNum) {
                break;
            }
        }
        /** @var Process $p */
        foreach ($this->_process as $p) {
        	// 移除消息队列
            msg_remove_queue($p->_msqid);
        }
        // 主进程退出
        fprintf(STDOUT, "master shutdown\n");
    }
	
    public function forkWorker()
    {
    	// 实例化 worker 进程对象
        $processObj = new Process();
        for ($i = 0; $i < $this->_processNum; $i++) {
            $key = ftok($this->_keyFile, $i);
            $mqsid = msg_get_queue($key);// 创建 worker 通信的消息队列
            $process = clone $processObj; // 克隆 worker 进程对象
            $process->_msqid = $mqsid; 
            $this->_process[$i] = $process;
            $this->idx = $i; // 方便 worker 进程使用进程对象
            /**
             * @var Process $this
             */
            $this->_process[$i]->_pid = pcntl_fork();// 派生 worker 子进程
            if ($this->_process[$i]->_pid == 0) {//子进程逻辑
                $this->Worker(); // 启动 worker 子进程
            } else {//父进程逻辑
                continue;
            }
        }
    }

    public function Listen()
    {
    	// 创建 sock 文件描述符实例
        $this->_sockfd = socket_create(AF_UNIX, SOCK_STREAM, 0);
        // 创建异常判断
        if (!is_resource($this->_sockfd)) {
            fprintf(STDOUT, "socket create fail:%s\n", socket_strerror(socket_last_error($this->_sockfd)));

        }
        // 移除上传遗留的 pool.sock 避免重复绑定错误
        unlink($this->_sockFile);
        // 将sockfd 套接字绑定到文件上
        if (!socket_bind($this->_sockfd, $this->_sockFile)) {
            fprintf(STDOUT, "socket bind fail\n", socket_strerror(socket_last_error($this->_sockfd)));
        }
		// 监听套接字上的连接
        socket_listen($this->_sockfd, 10);
        // 启动事件循环操作
        $this->evenLoop();
    }
	
	// worker 轮询算法实现
    public function selectWorker($data)
    {
        /**
         * @var Process $process
         */
         //轮询获取 worker 进程实例
        $process = $this->_process[$this->_roll++ % $this->_processNum];
        // 获取worker 消息队列 msq_id 
        $msgid = $process->_msqid;
        // 往worker 进程消息队列 投递 请求任务 msg_send 函数设置为非阻塞
        if (msg_send($msgid, 1, $data, true, false)) {
            fprintf(STDOUT, "send ok\n");
        }
    }

    public function evenLoop()
    {
		// 获取 sock 套接字 注册到 socket_select 可读事件
        $readFds = [$this->_sockfd];
	
        $writeFds = [];
        $exFds = [];
        while ($this->_run) {
        	//socket_select 函数就可以对集合$readFds中的数据是否发生可读行为进行监听【可写、异常等 我们暂且不表设置为空】以达到在同一个进程中实时处理多个IO的目的 接受套接字数组并等待它们改变状态。这里我们只监听了可读事件
            $ret = socket_select($readFds, $writeFds, $exFds, null, null);
            \pcntl_signal_dispatch();// 分发信号
   			// socket_select 函数异常直接退出
            if (false === $ret) {
                break;
            } else if ($ret === 0) {//没有事件发生返回 0
                continue;
            }
			
            if($readFds){
                foreach ($readFds as $fd){
                    if($fd == $this->_sockfd){
                    	// 接收客户端请求
                        $connfd = socket_accept($fd);
                        // 读取客户端发来数据
                        $data = socket_read($connfd,1024);
                        if($data){
                        	// 选择一个 worker 来处理 客户端请求
                            $this->selectWorker($data);
                        }
                        // 给客户端 一个响应
                        socket_write($connfd,"ok",2);
                        // 关闭客户端连接
                        socket_close($connfd);
                    }
                }
            }
        }
       	// 主进程退出
        // 关闭 sock 实例
        socket_close($this->_sockfd);
        /**
         * @var Process $p
         */
		// 给所有 worker 进程发送进程退出消息
        foreach ($this->_process as $p) {
            if (msg_send($p->_msqid, 1, 'quit')) {
                fprintf(STDOUT, "master send quit ok\n");
            }
        }

    }

	// 处理业务逻辑 worker 进程，具体流程由业务逻辑决定
    public function Worker()
    {
        fprintf(STDOUT, "child pid=%d start\n", posix_getpid());
        /**
         * @var Process $process
         */
        $process = $this->_process[$this->idx];
        $msgid = $process->_msqid;
        while (1) {
        	// 获取主进程投递过来的客户端请求消息
            if (msg_receive($msgid, 0, $msgType, 1024, $msg)) {
                fprintf(STDOUT, "child pid =%d recv:%s\n", posix_getpid(), $msg);
                // 监听退出命令
                if (strncasecmp($msg, 'quit', 4) == 0) {
                    break;
                }
            }
        }

        fprintf(STDOUT, "child pid=%d chutdown\n", posix_getpid());
        exit(0);
    }
}

(new Server(3));
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

客户端

namespace php_pool;
$sockFile = "pool.sock";
$_sockfd = socket_create(AF_UNIX, SOCK_STREAM, 0);
// 连接服务端
if(socket_connect($_sockfd,$sockFile)){
        $data = 'hello';
        // 发送请求消息给服务端
        socket_write($_sockfd,$data,strlen($data));
        // 接收服务端响应
        echo socket_read($_sockfd,1024)."\n\r";
        //关闭连接
        socket_close($_sockfd);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

首先我们运行服务端代码，服务运行后进程池有3个 worker 进程，没问题

接下来我们运行客户端代码，发个 hello 消息给服务端，让它给我们处理一下，多次运行服务端消息接收正常，服务端每次都轮询一个进程池里的worker 进程为我们处理任务

最后测试服务端退出，所有 worker 进程 正常退出 主进程回收完子进程资源也正常退出，不过 socket_select 函数报了警告，不过不用理会，这个是我们按下 ctrl+c 导致系统调用中断导致的警告，可以忽略

结语

进程池的设计根据业务不同写法会有所差异，但大致流程都差不多，本次编写的进程池也只是学习使用，并不特别完善，如果有问题请联系我改正谢谢！