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));

    //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-

协程的创建方式：

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的错误并不会影响协程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

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