php-fpm参数调优

关于php-fpm.conf参数调优,只对重要的参数进程调优.其它可参数前辈的.

http://php.net/manual/zh/install.fpm.configuration.php (官方的)

http://www.cnblogs.com/argb/p/3604340.html

http://www.cnblogs.com/jonsea/p/5522018.html

https://www.zybuluo.com/phper/note/89081

http://blog.64mazi.com/299/(推荐阅读)

1
2
3
4
5
关于emergency_restart_threshold 和emergency_restart_interval 参数,设置多少合适
额定内存为4G,拿出一半内存做php使用,即2G,每个php-fpm进程30m(经验值)
则2g最多可以承受2048/30~=68即设置为
emergency_restart_threshold=60(60个进程数)
emergency_restart_interval=60s(60秒内出现60个进程则优雅的重进php-fpm服务)
pm表示使用那种方式,有两个值可以选择,就是static(静态)或者dynamic(动态)。在更老一些的版本中,dynamic被称作apache-like。这个要注意看配置文件的说明。

下面4个参数的意思分别为:

pm.max_children:静态方式下开启的php-fpm进程数量。
pm.start_servers:动态方式下的起始php-fpm进程数量。
pm.min_spare_servers:动态方式下的最小php-fpm进程数量。
pm.max_spare_servers:动态方式下的最大php-fpm进程数量。

如果dm设置为static,那么其实只有pm.max_children这个参数生效。系统会开启设置数量的php-fpm进程。

如果dm设置为dynamic,那么pm.max_children参数失效,
后面3个参数生效。
系统会在php-fpm运行开始的时候启动pm.start_servers个php-fpm进程,
然后根据系统的需求动态在pm.min_spare_servers和pm.max_spare_servers之间调整php-fpm进程数。

公式:

假定一个php-fpm进程为30m计算,若4g内存,拿出一半做的php-fpm使用即2g
mem=2048pm=dynamicpm.maxchildren=100(

mem/20)
pm.start_servers = 75 (mem/30)minspareservers+(maxspareserversminspareservers)/2pm.minspareservers=50(

mem/40)
pm.max_spare_servers = 50 ($mem/20)

pm.max_children = 50 是最大可创建的子进程的数量。必须设置。这里表示最多只能50个子进程。

pm.start_servers = 20 随着php-fpm一起启动时创建的子进程数目。默认值:min_spare_servers + (max_spare_servers – min_spare_servers) / 2。这里表示,一起启动会有20个子进程。

pm.min_spare_servers = 10
设置服务器空闲时最小php-fpm进程数量。必须设置。如果空闲的时候,会检查如果少于10个,就会启动几个来补上。

pm.max_spare_servers = 30
设置服务器空闲时最大php-fpm进程数量。必须设置。如果空闲时,会检查进程数,多于30个了,就会关闭几个,达到30个的状态。

实际上的内存消耗是max_children*max_requests*每个请求使用内存
其它php优化
尽量少安装PHP模块,最简单是最好(快)的
socket连接FastCGI,/dev/shm是内存文件系统,socket放在内存中肯定会快些
cat >> /etc/security/limits.conf <<EOF
* soft nproc 65535
* hard nproc 65535
* soft nofile 65535
* hard nofile 65535
EOF
增加 PHP-FPM 打开文件描述符的限制:
# vi $php_install_dir/etc/php-fpm.conf
rlimit_files = 51200

以下摘自(麻子来了)http://blog.64mazi.com/299/

一、常用参数解释:


1)pm = dynamic #对于专用服务器,pm可以设置为static。
#如何控制子进程,选项有static和dynamic。
如果选择static,则由pm.max_children指定固定的子进程数。
如果选择dynamic,则由pm.max_children、pm.start_servers、pm.min_spare_servers、pm.max_spare_servers 参数决定.
2)pm.max_children

在同一时间最大的进程数

pm.max_children = 120

3)pm.start_servers

php-fpm启动时开启的等待请求到来的进程数,默认值为:min_spare_servers + (max_spare_servers - min_spare_servers) / 2

pm.start_servers = 80

4)pm.min_spare_servers
在空闲状态下,运行的最小进程数,如果小于此值,会创建新的进程
pm.min_spare_servers = 60

5)pm.max_spare_servers
在空闲状态下,运行的最大进程数,如果大于此值,会kill部分进程
pm.max_spare_servers = 120

6)pm.process_idle_timeout
空闲多少秒之后进程会被kill,默认为10s
pm.process_idle_timeout = 10s

7)pm.max_requests
每个进程处理多少个请求之后自动终止,可以有效防止内存溢出,如果为0则不会自动终止,默认为0#设置每个子进程重生之前服务的请求数. 对于可能存在内存泄漏的第三方模块来说是非常有用的. 
如果设置为 '0' 则一直接受请求. 等同于 PHP_FCGI_MAX_REQUESTS 环境变量. 默认值: 0.
pm.max_requests = 500 

8)pm.status_path
注册的URI,以展示php-fpm状态的统计信息
     pm.status_path = /status
     其中统计页面信息有:
     pool 进程池名称
     process manager 进程管理器名称(static, dynamic or ondemand)
     start time php-fpm启动时间
     start since php-fpm启动的总秒数
     accepted conn 当前进程池接收的请求数
     listen queue 等待队列的请求数
     max listen queue 自启动以来等待队列中最大的请求数
     listen queue len 等待连接socket队列大小
     idle processes 当前空闲的进程数
     active processes 活动的进程数
     total processes 总共的进程数(idle+active)
     max active processes 自启动以来活动的进程数最大值
     max children reached 达到最大进程数的次数

9)ping.path
ping url,可以用来测试php-fpm是否存活并可以响应
ping.path = /ping

10)ping.response
ping url的响应正文返回为 HTTP 200 的 text/plain 格式文本. 默认值: pong.
ping.response = pong


11)pid = run/php-fpm.pid
#pid设置,默认在安装目录中的var/run/php-fpm.pid,建议开启
 
12)error_log = log/php-fpm.log
#错误日志,默认在安装目录中的var/log/php-fpm.log
 
13)log_level = notice
#错误级别. 可用级别为: alert(必须立即处理), error(错误情况), warning(警告情况), notice(一般重要信息), debug(调试信息). 默认: notice.
 
14)emergency_restart_threshold = 60
emergency_restart_interval = 60s
#表示在emergency_restart_interval所设值内出现SIGSEGV或者SIGBUS错误的php-cgi进程数如果超过 emergency_restart_threshold个,php-fpm就会优雅重启。这两个选项一般保持默认值。
 
15)process_control_timeout = 0
#设置子进程接受主进程复用信号的超时时间. 可用单位: s(秒), m(分), h(小时), 或者 d(天) 默认单位: s(秒). 默认值: 0.
 
16)daemonize = yes
#后台执行fpm,默认值为yes,如果为了调试可以改为no。在FPM中,可以使用不同的设置来运行多个进程池。 这些设置可以针对每个进程池单独设置。
 
17)listen = 127.0.0.1:9000
#fpm监听端口,即nginx中php处理的地址,一般默认值即可。可用格式为: 'ip:port', 'port', '/path/to/unix/socket'. 每个进程池都需要设置.
 
18)listen.backlog = -1
#backlog数,-1表示无限制,由操作系统决定,此行注释掉就行。 19)listen.allowed_clients = 127.0.0.1
#允许访问FastCGI进程的IP,设置any为不限制IP,如果要设置其他主机的nginx也能访问这台FPM进程,listen处要设置成本地可被访问的IP。
默认值是any。每个地址是用逗号分隔. 如果没有设置或者为空,则允许任何服务器请求连接


