PHP【swoole】-编程知识

前言

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..."