前言
Swoole官方文档：Swoole 文档
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)*
创建 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 命令查看进程：
毫秒定时器
创建 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));
//get
if ($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");
}
//set
elseif ($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-协程间是同步执行的，并不会进行阻塞。而在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
创建 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)
}可以看到，输出内容包含了两个协程的结果。
Barrier
创建 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 处理异常。但必须在协程内捕获，不得跨协程捕获异常。
<?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的错误并不会影响协程2
Coroutine::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
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..."其中：
同时可以基于 $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协程系统杂项函数
<?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进程管理器
进程管理器，基于 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 种锁的类型
进程间锁示例：
<?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..."