20)#unix socket设置选项,如果使用tcp方式访问,这里注释即可。
listen.owner = www
listen.group = www
listen.mode = 0666
#启动进程的帐户和组
user = www
group = www
21)request_terminate_timeout = 0 #设置单个请求的超时中止时间. 
该选项可能会对php.ini设置中的'max_execution_time'因为某些特殊原因没有中止运行的脚本有用. 设置为 '0' 表示 'Off'.当经常出现502错误时可以尝试更改此选项。 

22)request_slowlog_timeout = 10s #当一个请求该设置的超时时间后,就会将对应的PHP调用堆栈信息完整写入到慢日志中. 设置为 '0' 表示 'Off' 

23)slowlog = log/$pool.log.slow #慢请求的记录日志,配合request_slowlog_timeout使用 

24)rlimit_files = 1024 #设置文件打开描述符的rlimit限制. 默认值: 系统定义值默认可打开句柄是1024,可使用 ulimit -n查看,ulimit -n 2048修改。 

25)rlimit_core = 0 #设置核心rlimit最大限制值. 可用值: 'unlimited' 、0或者正整数. 默认值: 系统定义值. 

26)chroot = #启动时的Chroot目录. 所定义的目录需要是绝对路径. 如果没有设置, 则chroot不被使用. 

27)chdir = #设置启动目录,启动时会自动Chdir到该目录. 所定义的目录需要是绝对路径. 默认值: 当前目录,或者/目录(chroot时) 

28)catch_workers_output = yes #重定向运行过程中的stdout和stderr到主要的错误日志文件中. 
如果没有设置, stdout 和 stderr 将会根据FastCGI的规则被重定向到 /dev/null . 
默认值: 空. 

二、php对子进程的三种管理方式
static:
表示在php-fpm运行时直接fork出 pm.max_chindren个子进程
dynamic:
表示,运行时fork出pm.start_servers个进程,随着负载的情况,动态的调整,最多不超过pm.max_children个进程。同时,保证闲置进程数不少于pm.min_spare_servers数量,否则新的进程会被创建,当然也不是无限制的创建,最多闲置进程不超过pm.max_spare_servers数量,超过则一些闲置进程被清理。
ondemand: 
当有请求时,创建进程,启动不创建,最多不超过pm.max_chindren进程数,当进程闲置会在pm.process_idle_timeout秒后被及时释放。


三、重要参数的理解与设置

 

【重要一】
request_terminate_timeout = 120
#表示等待120秒后,结束那些没有自动结束的php脚本,以释放占用的资源。
当PHP运行在php-fpm模式下,php.ini配置的max_execute_time是无效的,
需要在php-fpm.conf中配置另外一个配置项:request_terminate_timeout;以下是官方文档的说明:


set_time_limit()和max_execution_time只影响脚本本身执行的时间。
(这两个参数在php.ini中)任何发生在诸如使用system()的系统调用,流操作,数据库操作等的脚本执行的最大时间不包括其中.
还可以是以下情况:
参考:http://zyan.cc/tags/request_terminate_timeout/1/
  1. $ctx = stream_context_create(array(  
  2.    ‘http’ => array(  
  3.        ‘timeout’ => 1 //设置一个超时时间,单位为秒  
  4.        )
  5.    )
  6. );
  7. file_get_contents(“http://example.com/”, 0, $ctx);  //设置超时时间


下面4个参数的意思分别为:
pm.max_children:静态方式下开启的php-fpm进程数量。
pm.start_servers:动态方式下的起始php-fpm进程数量。
pm.min_spare_servers:动态方式下的最小php-fpm进程数量。
pm.max_spare_servers:动态方式下的最大php-fpm进程数量。

如果dm设置为static,那么其实只有pm.max_children这个参数生效。系统会开启设置数量的php-fpm进程。
如果dm设置为 dynamic,那么pm.max_children参数失效,后面3个参数生效。 
系统会在php-fpm运行开始 的时候启动pm.start_servers个php-fpm进程,
然后根据系统的需求动态在pm.min_spare_servers和 pm.max_spare_servers之间调整php-fpm进程数。

比如说512M的VPS,建议pm.max_spare_servers设置为20。至于pm.min_spare_servers,则建议根据服 
务器的负载情况来设置,比较合适的值在5~10之间。


pm = dynamic模式非常灵活,也通常是默认的选项。
但是,dynamic模式为了最大化地优化服务器响应,会造成更多内存使用,因为这种模式只会杀掉超出最大闲置进程数(pm.max_spare_servers)的闲置进程,
比如最大闲置进程数是30,然后网站经历了一次访问高峰,高峰期时共动态开启了50个进程全部忙碌,0个闲置进程数,
接着过了高峰期,可能没有一个请求,于是会有50个闲置进程,但是此时php-fpm只会杀掉20个子进程,
始终剩下30个进程继续作为闲置进程来等待请求,这可能就是为什么过了高峰期后即便请求数大量减少服务器内存使用却也没有大量减少,
也可能是为什么有些时候重启下服务器情况就会好很多,因为重启后,php-fpm

本文来自网络:http://www.cnblogs.com/300js/p/5920124.html

避免PHP-FPM内存泄漏导致内存耗尽

对于PHP-FPM多进程的模式,想要避免内存泄漏问题很简单,就是要让PHP-CGI在处理一定数量进程后退出即可。
否则PHP程序或第三方模块(如Imagemagick扩展)导致的内存泄漏问题会导致内存耗尽或不足。
php-fpm.conf中有相关配置:
#请自行按需求配置

pm.max_requests = 1024

实际上还有另一个跟它有关联的值max_children,这个是每次php-fpm会建立多少个进程,这样实际上的内存消耗是max_children*max_requests*每个请求使用内存。

另外一些粗暴的方法包括建立cron kill掉占用内存过多的php-cgi,

1.检查php进程的内存占用,杀掉内存使用超额的进程

一般情况下,如果php-cgi进程占用超过1%的内存,就得考虑一下是否要杀掉它了。因为普通情况下,php-cgi进程一般占用0.2%或以下。

这里提供一个脚本供各位使用,就是放在cron任务里,每分钟执行一次。

使用crontab -e 命令,然后添加如下调度任务

* * * * * /bin/bash /usr/local/script/kill_php_cgi.sh

kill_php_cgi.sh脚本如下

#!/bin/sh
#如果是要杀掉php-fpm的进程,下面的语句中php-cgi请改成php-fpm
pids=`ps -ef|grep php-cgi|grep -v “grep”|grep -v “$0″| awk ‘{print $2}’`
if [ “$pids” != “” ];then
for  pid  in   $pids;
do
kill -9 $pid
done

fi

2.增加内存,将PHP_FCGI_MAX_REQUESTS的值设置成跟你内存总存储量相对应的值

3.优化程序,降低处理每次请求占用的内存大小

如果PHP-FPM能够提供配置子进程内存超过指定大小就被kill,那就省事多了

以上是云栖社区小编为您精心准备的的内容,在云栖社区的博客、问答、公众号、人物、课程等栏目也有的相关内容,欢迎继续使用右上角搜索按钮进行搜索内存 , 程序 , 进程 , 内存泄漏 , 配置 php-fpm php fpm内存泄漏、如何避免内存泄漏、c 避免内存泄漏、避免内存泄漏、信息泄漏可导致,以便于您获取更多的相关知识。

本文转自网络  https://yq.aliyun.com/ziliao/9107

CentOS7 linux下yum安装redis以及使用

 一、安装redis

1、下载fedora的epel仓库

1
yum install epel-release

2、安装redis数据库

1
yum install redis

3、安装完毕后,使用下面的命令启动redis服务

1
2
3
4
5
6
7
8
# 启动redis
service redis start
# 停止redis
service redis stop
# 查看redis运行状态
service redis status
# 查看redis进程
ps -ef | grep redis

5、设置redis为开机自动启动

1
chkconfig redis on  或 systemctl enable redis.service

6、进入redis服务

1
2
3
4
# 进入本机redis
redis-cli
# 列出所有key
keys *

7、防火墙开放相应端口

1
2
3
4
5
6
7
8
# 开启6379
/sbin/iptables -I INPUT -p tcp --dport 6379 -j ACCEPT
# 开启6380
/sbin/iptables -I INPUT -p tcp --dport 6380 -j ACCEPT
# 保存
/etc/rc.d/init.d/iptables save
# centos 7下执行
service iptables save

二、修改redis默认端口和密码

1、打开配置文件

1
vi /etc/redis.conf

2、修改默认端口,查找 port 6379 修改为相应端口即可

3、修改默认密码,查找 requirepass foobared 将 foobared 修改为你的密码

4、使用配置文件启动 redis

1
redis-server /etc/redis.conf &

5、使用端口登录

1
redis-cli -h 127.0.0.1 -p 6179

6、此时再输入命令则会报错

7、输入刚才输入的密码

1
auth 111

8、停止redis

命令方式关闭redis

1
2
redis-cli -h 127.0.0.1 -p 6179
shutdown

进程号杀掉redis

1
2
ps -ef | grep redis
kill -9 XXX

三、使用redis desktop manager远程连接redis

1、访问如下网址下载redis desktop manager

1
2
<a href="https://redisdesktop.com/download" target="_blank">https://redisdesktop.com/download
</a>

2、安装后启动,新建一个连接

3、填写如下信息后点击“Test Connection”测试是否连接成功

4、如果长时间连接不上,可能有两种可能性

a)bind了127.0.01:只允许在本机连接redis

