前言
Swoole官方文档:Swoole 文档
Swoole 使 PHP 开发人员可以编写高性能高并发的 TCP、UDP、Unix Socket、HTTP、 WebSocket 等服务,让 PHP 不再局限于 Web 领域。Swoole4 协程的成熟将 PHP 带入了前所未有的时期, 为性能的提升提供了独一无二的可能性。Swoole 可以广泛应用于互联网、移动通信、云计算、 网络游戏、物联网(IOT)、车联网、智能家居等领域。使用 PHP + Swoole 可以使企业 IT 研发团队的效率大大提升,更加专注于开发创新产品。
Java 在运行前需要先编译,而 PHP 则可以直接将文件丢到服务器上就能运行,这就是解释执行与编译执行的区别。Java这类的语言拥有固定、明确的变量类型,被称为静态语言;而 PHP 这类的语言则可以归结为动态语言,特点是变量不用指定类型。
对于这两种语言编译运行方式来说,类似 Java 语言会将代码一次加载到内存,效率明显会提升不少,毕竟内存和硬盘的速度差距还是蛮大的。而且会一次性将很多初始对象,类模板文件加载,调用的时候不用重新再加载实例化,性能就会有更进一步的上升空间。但是,类似 Java 语言通常都是需要编译成一个可执行的中间文件的,如果有代码的更新,则必须重启整个程序。
解释执行语言优缺点很明显就和编译执行语言是反过来的了。解释执行语言每一次运行一个脚本,就需要将所有相关的文件全部加载一次,而且如果没别的优化的话(比如 OPcache),所有的相关文件都要从硬盘读取、加载内存、实例化这些步骤中从头走一遍。可想而知,他的效率和性能是完全无法与静态语言相比的。但是,优点也很明确,随时修改一个文件就可以随时上线,线上业务不用中断。
Swoole 是如何来解决效率性能问题的?它就是通过直接将代码加载到内存的方式,就像 Java 一样来启动一个进程,实现 PHP 代码的高性能执行。同时,尽量保持代码还是可以按照传统的方式来写,为 PHP 提供了一个高性能的解决方案。
安装
初学者建议直接在宝塔上安装PHP环境及Swoole拓展。注意:swoole程序只能在Linux上运行。
HTTP 服务器
创建http_server.php文件。
<?php$server = new swoole\Http\Server('0.0.0.0',1234);$server->set(['enable_coroutine'=>true]);// $server->on 以上的代码块在程序启动时会执行一次,且后续请求不会再触发,可以放上框架的初始化代码,这样就可以做到只在程序启动时初始化。$server->on('Request',function($request,$response){ // 有请求进入会执行此代码块static $a = 1;$a++;list($controller,$action) = explode('/',trim($request->server['request_uri'],'/'));$response->header('Content-Tpye','text/html;charset=utf-8');$response->end("<h1>hello!Swoole.controller is {$controller},action is {$action},a is {$a}</h1>");return;});$server->start();?>
将文件上传到服务器后,在目录下执行 php http_server.php 以启动服务。
浏览器访问 localhost:1234/index/test , 会得到以下输出:
hello!Swoole.controller is index,action is test,a is 2
可以发现,在不断请求接口下,$a 的值会一直累加(常驻内存),而不像之前传统的PHP开发中,变量每次都会被释放掉。
Coroutine 协程创建
<?php
$server = new swoole\Http\Server('0.0.0.0',1234);
$server->set(['enable_coroutine'=>true]);
$server->on('Request',function($request,$response){Coroutine::create(function(){Coroutine::sleep(2);var_dump('协程1');var_dump(time());});Coroutine::create(function(){Coroutine::sleep(2);var_dump('协程2');var_dump(time());});});
$server->start();?>
在服务器上执行 php http_server.php 重启服务,并访问之前的地址,控制台输出如下。
[root@VM-12-13-centos swoole]# php server_test.php
string(7) "协程1"
int(1709349803)
string(7) "协程2"
int(1709349803)
可以发现,打印出来的时间并没有相差两秒,协程间是同步执行的,并不会进行阻塞。而在rpm模式代码是从上往下同步执行的。
Websocket服务
创建 Websocket_test.php 文件,并上传到服务器。
<?php// 初始化 WebSocket 服务器,在本地监听 1234端口
$server = new Swoole\WebSocket\Server("localhost", 1234);// 建立连接时触发
$server->on('open', function (Swoole\WebSocket\Server $server, $request) {echo "server: handshake success with fd{$request->fd}\n";
});// 收到消息时触发推送
$server->on('message', function (Swoole\WebSocket\Server $server, $frame) {echo "receive from {$frame->fd}:{$frame->data},opcode:{$frame->opcode},fin:{$frame->finish}\n";$server->push($frame->fd, "from:{$frame->fd}:{$frame->data}");
});// 关闭 WebSocket 连接时触发
$server->on('close', function ($ser, $fd) {echo "client {$fd} closed\n";
});// 启动 WebSocket 服务器
$server->start();?>
本地创建 websocket_client.html 。
<!DOCTYPE html>
<html>
<head><meta charset="UTF-8"><title>Chat Client</title>
</head>
<body>
<script>window.onload = function () {var nick = prompt("Enter your nickname");var input = document.getElementById("input");input.focus();// 初始化客户端套接字并建立连接var socket = new WebSocket("ws://ip:1234");// 连接建立时触发socket.onopen = function (event) {console.log("Connection open ..."); }// 接收到服务端推送时执行socket.onmessage = function (event) {var msg = event.data;var node = document.createTextNode(msg);var div = document.createElement("div");div.appendChild(node);document.body.insertBefore(div, input);input.scrollIntoView();};// 连接关闭时触发socket.onclose = function (event) {console.log("Connection closed ..."); }input.onchange = function () {var msg = nick + ": " + input.value;// 将输入框变更信息通过 send 方法发送到服务器socket.send(msg);input.value = "";};}
</script>
<input id="input" style="width: 100%;">
</body>
</html>
在 Websocket_test.php 所在目录执行 php Websocket_test.php 以启动服务。本地打开websocket_client.html网页。
输入昵称后,在输入框发送消息。
控制台输出:
[root@VM-55-23-centos swoole]# php websocket_test.php
server: handshake success with fd1
receive from 1:Hola: hello,world~,opcode:1,fin:1
receive from 1:Hola: 你好,opcode:1,fin:1
执行异步任务 (Task)*
在 Server 程序中如果需要执行很耗时的操作,比如一个聊天服务器发送广播,Web 服务器中发送邮件。如果直接去执行这些函数就会阻塞当前进程,导致服务器响应变慢。
Swoole 提供了异步任务处理的功能,可以投递一个异步任务到 TaskWorker 进程池中执行,不影响当前请求的处理速度。
创建 task_test.php 文件,并上传到服务器。
<?php$serv = new Swoole\Server('127.0.0.1', 1234);// 设置工作进程数量。
$serv->set(['work_num' =>2, // worker_num是用来处理请求逻辑的进程数'task_worker_num' => 2 // task_num是异步任务投递进程,专门处理异步任务的执行,类似 fast-cgi
]);// 接收到数据时回调此函数,此回调函数在worker进程中执行。
$serv->on('Receive', function($serv, $fd, $reactor_id, $data) {//投递异步任务$task_id = $serv->task($data);// 线程间通信,向其他工作进程发送消息$serv->sendMessage('hello task process',$serv->worker_id);
});// 当工作进程收到由 $server->sendMessage() 发送的消息时会触发 onPipeMessage 事件。worker/task 进程都可能会触发 onPipeMessage 事件
$serv->on('pipeMessage',function($serv,$src_worker_id,$data){echo "#{$serv->worker_id} message from #$src_worker_id: $data\n";
})// 处理异步任务(此回调函数在task进程中执行)。
$serv->on('Task', function ($serv, $task_id, $reactor_id, $data) {echo "New AsyncTask[id={$task_id}]";// 返回任务执行的结果$serv->finish("{$data} -> OK");
});//处理异步任务的结果(此回调函数在worker进程中执行)。
$serv->on('Finish', function ($serv, $task_id, $data) {echo "AsyncTask[{$task_id}] Finish: {$data}".PHP_EOL;
});$serv->start();?>
使用 php task_test.php 运行后,再另开窗口使用 telnet 127.0.0.1 1234 连接此 TCP 服务,并发送消息。
[root@VM-55-23-centos ~]# telnet 127.0.0.1 1234
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
hello,world // 发送的消息
回到swoole程序窗口,可以看到当服务接收到数据(onReceive)后,会向task投递异步任务,在onTask中处理任务。
[root@VM-55-23-centos swoole]# php task_test.php
Dispatch AsyncTask: id=0
New AsyncTask[id=0]
AsyncTask[0] Finish: hello,world-> OK
调用 $serv->task()
后,程序立即返回,继续向下执行代码。onTask 回调函数 Task 进程池内被异步执行。执行完成后调用 $serv->finish()
返回结果。
更多事件参考:事件 | Swoole4 文档
当我们使用 ps -aux | grep task_test 命令查看进程:
- swoole启动的主进程是master进程负责全局管理,然后master进程会再fork一个manager进程。
- manager进程开始统一管理进程创建回收管理。
- manager进程根据设置的worker_num和task_worker_num来创建work进程和task进程。
因此启动swoole我们能看到的进程数是:2+worker_num+task_worker_num,2中包含manager进程和master进程
毫秒定时器
创建 timer_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine;// 创建协程容器
Coroutine\run(function(){// 创建定时器,2000ms执行一次,一直执行Swoole\Timer::tick(2000,function(int $timer_id , $parma1 , $parma2){echo "timer by tick,timer id is #$timer_id , after 2000ms , parma1:$parma1 , parma2:$parma2,".PHP_EOL;// 在定时器中创建定时器,4000ms执行一次,一直执行Swoole\Timer::tick(4000,function(int $timer_id){echo "timer tick by timer tick,timer id is {$timer_id} , after 4000ms,".PHP_EOL;// 清除指定id的定时器Swoole\Timer::clear($timer_id);});},"A","B");// 创建定时器,3000ms执行一次,只会执行一次Swoole\Timer::after(3000,function(){echo "timer tick by after , after 3000ms,".PHP_EOL;});// 获取定时器列表,循环输出定时器信息foreach(Swoole\Timer::list() as $timer_id){var_dump("timer info:");var_dump(Swoole\Timer::info($timer_id));};// 清除所有定时器// Swoole\Timer::clearAll();});?>
执行 php timer_server.php 以启动服务。
输出如下:
string(11) "timer info:"
array(5) {["exec_msec"]=>int(3000)["exec_count"]=>int(0)["interval"]=>int(0)["round"]=>int(0)["removed"]=>bool(false)
}
string(11) "timer info:"
array(5) {["exec_msec"]=>int(2000)["exec_count"]=>int(0)["interval"]=>int(2000)["round"]=>int(0)["removed"]=>bool(false)
}
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by after , after 3000ms,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by timer tick,timer id is 3 , after 4000ms,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by timer tick,timer id is 4 , after 4000ms,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by timer tick,timer id is 5 , after 4000ms,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by timer tick,timer id is 6 , after 4000ms,
timer by tick,timer id is #1 , after 2000ms , parma1:A , parma2:B,
timer tick by timer tick,timer id is 7 , after 4000ms,
定时器常用方法:
// 设置一个间隔时钟定时器。
Swoole\Timer::tick(int $msec, callable $callback_function, ...$params): int// 在指定的时间后执行函数。Swoole\Timer::after 函数是一个一次性定时器,执行完成后就会销毁。
Swoole\Timer::after(int $msec, callable $callback_function, ...$params): int// 使用定时器 ID 来删除定时器。
Swoole\Timer::clear(int $timer_id): bool// 清除当前 Worker 进程内的所有定时器。
Swoole\Timer::clearAll(): bool// 返回 timer 的信息。
Swoole\Timer::info(int $timer_id): array// 返回定时器迭代器,可使用 foreach 遍历当前 Worker 进程内所有 timer 的 id
Swoole\Timer::list(): Swoole\Timer\Iterator
高性能共享内存 Table
创建 timer_test.php 文件,并上传到服务器。
<?php
// 创建内存表并设置表大小、表字段
$table = new Swoole\Table(256);
$table->column('id', Swoole\Table::TYPE_INT);
$table->column('name', Swoole\Table::TYPE_STRING, 64);
$table->create();$serv = new Swoole\Server('127.0.0.1', 1234);// 设置数据包分发策略(分发给Woker进程)为轮询模式
$serv->set(['dispatch_mode' => 1]);
$serv->table = $table;// 接收到数据时触发
$serv->on('receive', function ($serv, $id, $reactor_id, $data) {$cmd = explode(" ", trim($data));//getif ($cmd[0] == 'get'){if (count($cmd) < 2){$cmd[1] = $id;}$get_id = intval($cmd[1]);// 从内存表中获取数据$info = $serv->table->get($get_id);$serv->send($id, var_export($info, true)."\n");}//setelseif ($cmd[0] == 'set'){// 往内存表中存放数据$ret = $serv->table->set($id, array('id' => $cmd[1], 'name' => $cmd[2]));if ($ret === false){$serv->send($id, "ERROR\n");}else{$serv->send($id, "OK\n");}}else{$serv->send($id, "command error.\n");}
});$serv->start();?>
使用 php task_test.php 运行后,再另开窗口使用 telnet 127.0.0.1234 连接此 TCP 服务,并发送数据。
[root@VM-55-23-centos ~]# telnet 127.0.0.1 1234
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
get 2
false
set 1 sam
OK
get 1
array ('id' => 1,'name' => 'sam',
)
当服务关闭时,内存表也会被释放。
协程
协程创建的常规方式
创建 coroutine_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine;
use function Swoole\Coroutine\run;// 必须使用run创建协程容器,才能使用协程。框架能直接使用go是因为框架启动时已创建了协程容器
run(function(){// 1.go()创建协程,开启短命名可用(默认开启)go(function(){// 使用协程中的Sleep才不会同步阻塞Coroutine::sleep(2);echo 'this is a coroutine by go'.PHP_EOL;});// 2.Coroutine::create() 原生创建协程Coroutine::create(function(){Coroutine::sleep(2);echo 'this is a coroutine by Coroutine::create'.PHP_EOL;});echo 'first-'.PHP_EOL;
});// 当执行完协程容器的代码块才会执行到这
echo 'end-'.PHP_EOL;?>
执行 php coroutine_test.php 以启动服务,得到以下输出:
first-
this is a coroutine by go
this is a coroutine by Coroutine::create
end-
协程的创建方式:
1.go();
2.Coroutine::create();
协程间是同步执行的,并不会进行阻塞。而在rpm模式代码是从上往下同步执行的。
并发执行协程
并发执行多个协程,并且通过数组,返回这些协程方法的返回值。
<?php
use Swoole\Coroutine;
use function Swoole\Coroutine\run;Coroutine\run(function(){// 并发执行多个协程,并且通过数组,返回这些协程方法的返回值。$result = Coroutine\batch(['name' => function(){Coroutine\System::sleep(2);return 'Hola'; // 返回结果},'area' => function(){Coroutine\System::sleep(2);return 'China'; // 返回结果},]);var_dump($result);
});?>
输出:
array(2) {["name"]=>string(4) "Hola"["area"]=>string(5) "China"
}
协程间通信
Channel
通道,用于协程间通讯,支持多生产者协程和多消费者协程。底层自动实现了协程的切换和调度。
1.通道与 PHP 的 Array 类似,仅占用内存,没有其他额外的资源申请,所有操作均为内存操作,无 IO 消耗
2.底层使用 PHP 引用计数实现,无内存拷贝。即使是传递巨大字符串或数组也不会产生额外性能消耗
3.channel 基于引用计数实现,是零拷贝的
创建 timer_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine;
use function Swoole\Coroutine\run;
use Swoole\Coroutine\Channel;run(function(){// 创建channel(相当于一个队列)$channel = new Channel(1);// 每隔1s向channel中写入一条数据,写五次go(function() use ($channel) {for($i = 0; $i < 5; $i++){Coroutine::sleep(1);//向通道中写入数据$channel->push(['rand' => rand(1000,9999) , 'index' => $i ]);}});// 一直轮询管道,有数据则输出Coroutine::create(function() use ($channel) {while(1){// 从通道中读取数据$data = $channel->pop(1);if($data){var_dump($data);}else{var_dump($channel->errCode === SWOOLE_CHANNEL_TIMEOUT);break;}}});});?>
执行 php channel_test.php 以启动服务,得到以下输出:
array(2) {["rand"]=>int(5746)["index"]=>int(0)
}
array(2) {["rand"]=>int(8235)["index"]=>int(1)
}
array(2) {["rand"]=>int(2584)["index"]=>int(2)
}
array(2) {["rand"]=>int(6474)["index"]=>int(3)
}
array(2) {["rand"]=>int(6893)["index"]=>int(4)
}
array(2) {["rand"]=>int(3986)["index"]=>int(5)
}
WaitGroup
创建 waitGroup_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine;
use Swoole\Coroutine\WaitGroup;
use Swoole\Coroutine\Http\Client;
use function Swoole\Coroutine\run;run(function () {$wg = new WaitGroup();$result = [];$wg->add();//启动第一个协程Coroutine::create(function () use ($wg, &$result) {Coroutine::sleep(2);$result[] = 123;$wg->done();});$wg->add();//启动第二个协程Coroutine::create(function () use ($wg, &$result) {Coroutine::sleep(2);$result[] = 321;$wg->done();});//挂起当前协程,等待所有任务完成后恢复$wg->wait();//这里 $result 包含了 2 个任务执行结果var_dump($result);
});?>
执行 php wiatGroup_test.php 以启动服务,得到以下输出:
array(2) {[0]=>int(123)[1]=>int(321)
}
可以看到,输出内容包含了两个协程的结果。
1.add 方法增加计数
2.done 表示任务已完成
3.wait 等待所有任务完成恢复当前协程的执行
4.WaitGroup 对象可以复用,add、done、wait 之后可以再次使用
Barrier
在 Swoole Library 中底层提供了一个更便捷的协程并发管理工具:
Coroutine\Barrier
协程屏障,或者叫协程栅栏。基于PHP
引用计数和Coroutine API
实现。相比于 Coroutine\WaitGroup,
Coroutine\Barrier
使用更简单一些,只需通过参数传递或者闭包的use
语法,引入子协程函数上即可。
创建 barrier_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine\Barrier;
use function Swoole\Coroutine\run;
use Swoole\Coroutine;run(function () {// 创建一个新的协程屏障$barrier = Barrier::make();$count = 0;Coroutine::create(function () use ($barrier, &$count) {Coroutine::sleep(0.5);$count++;});Coroutine::create(function () use ($barrier, &$count) {Coroutine::sleep(1);$count++;});// 会自动挂起当前协程,等待引用该协程屏障的子协程退出Barrier::wait($barrier);var_dump($count);
});?>
执行 php barrier_test.php 以启动服务,得到以下输出:
int(2)
协程内异常处理
try/catch
在协程编程中可直接使用 try/catch 处理异常。但必须在协程内捕获,不得跨协程捕获异常。
不仅是应用层 throw 的 Exception,底层的一些错误也是可以被捕获的,如 function、class、method 不存在
<?php
use function Swoole\Coroutine\run;
use Swoole\Coroutine;run(function(){Coroutine::create(function () use ($barrier, &$count) {try{// 不存在的方法xxx();}catch(\Throwable $e){var_dump($e->getMessage());}});// 协程1的错误并不会影响协程2Coroutine::create(function () use ($barrier, &$count) {Coroutine::sleep(2);echo 'go go go';});});?>
输出:
string(32) "Call to undefined function xxx()"
go go go
错误必须捕获。
register_shutdown_function
由于 Swoole 是常驻内存的,所以禁止在 Swoole 中使用 exit/die,会导致当前工作的 Worker 进程、Task 进程、User 进程、以及 Swoole\Process 进程会立即退出。
Server 运行期一旦发生致命错误,那客户端连接将无法得到回应。如 Web 服务器,如果有致命错误应当向客户端发送 HTTP 500 错误信息。
在 PHP 中可以通过 register_shutdown_function + error_get_last 2 个函数来捕获致命错误,并将错误信息发送给客户端连接。
致命错误捕获示例代码:
<?php$http = new Swoole\Http\Server('127.0.0.1', 1234);
$http->on('request', function ($request, $response) {register_shutdown_function(function () use ($response) {$error = error_get_last();var_dump($error);switch ($error['type'] ?? null) {case E_ERROR :case E_PARSE :case E_CORE_ERROR :case E_COMPILE_ERROR :// log or send:// error_log($message);// $server->send($fd, $error['message']);$response->status(500);$response->end($error['message']);break;}});exit(0);
});
$http->start();?>
协程调度
用户的每个请求都会创建一个协程,请求结束后协程结束,如果同时有成千上万的并发请求,某一时刻某个进程内部会存在成千上万的协程,那么 CPU 资源是有限的,到底执行哪个协程的代码?
决定到底让 CPU 执行哪个协程的代码的决断过程就是协程调度
,Swoole
的调度策略又是怎么样的呢?
-
首先,在执行某个协程代码的过程中发现这行代码遇到了 Co::sleep() 或者产生了网络 IO,例如 MySQL->query(),这肯定是一个耗时的过程,Swoole 就会把这个 MySQL 连接的 Fd 放到 EventLoop 中。
- 然后让出这个协程的 CPU 给其他协程使用:即 yield(挂起)
- 等待 MySQL 数据返回后再继续执行这个协程:即 resume(恢复)
-
其次,如果协程的代码有 CPU 密集型代码,可以开启 enable_preemptive_scheduler,Swoole 会强行让这个协程让出 CPU。
协程调度创建协程
创建文件 coroutine_scheduler_test.php 文件,并上传到服务器。
<?php
use Swoole\Coroutine;$scheduler = new Coroutine\Scheduler();
// 设置可创建的最大协程数为200
$scheduler->set(['max_coroutine'=>200]);// 添加任务。等待调用 start 方法时,一起启动并执行。
$scheduler->add(function($a,$b){Coroutine\System::sleep(1);var_dump($a);var_dump(time());var_dump($b);var_dump(time());
},'aaa','bbb');// 添加任务。等待调用 start 方法时,一起启动并执行。
$scheduler->add(function($c){Coroutine\System::sleep(1);var_dump($c);var_dump(time());
},'ccc');// 创建并行协程。在 start 时会同时启动 $num 个 $fn 协程,并行地执行。
$scheduler->parallel(2,function($c){Coroutine\System::sleep(1);echo "$c cid is ".Coroutine::getCid().'\n';var_dump(time());
},'ccc');var_dump('start...');
// 启动程序。遍历 add 和 parallel 方法添加的协程任务,并执行。
$scheduler->start();
var_dump('end...');
执行 php coroutine_scheduler_test.php 以启动服务,得到以下输出:
string(8) "start..."
string(3) "aaa"
int(1709353338)
string(3) "bbb"
int(1709353338)
ccc cid is 4
int(1709353338)
ccc cid is 3
int(1709353338)
string(3) "ccc"
int(1709353338)
string(6) "end..."
其中:
1.$scheduler->add(); // 向协程调度中添加一个任务
2.$scheduler->parallel(); // 创建并行协程。在 start 时会同时启动 $num 个 $fn 协程,并行地执行。
3.$scheduler->start(); // 遍历 add 和 parallel 方法添加的协程任务,并执行。
同时可以基于 $scheduler->parallel() 可以做到创建N个协程执行相同的任务:
<?php
use Swoole\Coroutine;
use function Swoole\Coroutine\run;Coroutine\run(function(){$result = [];Coroutine\parallel(3,function() use (&$result){Coroutine\System::sleep(2);$result[] = 333;});var_dump($result);});
输出:
array(3) {[0]=>int(333)[1]=>int(333)[2]=>int(333)
}
协程的挂起和恢复
代码示例:
<?php
use Swoole\Coroutine;function test(){var_dump('this is a test function');
}// 创建协程容器
Coroutine\run(function(){// 协程创建后返回协程id$cid1 = Coroutine::create(function(){echo "co 1 start\n";// 手动让出当前协程的执行权。Coroutine::yield();echo "co 1 end\n";}); // 协程创建后返回协程id$cid2 = Coroutine::create(function(){test();}); var_dump('--------');Coroutine::create(function() use ($cid1,$cid2){var_dump("co 2 start\n");Coroutine::sleep(1);// 唤醒当前协程Coroutine::resume($cid1);echo "co 2 end\n";}); // 会在协程关闭之前 (即协程函数执行完毕时) 进行调用Cortoutine::defer(function(){var_dump('coroutine end');});var_dump('=======');$coList = Coroutine::listCoroutines();foreach($coList as $cid){var_dump(Coroutine::getBackTrace($cid));}var_dump('++++++++');
});?>
启动服务,会得到以下输出:
co 1 start
string(23) "this is a test function"
string(8) "--------"
string(11) "co 2 start
"
string(7) "======="
array(1) {[0]=>array(6) {["line"]=>int(58)["function"]=>string(5) "sleep"["class"]=>string(16) "Swoole\Coroutine"["type"]=>string(2) "::"["args"]=>array(1) {[0]=>int(1)}}
}
array(1) {[0]=>array(6) {["line"]=>int(46)["function"]=>string(5) "yield"["class"]=>string(16) "Swoole\Coroutine"["type"]=>string(2) "::"["args"]=>array(0) {}}
}
array(1) {[0]=>array(6) {["line"]=>int(68)["function"]=>string(12) "getBackTrace"["class"]=>string(16) "Swoole\Coroutine"["type"]=>string(2) "::"["args"]=>array(1) {[0]=>int(1)}}
}
string(8) "++++++++"
co 1 end
co 2 end
yield():手动让出当前协程的执行权。而不是基于 IO 的协程调度。
必须与 Coroutine::resume() 方法成对使用。该协程 yield 以后,必须由其他外部协程 resume,否则将会造成协程泄漏,被挂起的协程永远不会执行。
resume():手动恢复某个协程,使其继续运行,不是基于 IO 的协程调度。当前协程处于挂起状态时,另外的协程中可以使用 resume 再次唤醒当前协程
协程系统杂项函数
<?php
use Swoole\Coroutine;Coroutine\run(function(){Coroutine::create(function(){// 执行一条 shell 指令。底层自动进行协程调度。$ret = Coroutine\System::exec('mkdir test_dir');var_dump($ret);});Coroutine::create(function(){// 将域名解析为 IP。基于同步的线程池模拟实现,底层自动进行协程调度。$ret = Coroutine\system::gethostbyname("www.baidu.com", AF_INET);var_dump($ret);});Coroutine::create(function(){// 进行 DNS 解析,查询域名对应的 IP 地址。$ret = Coroutine\System::getaddrinfo("www.baidu.com");var_dump($ret);});Coroutine::create(function(){// 域名地址查询。$ret = Coroutine\System::dnsLookup("www.baidu.com");var_dump($ret);});
});?>
进程池及进程间通信
进程池,基于 Swoole\Server 的 Manager 管理进程模块实现。可管理多个工作进程。该模块的核心功能为进程管理,相比 Process 实现多进程,Process\Pool 更加简单,封装层次更高。
创建进程池及进程间通信
创建文件 process_pool_test.php,并上传服务器:
<?php
use Swoole\Process;
use Swoole\Coroutine;
use Swoole\Process\Pool;
/*** 创建进程池参数:* 1.设置工作进程数量* 2.设置通信方式* 3.当通信方式为队列(SWOOLE_IPC_MSGQUEUE)时,需要设置队列的key* 4.是否开启协程*/
$pool = new Swoole\Process\Pool(2,SWOOLE_IPC_UNIXSOCK,0, true);// 子进程启动,自动创建协程容器及协程
$pool->on('workerstart', function(Swoole\Process\Pool $pool, int $workerId){var_dump($workerId);// 获取当前工作进程对象。返回 Swoole\Process 对象。$process = $pool->getProcess(0);// 导出socket对象,实现 Worker 进程间通信$socket = $process->exportsocket();if($workerId ==0){var_dump('000');// recv() 暂时挂起,等待数据到来恢复echo $socket->recv();// 向socket发送数据$socket->send("hello proc1\n");echo "proce stop\n";} else {var_dump('111');$socket->send("hello proc0\n");// recv() 暂时挂起,等待数据到来恢复echo $socket->recv();// 向socket发送数据echo "proc1 stop\n";// 不关闭pool的情况下,会有两个协程循环收发消息$pool->shutdown();}
});$pool->start();?>
启动服务,输出如下:
int(0)
string(3) "000"
int(1)
string(3) "111"
hello proc0
proce stop
hello proc1
proc1 stop
Swoole\Process\Pool 一共提供了三种进程间通信的方式:
1.消息队列:SWOOLE_IPC_MSGQUEUE,需设置队列key。
2.Socket 通信:SWOOLE_IPC_SOCKET,如果客户端与服务端不在同一服务器可使用该方式。
3.UnixSocket:SWOOLE_IPC_UNIXSOCKET,推荐
进程管理器
进程管理器,基于 Process\Pool 实现。可以管理多个进程。相比与 Process\Pool ,可以非常方便的创建多个执行不同任务的进程,并且可以控制每一个进程是否要处于协程环境。
use Swoole\Process\Manager;
use Swoole\Process\Pool;$pm = new Manager();for ($i = 0; $i < 2; $i++) {// 增加一个工作进程$pm->add(function (Pool $pool, int $workerId) {});
}$pm->start();
子进程的创建以及回收
例如我们启动的一个服务就可以理解为一个进程,当服务结束也意味着进程结束。而在主进程所另外创建的进程就被称为“子进程”。
而当主进程结束时,子进程还未结束(回收),子进程则会变成僵尸进程,所以主进程结束前需保证子进程全部结束。
<?php
use Swoole\Process;// 循环创建三个子进程
for ($n = 1; $n <= 3; $n++) {// 创建子进程$process = new Process(function () use ($n) {echo 'Child #' . getmypid() . " start and sleep {$n}s" . PHP_EOL;sleep($n);echo 'Child #' . getmypid() . ' exit' . PHP_EOL;});$process->start();
}// 主进程必须等待回收子进程,否则会让子进程变成僵尸进程// 阻塞等待,会阻塞主进程,等待子进程结束
for ($n = 3; $n--;) {$status = Process::wait(true); // 等待阻塞echo "Recycled #{$status['pid']}, code={$status['code']}, signal={$status['signal']}" . PHP_EOL;
}?>
结果:
Child #18130 start and sleep 1s
Child #18131 start and sleep 2s
Child #18132 start and sleep 3s
Child #18130 exit
Recycled #18130, code=0, signal=0
Child #18131 exit
Recycled #18131, code=0, signal=0
Child #18132 exit
Recycled #18132, code=0, signal=0
回收子进程也可以通过另一个方式:
<?php
use Swoole\Process;
use Swoole\Coroutine;
use function Swoole\Coroutine\run;// 循环创建三个子进程
for ($n = 1; $n <= 3; $n++) {// 创建子进程$process = new Process(function () use ($n) {echo 'Child #' . getmypid() . " start and sleep {$n}s" . PHP_EOL;sleep($n);echo 'Child #' . getmypid() . ' exit' . PHP_EOL;});$process->start();
}Coroutine\run(functiom(){while($ret = Swoole\Coroutine\System::wait(5)){ // 等待子进程全部退出echo "子进程结束:子进程为{$ret['pid']}".PHP_EOL;}
});?>
结果:
Child #20899 start and sleep 1s
Child #20900 start and sleep 2s
Child #20901 start and sleep 3s
Child #20899 exit
子进程结束:子进程为20899
Child #20900 exit
子进程结束:子进程为20900
Child #20901 exit
子进程结束:子进程为20901
示例:创建子进程,完成父子通信并监听子进程退出后对其进行回收。
<?php
use Swoole\Process;
use Swoole\Coroutine;
use Swoole\Timer;
use function Swoole\Coroutine\run;// 创建子进程,并每隔2s向主进程发送消息
$process = new Process(function($proc){Timer::tick(2000,function() use ($proc){$socket = $proc->exportSocket(); $socket->send("hello master,this is child , send 2000ms");var_dump($socket->recv());});
},false,1,true);// 启动子进程
$process->start();// 主进程创建协程
Coroutine\run(function() use ($process){// 非阻塞监听子进程退出,监听到后解除监听、并清除所有定时器Process::signal(SIGCHLD,function(){while($ret = Process::wait(false)){Process::signal(SIGCHLD , null); // 解除监听Timer::clearAll(); // 清除所有定时器,包括父进程}});$socket = $process->exportsocket(); // 主进程每隔一秒向子进程发送消息Timer::tick(1000,function() use ($socket){echo "this is parent tick".PHP_EOL;$socket->send("hello child,this is master, send 1000ms");});$count = 2;while($count){$count--;var_dump($socket->recv());if($count == 0){// 杀死子进程Process::kill($process->pid);}};});?>
结果:
this is parent tick
this is parent tick
string(78) "hello child,this is master, send 1000mshello child,this is master, send 1000ms"
string(40) "hello master,this is child , send 2000ms"
this is parent tick
this is parent tick
string(78) "string(40) "hello child,this is master, send 1000mshello child,this is master, send 1000mshello master,this is child , send 2000ms"
锁
进程间锁 Lock
PHP 代码中可以很方便地创建一个锁,用来实现数据同步。Lock 类支持 5 种锁的类型
锁类型 | 说明 |
---|---|
SWOOLE_MUTEX | 互斥锁 |
SWOOLE_RWLOCK | 读写锁 |
SWOOLE_SPINLOCK | 自旋锁 |
SWOOLE_FILELOCK | 文件锁 (废弃) |
SWOOLE_SEM | 信号量 (废弃) |
进程间锁示例:
<?php$lock = new Swoole\Lock(SWOOLE_MUTEX);
echo "[Master]create lock\n";
$lock->lock();
if (pcntl_fork() > 0)
{sleep(1);$lock->unlock();
}
else
{echo "[Child] Wait Lock\n";$lock->lock();echo "[Child] Get Lock\n";$lock->unlock();exit("[Child] exit\n");
}
echo "[Master]release lock\n";
unset($lock);
sleep(1);
echo "[Master]exit\n";?>
输出结果:
[Master]create lock
[Child] Wait Lock
[Master]release lock
[Child] Get Lock
[Child] exit
[Master]exit
无法在协程中石油锁。
进程间无锁计数器 Atomic
Atomic 是 Swoole 底层提供的原子计数操作类,可以方便整数的无锁原子增减。
当有并发请求对计数器进行操作时,Atomic会自带锁。
<?php$atomic = new Swoole\Atomic();$serv = new Swoole\Server('127.0.0.1', '1234');$serv->set(['worker_num' => 1,'log_file' => '/dev/null'
]);
$serv->atomic = $atomic;
$serv->on("receive", function ($serv, $fd, $reactor_id, $data) {$cmd = explode(" ", trim($data));$cmd[1] = $cmd[1] ?? 1;if($cmd[0] == 'add'){$serv->atomic->add($cmd[1]);$serv->send($fd,'add ok,now num is '.$serv->atomic->get().PHP_EOL);}else if($cmd[0] == 'sub'){$serv->atomic->sub($cmd[1]);$serv->send($fd,'sub ok,now num is '.$serv->atomic->get().PHP_EOL);}else{$serv->send($fd,"unkown command {$cmd[0]}".PHP_EOL);}
});$serv->start();?>
启动发我,再另开窗口使用 telnet 127.0.0.1 1234 连接此 TCP 服务,并发送消息。
[root@VM-55-23-centos ~]# telnet 127.0.0.1 1234
Trying 127.0.0.1...
Connected to 127.0.0.1.
Escape character is '^]'.
add 1
add ok,now num is 1
add 10
add ok,now num is 11
sub 5
sub ok,now num is 6
协程实际IO操作
在之前的示例中,大部分是利用 Couroutine::sleep 来模拟 IO 操作。接下来使用协程进行文件写入、数据库、网络请求操作。
<?php
use function Swoole\Coroutine\run;
use Swoole\Coroutine;// 设置协程化类型
Coroutine::set(['hook_flags' => ' SWOOLE_HOOK_TPC | SWOOLE_HOOK_FILE | SWOOLE_HOOK_CURL ',
]);run(function(){var_dump('====first====');Coroutine::Create(function(){$fp = fopen('./test_dir/test.log','a+');fwrite($fp,str_repeat('A',1024));fwrite($fp,str_repeat('B',1024));var_dump('wirte file success');});var_dump('====2====');Coroutine::Create(function(){$mysqli = @new mysqli('127.0.0.1','root','123456','db_name',3306);if($mysqli->connect_errno != 0){var_dump('数据库连接失败:'.$mysqli->connect_errno.'--'.$mysqli->connect_error);return;}$mysqli->query("set name 'utf8'");$res = $mysqli->query('select * from user');if($res){while($row = $res->fetch_assoc()){echo json_encode($row) . "\n";}}$res->free();var_dump('read mysql success');});var_dump('====3====');Coroutine::Create(function(){$ch = curl_init();curl_setopt($ch , CURLOPT_URL , 'http://www.baidu.com/');curl_setopt($ch , CURLOPT_HEADER , false);curl_setopt($ch , CURLOPT_RETURNTRANSFER , 1);$result = curl_exec($ch);curl_close($ch);var_dump('curl success');});var_dump('====end====');});?>
输出结果:
string(13) "====first===="
string(18) "wirte file success"
string(9) "====2===="
{"id":"1","name":"Ho","age":"1"}
{"id":"2","name":"La","age":"2"}
string(18) "read mysql success"
string(9) "====3===="
string(12) "curl success"
string(11) "====end===="
在协程的使用中,由于协程间是同时进行的,且任务的执行是由CPU进行调度的,任务的执行顺序无法保证。
实现MySQL连接对象单例模式
新建文件 mysql/pool.php ,用于连接池相关初始化工作。
<?php
use Swoole\Coroutine;
use function Swoole\Coroutine\run;
use Swoole\Coroutine\Channel;class Pool
{// 连接池,用于存放连接private $pool = null;// 连接配置信息private static $config;// 单例模式private static $instance;// 唯一公开的方法,用于获取单例public static function getInstance(array $config){if(empty(self::$instance)){if(empty($config)){throw new RuntimeException('Config is empty.');}else{self::$config = $config;}self::$instance = new static($config);}return self::$instance;}// 初始化连接池private function __construct($config){if(empty($this->$pool)){// 一键协程化Coroutine::set(['hook_flags' => 'SWOOLE_HOOK_TPC' ]);run(function() use ($config){$this->pool = new Channel($config['pool_size']);for($i = 0 ; $i < $config['pool_size'] ; $i++){go(function() use ($config){try{// 获取连接对象,放入连接池中$mysqli = @new mysqli($config['host'],$config['username'],$config['password'],$config['db_name'],$config['port']);$this->pool->push($mysqli,$config['time_out']);}catch(Throwable $e){var_dump($e);throw new RuntimeException('MySQL connet error:'.$mysqli->errow , $mysqli->errno);}});}});}}// 从连接池获取MySQL连接public function getDBManager(){go(function(){if($this->pool->length() > 0){$this->$mysqli = $this->pool->pop(self::$config['time_out']);if($this->$mysqli === false){throw new RuntimeException('get MySQL failed.');}}else{throw new RuntimeException('MySQL pool is empty.');}});return $this->$mysqli;}// 获取当前连接池中的剩余连接数public function getPoolSize(){return $this->pool->length();}// 向连接池归还连接public function push($mysqli){$this->pool->push($mysqli , $config['time_out']);}// 防止被克隆private function _clone(){}}?>
新建文件 pool_test.php。
<?php
use Swoole\Coroutine;
use Swoole\Event;
var_dump('start...');// 定义数据库连接信息
$config = ['pool_size'=> 5,'host' => '127.0.0.1','username' => 'root','password' => '123456','db_name' => 'test','port' => '3306','time_out' => 1
];
include('./mysql/Pool.php');// 获取连接池实例
$mysqlPool = Pool::getInstance($config);
var_dump('当前连接池内剩余连接数:'.$mysqlPool->getPoolSize());$lock = new Swoole\Lock(SWOOLE_MUTEX);// 获取连接前加锁,防止破坏单例
$lock->lock();// 从连接池中获取一个连接
$mysqli = $mysqlPool->getDBManager();// 获取连接成功后解锁
$lock->unlock();var_dump('当前连接池内剩余连接数:'.$mysqlPool->getPoolSize());// 创建协程
Coroutine::Create(function() use ($mysqli , $mysqlPool){$list = $mysqli->query('select * from user');if($list){var_dump('查询结果:');while($row = $list->fetch_assoc()){echo json_encode($row) . "\n";}}// 释放变量$list->free();// 协程结束前调用该方法Coroutine::defer(function() use ($mysqli , $mysqlPool){var_dump('归还连接...');// 将连接对象push进连接池$mysqlPool->push($mysqli);var_dump('当前连接池内剩余连接数:'.$mysqlPool->getPoolSize());});});var_dump('end...');
Event::wait();
?>
输出结果:
string(8) "start..."
string(35) "当前连接池内剩余连接数:5"
string(35) "当前连接池内剩余连接数:4"
string(13) "查询结果:"
{"id":"1","name":"Ho","age":"1"}
{"id":"2","name":"La","age":"2"}
string(15) "归还连接..."
string(35) "当前连接池内剩余连接数:5"
string(6) "end..."