swoole_process实现进程池的方法示例
swoole —— 从新定义PHPphp
swoole 的进程之间有两种通讯方式，一种是消息队列(queue)，另外一种是管道(pipe)，对swoole_process 的研究在swoole中显得尤其重要。html
预备知识前端
IO多路复用sql
swoole 中的io多路复用表现为底层的 epoll进程模型，在C语言中表现为 epoll 函数。thinkphp
- 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_processswoole
- 基于C语言封装的进程管理模块，方便php来调用
- 内置管道、消息队列接口，方便实现进程间通讯
咱们在php-fpm.conf配置文件中发现，php-fpm中有两种进程池管理设置。架构
- 静态模式 即初始化固定的进程数，当来了一个请求时，从中选取一个进程来处理。
- 动态模式 指定最小、最大进程数，当请求量过大，进程数不超过最大限制时，新增线程去处理请求
接下来用swoole代码来实现，这里只是为理解swoole_process、进程间通讯、定时器等使用，实际状况使用封装好的swoole_server来实现task任务队列池会更方便。微信公众平台
假若有个定时投递的任务队列：
|
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
|
<?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();
|
相关文章
- 1. 使用 swoole_process 实现 PHP 进程池
- 2. swoole_process实现多进程
- 3. Erlang实现进程池
- 4. python进程池编程实例
- 5. python 进程池（multiprocessing.Pool）和线程池（threadpool.ThreadPool）的区别与实例
- 6. 多种方法实现 python 线程池
- 7. Dubbo+Zookeeper实现简单的远程方法调用示例
- 8. swoole_process父子进程管道通讯案例
- 9. python 进程池pool简单实例
- 10. Swoole 源码分析——进程管理 Swoole_Process
- 更多相关文章...
- • Spring实例化Bean的三种方法 - Spring教程
- • 现实生活中的 XML - XML 教程
- • ☆基于Java Instrument的Agent实现
- • SpringBoot中properties文件不能自动提示解决方法
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
欢迎关注本站公众号,获取更多信息
相关文章