b)protected-mode设置了yes(使用redis desktop manager工具需要配置,其余不用)

解决办法:

1
2
3
4
5
# 打开redis配置文件
vi /etc/redis.conf
# 找到 bind 127.0.0.1 将其注释
# 找到 protected-mode yes 将其改为
protected-mode no

5、重启redis

1
2
service redis stop
service redis start

6、再次连接即可

 

参考:

https://www.cnblogs.com/wiseroll/p/7061673.html

http://blog.csdn.net/gebitan505/article/details/54602662

http://blog.csdn.net/lh2420124680/article/details/75426144

 

本文转发自:https://www.cnblogs.com/rslai/p/8249812.html

session的垃圾回收机制

session.gc_maxlifetime

session.gc_probability

session.gc_divisor

session.gc_divisor 与 session.gc_probability 合起来定义了在每个会话初始化时启动 gc(garbage collection 垃圾回收)进程的概率。此概率用 gc_probability/gc_divisor 计算得来。例如 1/100 意味着在每个请求中有 1% 的概率启动 gc 进程。session.gc_divisor 默认为 100。

比如:session.gc_maxlifetime=30,session.gc_divisor=1000,session.gc_probability=1,就表示每一千个用户调用session_start()的时候,就百分百的会执行一次垃圾回收机制,将磁盘上没用的session文件删除。

注意:一般对于一些大型的门户网站,建议将session.gc_divisor调大一点,减少开销

接下来,我通过一个例子演示下,如何配置才能让调用gc(垃圾回收)进程呢!

通过配置php.ini文件,修改以下几个信息:

  1. session.gc_maxlifetime = 60//当session文件在60s后还没有被访问的话,则该session文件将会被视为“垃圾文件”,并且等待gc(垃圾回收)进程的调用的时候被清理掉
  2. session.gc_probability = 1000

因为gc进程被调用的概率是通过gc_probability/gc_divisor 计算得来的,这里我将session.gc_probability改成1000,而session.gc_divisor 默认情况下也是1000。则gc进程在每次执行session_start()函数的时候都会被调用到。

以下我通过截图简单的说明下:

我开启三个会话,则创建三个对应的session文件,当每个文件在30秒内都没被调用的话,就会被当成是“垃圾文件”,等到gc进程调用的时候,“垃圾文件”就会被unlink,因为之前我已经通过修改php.ini配置文件,将gc被调用的概率改成百分百,所以接下来,如果我重新使用任何一个浏览器刷新下页面的时候,三个session文件,应该只剩下一个了

本文转自网络 https://www.cnblogs.com/hongfei/archive/2012/06/17/2552434.html

Access-Control-Allow-Origin跨域访问

PHP 设置跨域访问:

header("Access-Control-Allow-Origin:http://api.maxianwei.cn");

允许api.maxianwei.cn访问 ,那么要允许所有域名访问呢:

header("Access-Control-Allow-Origin:*");

要设置多个域名呢? Access-Control-Allow-Origin 只能设置一个域名 用逗号或其他连接符合是不行的,其实只要判断请求来源域名是不是自己允许的,然后设置即可代码如下:

$orginAllows = ['http://api.maxianwei.cn','http://api2.maxianwei.cn'];
$referOrgin = $_SERVER['HTTP_REFERER'];
$pattern = '/(http[s]{0,1}:\/\/[^\/]*)\/.*/';
preg_match($pattern, $referOrgin, $match);
$refer = $match[1];
if(in_array($refer,$orginAllows)){
   header("Access-Control-Allow-Origin:{$refer}");
}

PHP 将大量数据导出到 Excel 的方法一

数据量很大时(5万条以上),用 PHPExcel 导出 xls 将十分缓慢且占用很大内存,最终造成运行超时或内存不足。

set_time_limit(0);
ini_set(“memory_limit”,”512M”);

可以通过设置 PHP 的运行时间和内存限制来阻止错误发生,但仍然不能确保导出完成。

<?php
// 不限制脚本执行时间以确保导出完成
set_time_limit(0);
// 输出Excel文件头,可把user.csv换成你要的文件名
header('Content-Type: application/vnd.ms-excel');
header('Content-Disposition: attachment;filename="amazon_product_quantity.csv"');
header('Cache-Control: max-age=0');
 
// 从数据库中获取数据,为了节省内存,不要把数据一次性读到内存,从句柄中一行一行读即可
$sql = 'select * from amazon_product_quantity';
$stmt = mysql_query($sql);
 
// 打开PHP文件句柄,php://output 表示直接输出到浏览器
$fp = fopen('php://output', 'a');
 
// 输出Excel列名信息
$head = array('code', 'sub_code', 'sp_code', 'quantity');
foreach ($head as $i => $v) {
 // CSV的Excel支持GBK编码,一定要转换,否则乱码
 $head[$i] = iconv('utf-8', 'gb2312', $v);
}
 
// 将数据通过fputcsv写到文件句柄
fputcsv($fp, $head);
 
// 计数器
$cnt = 0;
// 每隔$limit行,刷新一下输出buffer,不要太大,也不要太小
$limit = 100000;
 
// 逐行取出数据,不浪费内存
while ($row = mysql_fetch_array($stmt,MYSQL_ASSOC)) {
 $cnt ++;
 if ($limit == $cnt) { //刷新一下输出buffer,防止由于数据过多造成问题
 ob_flush();
 flush();
 $cnt = 0;
 }
 
 foreach ($row as $i => $v) {
 $row[$i] = iconv('utf-8', 'gb2312', $v);
 }
 fputcsv($fp, $row); 
}
?>

ssh scp的使用

1. ssh 命令scp 上传

scp /path/filename username@servername:/path/

例如scp /var/www/test.php root@192.168.0.101:/var/www/ 把本机/var/www/目录下的test.php文件上传到192.168.0.101这台服务器上的/var/www/目录中

2.ssh 命令scp下载

下载文件我们经常使用wget,但是如果没有http服务,如何从服务器上下载文件呢?

scp username@servername:/path/filename /var/www/local_dir(本地目录)

例如scp root@192.168.0.101:/var/www/test.txt 把192.168.0.101上的/var/www/test.txt 的文件下载到/var/www/local_dir(本地目录)

在PHP中使用协程实现多任务调度

本文转自: http://www.laruence.com/2015/05/28/3038.html 

PHP5.5一个比较好的新功能是加入了对迭代生成器和协程的支持.对于生成器,PHP的文档和各种其他的博客文章已经有了非常详细的讲解.协程相对受到的关注就少了,因为协程虽然有很强大的功能但相对比较复杂, 也比较难被理解,解释起来也比较困难.

这篇文章将尝试通过介绍如何使用协程来实施任务调度, 来解释在PHP中的协程.

我将在前三节做一个简单的背景介绍.如果你已经有了比较好的基础,可以直接跳到“协同多任务处理”一节.

迭代生成器

(迭代)生成器也是一个函数,不同的是这个函数的返回值是依次返回,而不是只返回一个单独的值.或者,换句话说,生成器使你能更方便的实现了迭代器接口.下面通过实现一个xrange函数来简单说明:

 

  <?php
 function xrange($start, $end, $step = 1) {
   for ($i = $start; $i <= $end; $i += $step) {
     yield $i;
   }
 }
 
 foreach (xrange(1, 1000000) as $num) {
     echo $num, "\n";
 }

上面这个xrange()函数提供了和PHP的内建函数range()一样的功能.但是不同的是range()函数返回的是一个包含值从1到100万0的数组(注:请查看手册). 而xrange()函数返回的是依次输出这些值的一个迭代器, 而不会真正以数组形式返回.

这种方法的优点是显而易见的.它可以让你在处理大数据集合的时候不用一次性的加载到内存中.甚至你可以处理无限大的数据流.

当然,也可以不同通过生成器来实现这个功能,而是可以通过继承Iterator接口实现.但通过使用生成器实现起来会更方便,不用再去实现iterator接口中的5个方法了.

生成器为可中断的函数

要从生成器认识协程, 理解它内部是如何工作是非常重要的: 生成器是一种可中断的函数, 在它里面的yield构成了中断点.

还是看上面的例子, 调用xrange(1,1000000)的时候, xrange()函数里代码其实并没有真正地运行. 它只是返回了一个迭代器:

  1. <?php
  2. $range = xrange(1, 1000000);
  3. var_dump($range); // object(Generator)#1
  4. var_dump($range instanceof Iterator); // bool(true)
  5. ?>

这也解释了为什么xrange叫做迭代生成器, 因为它返回一个迭代器, 而这个迭代器实现了Iterator接口.

调用迭代器的方法一次, 其中的代码运行一次.例如, 如果你调用$range->rewind(), 那么xrange()里的代码就会运行到控制流第一次出现yield的地方. 而函数内传递给yield语句的返回值可以通过$range->current()获取.

为了继续执行生成器中yield后的代码, 你就需要调用$range->next()方法. 这将再次启动生成器, 直到下一次yield语句出现. 因此,连续调用next()和current()方法, 你就能从生成器里获得所有的值, 直到再没有yield语句出现.

对xrange()来说, 这种情形出现在$i超过$end时. 在这中情况下, 控制流将到达函数的终点,因此将不执行任何代码.一旦这种情况发生,vaild()方法将返回假, 这时迭代结束.

协程

协程的支持是在迭代生成器的基础上, 增加了可以回送数据给生成器的功能(调用者发送数据给被调用的生成器函数). 这就把生成器到调用者的单向通信转变为两者之间的双向通信.

传递数据的功能是通过迭代器的send()方法实现的. 下面的logger()协程是这种通信如何运行的例子:

  1. <?php
  2. function logger($fileName) {
  3.     $fileHandle = fopen($fileName, ‘a’);
  4.     while (true) {
  5.         fwrite($fileHandle, yield . “\n”);
  6.     }
  7. }
  8. $logger = logger(__DIR__ . ‘/log’);
  9. $logger->send(‘Foo’);
  10. $logger->send(‘Bar’)
  11. ?>

正如你能看到,这儿yield没有作为一个语句来使用, 而是用作一个表达式, 即它能被演化成一个值. 这个值就是调用者传递给send()方法的值. 在这个例子里, yield表达式将首先被”Foo”替代写入Log, 然后被”Bar”替代写入Log.

上面的例子里演示了yield作为接受者, 接下来我们看如何同时进行接收和发送的例子:

  1. <?php
  2. function gen() {
  3.     $ret = (yield ‘yield1’);
  4.     var_dump($ret);
  5.     $ret = (yield ‘yield2’);
  6.     var_dump($ret);
  7. }
  8. $gen = gen();
  9. var_dump($gen->current()); // string(6) “yield1”
  10. var_dump($gen->send(‘ret1’)); // string(4) “ret1” (the first var_dump in gen)
  11.                               // string(6) “yield2” (the var_dump of the ->send() return value)
  12. var_dump($gen->send(‘ret2’)); // string(4) “ret2” (again from within gen)
  13.                               // NULL (the return value of ->send())
  14. ?>

要很快的理解输出的精确顺序可能稍微有点困难, 但你确定要搞清楚为什按照这种方式输出. 以便后续继续阅读.

另外, 我要特别指出的有两点:

第一点,yield表达式两边的括号在PHP7以前不是可选的, 也就是说在PHP5.5和PHP5.6中圆括号是必须的.

第二点,你可能已经注意到调用current()之前没有调用rewind().这是因为生成迭代对象的时候已经隐含地执行了rewind操作.

多任务协作

如果阅读了上面的logger()例子, 你也许会疑惑“为了双向通信我为什么要使用协程呢?我完全可以使用其他非协程方法实现同样的功能啊?”, 是的, 你是对的, 但上面的例子只是为了演示了基本用法, 这个例子其实并没有真正的展示出使用协程的优点.

正如上面介绍里提到的,协程是非常强大的概念,不过却应用的很稀少而且常常十分复杂.要给出一些简单而真实的例子很难.

在这篇文章里,我决定去做的是使用协程实现多任务协作.我们要解决的问题是你想并发地运行多任务(或者“程序”).不过我们都知道CPU在一个时刻只能运行一个任务(不考虑多核的情况).因此处理器需要在不同的任务之间进行切换,而且总是让每个任务运行 “一小会儿”.

多任务协作这个术语中的“协作”很好的说明了如何进行这种切换的:它要求当前正在运行的任务自动把控制传回给调度器,这样就可以运行其他任务了. 这与“抢占”多任务相反, 抢占多任务是这样的:调度器可以中断运行了一段时间的任务, 不管它喜欢还是不喜欢. 协作多任务在Windows的早期版本(windows95)和Mac OS中有使用, 不过它们后来都切换到使用抢先多任务了. 理由相当明确:如果你依靠程序自动交出控制的话, 那么一些恶意的程序将很容易占用整个CPU, 不与其他任务共享.

现在你应当明白协程和任务调度之间的关系:yield指令提供了任务中断自身的一种方法, 然后把控制交回给任务调度器. 因此协程可以运行多个其他任务. 更进一步来说, yield还可以用来在任务和调度器之间进行通信.

为了实现我们的多任务调度, 首先实现“任务” — 一个用轻量级的包装的协程函数:

  1. <?php
  2. class Task {
  3.     protected $taskId;
  4.     protected $coroutine;
  5.     protected $sendValue = null;
  6.     protected $beforeFirstYield = true;
  7.     public function __construct($taskId, Generator $coroutine) {
  8.         $this->taskId = $taskId;
  9.         $this->coroutine = $coroutine;
  10.     }
  11.     public function getTaskId() {
  12.         return $this->taskId;
  13.     }
  14.     public function setSendValue($sendValue) {
  15.         $this->sendValue = $sendValue;
  16.     }
  17.     public function run() {
  18.         if ($this->beforeFirstYield) {
  19.             $this->beforeFirstYield = false;
  20.             return $this->coroutine->current();
  21.         } else {
  22.             $retval = $this->coroutine->send($this->sendValue);
  23.             $this->sendValue = null;
  24.             return $retval;
  25.         }
  26.     }
  27.     public function isFinished() {
  28.         return !$this->coroutine->valid();
  29.     }
  30. }

如代码, 一个任务就是用任务ID标记的一个协程(函数). 使用setSendValue()方法, 你可以指定哪些值将被发送到下次的恢复(在之后你会了解到我们需要这个), run()函数确实没有做什么, 除了调用send()方法的协同程序, 要理解为什么添加了一个 beforeFirstYieldflag变量, 需要考虑下面的代码片段:

  1. <?php
  2. function gen() {
  3.     yield ‘foo’;
  4.     yield ‘bar’;
  5. }
  6. $gen = gen();
  7. var_dump($gen->send(‘something’));
  8. // 如之前提到的在send之前, 当$gen迭代器被创建的时候一个renwind()方法已经被隐式调用
  9. // 所以实际上发生的应该类似:
  10. //$gen->rewind();
  11. //var_dump($gen->send(‘something’));
  12. //这样renwind的执行将会导致第一个yield被执行, 并且忽略了他的返回值.
  13. //真正当我们调用yield的时候, 我们得到的是第二个yield的值! 导致第一个yield的值被忽略.
  14. //string(3) “bar”

通过添加 beforeFirstYieldcondition 我们可以确定第一个yield的值能被正确返回.

调度器现在不得不比多任务循环要做稍微多点了, 然后才运行多任务:

  1. <?php
  2. class Scheduler {
  3.     protected $maxTaskId = 0;
  4.     protected $taskMap = []; // taskId => task
  5.     protected $taskQueue;
  6.     public function __construct() {
  7.         $this->taskQueue = new SplQueue();
  8.     }
  9.     public function newTask(Generator $coroutine) {
  10.         $tid = ++$this->maxTaskId;
  11.         $task = new Task($tid, $coroutine);
  12.         $this->taskMap[$tid] = $task;
  13.         $this->schedule($task);
  14.         return $tid;
  15.     }
  16.     public function schedule(Task $task) {
  17.         $this->taskQueue->enqueue($task);
  18.     }
  19.     public function run() {
  20.         while (!$this->taskQueue->isEmpty()) {
  21.             $task = $this->taskQueue->dequeue();
  22.             $task->run();
  23.             if ($task->isFinished()) {
  24.                 unset($this->taskMap[$task->getTaskId()]);
  25.             } else {
  26.                 $this->schedule($task);
  27.             }
  28.         }
  29.     }
  30. }
  31. ?>

newTask()方法(使用下一个空闲的任务id)创建一个新任务,然后把这个任务放入任务map数组里. 接着它通过把任务放入任务队列里来实现对任务的调度. 接着run()方法扫描任务队列, 运行任务.如果一个任务结束了, 那么它将从队列里删除, 否则它将在队列的末尾再次被调度.

让我们看看下面具有两个简单(没有什么意义)任务的调度器:

  1. <?php
  2. function task1() {
  3.     for ($i = 1; $i <= 10; ++$i) {
  4.         echo “This is task 1 iteration $i.\n”;
  5.         yield;
  6.     }
  7. }
  8. function task2() {
  9.     for ($i = 1; $i <= 5; ++$i) {
  10.         echo “This is task 2 iteration $i.\n”;
  11.         yield;
  12.     }
  13. }
  14. $scheduler = new Scheduler;
  15. $scheduler->newTask(task1());
  16. $scheduler->newTask(task2());
  17. $scheduler->run();

两个任务都仅仅回显一条信息,然后使用yield把控制回传给调度器.输出结果如下:

  1. This is task 1 iteration 1.
  2. This is task 2 iteration 1.
  3. This is task 1 iteration 2.
  4. This is task 2 iteration 2.
  5. This is task 1 iteration 3.
  6. This is task 2 iteration 3.
  7. This is task 1 iteration 4.
  8. This is task 2 iteration 4.
  9. This is task 1 iteration 5.
  10. This is task 2 iteration 5.
  11. This is task 1 iteration 6.
  12. This is task 1 iteration 7.
  13. This is task 1 iteration 8.
  14. This is task 1 iteration 9.
  15. This is task 1 iteration 10.

输出确实如我们所期望的:对前五个迭代来说,两个任务是交替运行的, 而在第二个任务结束后, 只有第一个任务继续运行.

与调度器之间通信

既然调度器已经运行了, 那么我们来看下一个问题:任务和调度器之间的通信.

我们将使用进程用来和操作系统会话的同样的方式来通信:系统调用.

我们需要系统调用的理由是操作系统与进程相比它处在不同的权限级别上. 因此为了执行特权级别的操作(如杀死另一个进程), 就不得不以某种方式把控制传回给内核, 这样内核就可以执行所说的操作了. 再说一遍, 这种行为在内部是通过使用中断指令来实现的. 过去使用的是通用的int指令, 如今使用的是更特殊并且更快速的syscall/sysenter指令.

我们的任务调度系统将反映这种设计:不是简单地把调度器传递给任务(这样就允许它做它想做的任何事), 我们将通过给yield表达式传递信息来与系统调用通信. 这儿yield即是中断, 也是传递信息给调度器(和从调度器传递出信息)的方法.

为了说明系统调用, 我们对可调用的系统调用做一个小小的封装:

  1. <?php
  2. class SystemCall {
  3.     protected $callback;
  4.     public function __construct(callable $callback) {
  5.         $this->callback = $callback;
  6.     }
  7.     public function __invoke(Task $task, Scheduler $scheduler) {
  8.         $callback = $this->callback;
  9.         return $callback($task, $scheduler);
  10.     }
  11. }

它和其他任何可调用的对象(使用_invoke)一样的运行, 不过它要求调度器把正在调用的任务和自身传递给这个函数.

为了解决这个问题我们不得不微微的修改调度器的run方法:

  1. <?php
  2. public function run() {
  3.     while (!$this->taskQueue->isEmpty()) {
  4.         $task = $this->taskQueue->dequeue();
  5.         $retval = $task->run();
  6.         if ($retval instanceof SystemCall) {
  7.             $retval($task, $this);
  8.             continue;
  9.         }
  10.         if ($task->isFinished()) {
  11.             unset($this->taskMap[$task->getTaskId()]);
  12.         } else {
  13.             $this->schedule($task);
  14.         }
  15.     }
  16. }

第一个系统调用除了返回任务ID外什么都没有做:

  1. <?php
  2. function getTaskId() {
  3.     return new SystemCall(function(Task $task, Scheduler $scheduler) {
  4.         $task->setSendValue($task->getTaskId());
  5.         $scheduler->schedule($task);
  6.     });
  7. }

这个函数设置任务id为下一次发送的值, 并再次调度了这个任务 .由于使用了系统调用, 所以调度器不能自动调用任务, 我们需要手工调度任务(稍后你将明白为什么这么做). 要使用这个新的系统调用的话, 我们要重新编写以前的例子:

  1. <?php
  2. function task($max) {
  3.     $tid = (yield getTaskId()); // <– here’s the syscall!
  4.     for ($i = 1; $i <= $max; ++$i) {
  5.         echo “This is task $tid iteration $i.\n”;
  6.         yield;
  7.     }
  8. }
  9. $scheduler = new Scheduler;
  10. $scheduler->newTask(task(10));
  11. $scheduler->newTask(task(5));
  12. $scheduler->run();
  13. ?>

这段代码将给出与前一个例子相同的输出. 请注意系统调用如何同其他任何调用一样正常地运行, 只不过预先增加了yield.

要创建新的任务, 然后再杀死它们的话, 需要两个以上的系统调用:

  1. <?php
  2. function newTask(Generator $coroutine) {
  3.     return new SystemCall(
  4.         function(Task $task, Scheduler $scheduler) use ($coroutine) {
  5.             $task->setSendValue($scheduler->newTask($coroutine));
  6.             $scheduler->schedule($task);
  7.         }
  8.     );
  9. }
  10. function killTask($tid) {
  11.     return new SystemCall(
  12.         function(Task $task, Scheduler $scheduler) use ($tid) {
  13.             $task->setSendValue($scheduler->killTask($tid));
  14.             $scheduler->schedule($task);
  15.         }
  16.     );
  17. }

killTask函数需要在调度器里增加一个方法:

  1. <?php
  2. public function killTask($tid) {
  3.     if (!isset($this->taskMap[$tid])) {
  4.         return false;
  5.     }
  6.     unset($this->taskMap[$tid]);
  7.     // This is a bit ugly and could be optimized so it does not have to walk the queue,
  8.     // but assuming that killing tasks is rather rare I won’t bother with it now
  9.     foreach ($this->taskQueue as $i => $task) {
  10.         if ($task->getTaskId() === $tid) {
  11.             unset($this->taskQueue[$i]);
  12.             break;
  13.         }
  14.     }
  15.     return true;
  16. }

用来测试新功能的微脚本:

  1. <?php
  2. function childTask() {
  3.     $tid = (yield getTaskId());
  4.     while (true) {
  5.         echo “Child task $tid still alive!\n”;
  6.         yield;
  7.     }
  8. }
  9. function task() {
  10.     $tid = (yield getTaskId());
  11.     $childTid = (yield newTask(childTask()));
  12.     for ($i = 1; $i <= 6; ++$i) {
  13.         echo “Parent task $tid iteration $i.\n”;
  14.         yield;
  15.         if ($i == 3) yield killTask($childTid);
  16.     }
  17. }
  18. $scheduler = new Scheduler;
  19. $scheduler->newTask(task());
  20. $scheduler->run();
  21. ?>

这段代码将打印以下信息:

  1. Parent task 1 iteration 1.
  2. Child task 2 still alive!
  3. Parent task 1 iteration 2.
  4. Child task 2 still alive!
  5. Parent task 1 iteration 3.
  6. Child task 2 still alive!
  7. Parent task 1 iteration 4.
  8. Parent task 1 iteration 5.
  9. Parent task 1 iteration 6.

经过三次迭代以后子任务将被杀死, 因此这就是”Child is still alive”消息结束的时候. 不过你要明白这还不是真正的父子关系. 因为在父任务结束后子任务仍然可以运行, 子任务甚至可以杀死父任务. 可以修改调度器使它具有更层级化的任务结构, 不过这个不是我们这个文章要继续讨论的范围了.

现在你可以实现许多进程管理调用. 例如 wait(它一直等待到任务结束运行时), exec(它替代当前任务)和fork(它创建一个当前任务的克隆). fork非常酷,而 且你可以使用PHP的协程真正地实现它, 因为它们都支持克隆.

让我们把这些留给有兴趣的读者吧,我们来看下一个议题.

非阻塞IO

很明显, 我们的任务管理系统的真正很酷的应用应该是web服务器. 它有一个任务是在套接字上侦听是否有新连接, 当有新连接要建立的时候, 它创建一个新任务来处理新连接.

Web服务器最难的部分通常是像读数据这样的套接字操作是阻塞的. 例如PHP将等待到客户端完成发送为止. 对一个Web服务器来说, 这有点不太高效. 因为服务器在一个时间点上只能处理一个连接.

解决方案是确保在真正对套接字读写之前该套接字已经“准备就绪”. 为了查找哪个套接字已经准备好读或者写了, 可以使用 流选择函数.

首先,让我们添加两个新的 syscall, 它们将等待直到指定socket 准备好:

  1. <?php
  2. function waitForRead($socket) {
  3.     return new SystemCall(
  4.         function(Task $task, Scheduler $scheduler) use ($socket) {
  5.             $scheduler->waitForRead($socket, $task);
  6.         }
  7.     );
  8. }
  9. function waitForWrite($socket) {
  10.     return new SystemCall(
  11.         function(Task $task, Scheduler $scheduler) use ($socket) {
  12.             $scheduler->waitForWrite($socket, $task);
  13.         }
  14.     );
  15. }

这些 syscall 只是在调度器中代理其各自的方法:

  1. <?php
  2. // resourceID => [socket, tasks]
  3. protected $waitingForRead = [];
  4. protected $waitingForWrite = [];
  5. public function waitForRead($socket, Task $task) {
  6.     if (isset($this->waitingForRead[(int) $socket])) {
  7.         $this->waitingForRead[(int) $socket][1][] = $task;
  8.     } else {
  9.         $this->waitingForRead[(int) $socket] = [$socket, [$task]];
  10.     }
  11. }
  12. public function waitForWrite($socket, Task $task) {
  13.     if (isset($this->waitingForWrite[(int) $socket])) {
  14.         $this->waitingForWrite[(int) $socket][1][] = $task;
  15.     } else {
  16.         $this->waitingForWrite[(int) $socket] = [$socket, [$task]];
  17.     }
  18. }

waitingForRead 及 waitingForWrite 属性是两个承载等待的socket 及等待它们的任务的数组. 有趣的部分在于下面的方法,它将检查 socket 是否可用, 并重新安排各自任务:

  1. <?php
  2. protected function ioPoll($timeout) {
  3.     $rSocks = [];
  4.     foreach ($this->waitingForRead as list($socket)) {
  5.         $rSocks[] = $socket;
  6.     }
  7.     $wSocks = [];
  8.     foreach ($this->waitingForWrite as list($socket)) {
  9.         $wSocks[] = $socket;
  10.     }
  11.     $eSocks = []; // dummy
  12.     if (!stream_select($rSocks, $wSocks, $eSocks, $timeout)) {
  13.         return;
  14.     }
  15.     foreach ($rSocks as $socket) {
  16.         list(, $tasks) = $this->waitingForRead[(int) $socket];
  17.         unset($this->waitingForRead[(int) $socket]);
  18.         foreach ($tasks as $task) {
  19.             $this->schedule($task);
  20.         }
  21.     }
  22.     foreach ($wSocks as $socket) {
  23.         list(, $tasks) = $this->waitingForWrite[(int) $socket];
  24.         unset($this->waitingForWrite[(int) $socket]);
  25.         foreach ($tasks as $task) {
  26.             $this->schedule($task);
  27.         }
  28.     }
  29. }

stream_select 函数接受承载读取、写入以及待检查的socket的数组(我们无需考虑最后一类). 数组将按引用传递, 函数只会保留那些状态改变了的数组元素. 我们可以遍历这些数组, 并重新安排与之相关的任务.

为了正常地执行上面的轮询动作, 我们将在调度器里增加一个特殊的任务:

  1. <?php
  2. protected function ioPollTask() {
  3.     while (true) {
  4.         if ($this->taskQueue->isEmpty()) {
  5.             $this->ioPoll(null);
  6.         } else {
  7.             $this->ioPoll(0);
  8.         }
  9.         yield;
  10.     }
  11. }
  12. ?>

需要在某个地方注册这个任务, 例如, 你可以在run()方法的开始增加$this->newTask($this->ioPollTask()). 然后就像其他任务一样每执行完整任务循环一次就执行轮询操作一次(这么做一定不是最好的方法), ioPollTask将使用0秒的超时来调用ioPoll, 也就是stream_select将立即返回(而不是等待).

只有任务队列为空时,我们才使用null超时,这意味着它一直等到某个套接口准备就绪.如果我们没有这么做,那么轮询任务将一而再, 再而三的循环运行, 直到有新的连接建立. 这将导致100%的CPU利用率. 相反, 让操作系统做这种等待会更有效.

现在编写服务器就相对容易了:

  1. <?php
  2. function server($port) {
  3.     echo “Starting server at port $port…\n”;
  4.     $socket = @stream_socket_server(“tcp://localhost:$port”, $errNo, $errStr);
  5.     if (!$socket) throw new Exception($errStr, $errNo);
  6.     stream_set_blocking($socket, 0);
  7.     while (true) {
  8.         yield waitForRead($socket);
  9.         $clientSocket = stream_socket_accept($socket, 0);
  10.         yield newTask(handleClient($clientSocket));
  11.     }
  12. }
  13. function handleClient($socket) {
  14.     yield waitForRead($socket);
  15.     $data = fread($socket, 8192);
  16.     $msg = “Received following request:\n\n$data”;
  17.     $msgLength = strlen($msg);
  18.     $response = <<<res
  19. HTTP/1.1 200 OK\r
  20. ContentType: text/plain\r
  21. ContentLength: $msgLength\r
  22. Connection: close\r
  23. \r
  24. $msg
  25. RES;
  26.     yield waitForWrite($socket);
  27.     fwrite($socket, $response);
  28.     fclose($socket);
  29. }
  30. $scheduler = new Scheduler;
  31. $scheduler->newTask(server(8000));
  32. $scheduler->run();

这段代码实现了接收localhost:8000上的连接, 然后返回发送来的内容作为HTTP响应. 当然它还能处理真正的复杂HTTP请求, 上面的代码片段只是演示了一般性的概念.

你可以使用类似于ab -n 10000 -c 100 localhost:8000/这样命令来测试服务器. 这条命令将向服务器发送10000个请求, 并且其中100个请求将同时到达. 使用这样的数目, 我得到了处于中间的10毫秒的响应时间. 不过还有一个问题:有少数几个请求真正处理的很慢(如5秒), 这就是为什么总吞吐量只有2000请求/秒(如果是10毫秒的响应时间的话, 总的吞吐量应该更像是10000请求/秒)

协程堆栈

如果你试图用我们的调度系统建立更大的系统的话, 你将很快遇到问题:我们习惯了把代码分解为更小的函数, 然后调用它们. 然而, 如果使用了协程的话, 就不能这么做了. 例如,看下面代码:

  1. <?php
  2. function echoTimes($msg, $max) {
  3.     for ($i = 1; $i <= $max; ++$i) {
  4.         echo “$msg iteration $i\n”;
  5.         yield;
  6.     }
  7. }
  8. function task() {
  9.     echoTimes(‘foo’, 10); // print foo ten times
  10.     echo “—\n”;
  11.     echoTimes(‘bar’, 5); // print bar five times
  12.     yield; // force it to be a coroutine
  13. }
  14. $scheduler = new Scheduler;
  15. $scheduler->newTask(task());
  16. $scheduler->run();

这段代码试图把重复循环“输出n次“的代码嵌入到一个独立的协程里,然后从主任务里调用它. 然而它无法运行. 正如在这篇文章的开始所提到的, 调用生成器(或者协程)将没有真正地做任何事情, 它仅仅返回一个对象.这 也出现在上面的例子里:echoTimes调用除了放回一个(无用的)协程对象外不做任何事情.

为了仍然允许这么做,我们需要在这个裸协程上写一个小小的封装.我们将调用它:“协程堆栈”. 因为它将管理嵌套的协程调用堆栈. 这将是通过生成协程来调用子协程成为可能:

  1. $retval = (yield someCoroutine($foo, $bar));

使用yield,子协程也能再次返回值:

  1. yield retval(“I’m a return value!”);

retval函数除了返回一个值的封装外没有做任何其他事情.这个封装将表示它是一个返回值.

  1. <?php
  2. class CoroutineReturnValue {
  3.     protected $value;
  4.     public function __construct($value) {
  5.         $this->value = $value;
  6.     }
  7.     public function getValue() {
  8.         return $this->value;
  9.     }
  10. }
  11. function retval($value) {
  12.     return new CoroutineReturnValue($value);
  13. }

为了把协程转变为协程堆栈(它支持子调用),我们将不得不编写另外一个函数(很明显,它是另一个协程):

  1. <?php
  2. function stackedCoroutine(Generator $gen) {
  3.     $stack = new SplStack;
  4.     for (;;) {
  5.         $value = $gen->current();
  6.         if ($value instanceof Generator) {
  7.             $stack->push($gen);
  8.             $gen = $value;
  9.             continue;
  10.         }
  11.         $isReturnValue = $value instanceof CoroutineReturnValue;
  12.         if (!$gen->valid() || $isReturnValue) {
  13.             if ($stack->isEmpty()) {
  14.                 return;
  15.             }
  16.             $gen = $stack->pop();
  17.             $gen->send($isReturnValue ? $value->getValue() : NULL);
  18.             continue;
  19.         }
  20.         $gen->send(yield $gen->key() => $value);
  21.     }
  22. }

这个函数在调用者和当前正在运行的子协程之间扮演着简单代理的角色.在$gen->send(yield $gen->key()=>$value);这行完成了代理功能.另外它检查返回值是否是生成器,万一是生成器的话,它将开始运行这个生成器,并把前一个协程压入堆栈里.一旦它获得了CoroutineReturnValue的话,它将再次请求堆栈弹出,然后继续执行前一个协程.

为了使协程堆栈在任务里可用,任务构造器里的$this-coroutine =$coroutine;这行需要替代为$this->coroutine = StackedCoroutine($coroutine);.

现在我们可以稍微改进上面web服务器例子:把wait+read(和wait+write和warit+accept)这样的动作分组为函数.为了分组相关的 功能,我将使用下面类:

  1. <?php
  2. class CoSocket {
  3.     protected $socket;
  4.     public function __construct($socket) {
  5.         $this->socket = $socket;
  6.     }
  7.     public function accept() {
  8.         yield waitForRead($this->socket);
  9.         yield retval(new CoSocket(stream_socket_accept($this->socket, 0)));
  10.     }
  11.     public function read($size) {
  12.         yield waitForRead($this->socket);
  13.         yield retval(fread($this->socket, $size));
  14.     }
  15.     public function write($string) {
  16.         yield waitForWrite($this->socket);
  17.         fwrite($this->socket, $string);
  18.     }
  19.     public function close() {
  20.         @fclose($this->socket);
  21.     }
  22. }

现在服务器可以编写的稍微简洁点了:

  1. <?php
  2. function server($port) {
  3.     echo “Starting server at port $port…\n”;
  4.     $socket = @stream_socket_server(“tcp://localhost:$port”, $errNo, $errStr);
  5.     if (!$socket) throw new Exception($errStr, $errNo);
  6.     stream_set_blocking($socket, 0);
  7.     $socket = new CoSocket($socket);
  8.     while (true) {
  9.         yield newTask(
  10.             handleClient(yield $socket->accept())
  11.         );
  12.     }
  13. }
  14. function handleClient($socket) {
  15.     $data = (yield $socket->read(8192));
  16.     $msg = “Received following request:\n\n$data”;
  17.     $msgLength = strlen($msg);
  18.     $response = <<<res
  19. HTTP/1.1 200 OK\r
  20. ContentType: text/plain\r
  21. ContentLength: $msgLength\r
  22. Connection: close\r
  23. \r
  24. $msg
  25. RES;
  26.     yield $socket->write($response);
  27.     yield $socket->close();
  28. }

错误处理

作为一个优秀的程序员, 相信你已经察觉到上面的例子缺少错误处理. 几乎所有的 socket 都是易出错的. 我没有这样做的原因一方面固然是因为错误处理的乏味(特别是 socket), 另一方面也在于它很容易使代码体积膨胀.

不过, 我仍然想讲下常见的协程错误处理:协程允许使用 throw() 方法在其内部抛出一个错误.

throw() 方法接受一个 Exception, 并将其抛出到协程的当前悬挂点, 看看下面代码:

  1. <?php
  2. function gen() {
  3.     echo “Foo\n”;
  4.     try {
  5.         yield;
  6.     } catch (Exception $e) {
  7.         echo “Exception: {$e->getMessage()}\n”;
  8.     }
  9.     echo “Bar\n”;
  10. }
  11. $gen = gen();
  12. $gen->rewind(); // echos “Foo”
  13. $gen->throw(new Exception(‘Test’)); // echos “Exception: Test”
  14.                                     // and “Bar”

这非常好, 有没有? 因为我们现在可以使用系统调用以及子协程调用异常抛出了.

不过我们要对系统调用Scheduler::run() 方法做一些小调整:

  1. <?php
  2. if ($retval instanceof SystemCall) {
  3.     try {
  4.         $retval($task, $this);
  5.     } catch (Exception $e) {
  6.         $task->setException($e);
  7.         $this->schedule($task);
  8.     }
  9.     continue;
  10. }

Task 类也要添加 throw 调用处理:

  1. <?php
  2. class Task {
  3.     // …
  4.     protected $exception = null;
  5.     public function setException($exception) {
  6.         $this->exception = $exception;
  7.     }
  8.     public function run() {
  9.         if ($this->beforeFirstYield) {
  10.             $this->beforeFirstYield = false;
  11.             return $this->coroutine->current();
  12.         } elseif ($this->exception) {
  13.             $retval = $this->coroutine->throw($this->exception);
  14.             $this->exception = null;
  15.             return $retval;
  16.         } else {
  17.             $retval = $this->coroutine->send($this->sendValue);
  18.             $this->sendValue = null;
  19.             return $retval;
  20.         }
  21.     }
  22.     // …
  23. }

现在, 我们已经可以在系统调用中使用异常抛出了!例如,要调用 killTask,让我们在传递 ID 不可用时抛出一个异常:

  1. <?php
  2. function killTask($tid) {
  3.     return new SystemCall(
  4.         function(Task $task, Scheduler $scheduler) use ($tid) {
  5.             if ($scheduler->killTask($tid)) {
  6.                 $scheduler->schedule($task);
  7.             } else {
  8.                 throw new InvalidArgumentException(‘Invalid task ID!’);
  9.             }
  10.         }
  11.     );
  12. }

试试看:

  1. <?php
  2. function task() {
  3.     try {
  4.         yield killTask(500);
  5.     } catch (Exception $e) {
  6.         echo ‘Tried to kill task 500 but failed: ‘, $e->getMessage(), “\n”;
  7.     }
  8. }

这些代码现在尚不能正常运作,因为 stackedCoroutine 函数无法正确处理异常.要修复需要做些调整:

  1. <?php
  2. function stackedCoroutine(Generator $gen) {
  3.     $stack = new SplStack;
  4.     $exception = null;
  5.     for (;;) {
  6.         try {
  7.             if ($exception) {
  8.                 $gen->throw($exception);
  9.                 $exception = null;
  10.                 continue;
  11.             }
  12.             $value = $gen->current();
  13.             if ($value instanceof Generator) {
  14.                 $stack->push($gen);
  15.                 $gen = $value;
  16.                 continue;
  17.             }
  18.             $isReturnValue = $value instanceof CoroutineReturnValue;
  19.             if (!$gen->valid() || $isReturnValue) {
  20.                 if ($stack->isEmpty()) {
  21.                     return;
  22.                 }
  23.                 $gen = $stack->pop();
  24.                 $gen->send($isReturnValue ? $value->getValue() : NULL);
  25.                 continue;
  26.             }
  27.             try {
  28.                 $sendValue = (yield $gen->key() => $value);
  29.             } catch (Exception $e) {
  30.                 $gen->throw($e);
  31.                 continue;
  32.             }
  33.             $gen->send($sendValue);
  34.         } catch (Exception $e) {
  35.             if ($stack->isEmpty()) {
  36.                 throw $e;
  37.             }
  38.             $gen = $stack->pop();
  39.             $exception = $e;
  40.         }
  41.     }
  42. }

结束语

在这篇文章里,我使用多任务协作构建了一个任务调度器, 其中包括执行“系统调用”, 做非阻塞操作和处理错误. 所有这些里真正很酷的事情是任务的结果代码看起来完全同步, 甚至任务正在执行大量的异步操作的时候也是这样.

如果你打算从套接口读取数据的话, 你将不需要传递某个回调函数或者注册一个事件侦听器. 相反, 你只要书写yield $socket->read(). 这儿大部分都是你常常也要编写的,只 在它的前面增加yield.

当我第一次听到协程的时候, 我发现这个概念完全令人折服, 正是因为这个激励我在PHP中实现了它. 同时我发现协程真正非常的令人惊叹:在令人敬畏的代码和一大堆乱代码之间只有一线之隔, 我认为协程恰好处在这条线上, 不多不少. 不过, 要说使用上面所述的方法书写异步代码是否真的有益, 这个就见仁见智了.

但, 不管咋样, 我认为这是一个有趣的话题, 而且我希望你也能找到它的乐趣. 欢迎评论:)


	

UUID

百度百科的链接:https://baike.baidu.com/item/UUID/5921266

UUID 的目的是让分布式系统中的所有元素,都能有唯一的辨识资讯。

UUID 能用完吗?UUID是由一组32位数的16进制数字所构成,是故UUID理论上的总数为16^32=2^128,约等于3.4 x 10^38。也就是说若每纳秒产生1兆个UUID,要花100亿年才会将所有UUID用完。

UUID的标准型式包含32个16进制数字,以连字号分为五段,形式为8-4-4-4-12的32个字符。示例:
  • 550e8400-e29b-41d4-a716-446655440000
    PHP 怎么产生呢?

    function uuid($prefix = ''){
       $chars = md5(uniqid(mt_rand(), true));
       $uuid  = substr($chars,0,8) . '-';
       $uuid .= substr($chars,8,4) . '-';
       $uuid .= substr($chars,12,4) . '-';
       $uuid .= substr($chars,16,4) . '-';
       $uuid .= substr($chars,20,12);
       return $prefix . $uuid;
    }

    我平时习惯这样产生32位的:

    function actId($attach=''){
        $str = uniqid(rand(), true);
        $str .= $attach;
        return md5($str);
    }

    我们平时做API的时候经常会用到token,下面方法会生成一个40位的token

    function setToken(){
       $str = md5(uniqid(md5(microtime(true)),true));
       $str = sha1($str);
       return $str;
    }