PHP中的协程

博主： dafenqi
发布时间：2023 年 08 月 10 日
102 次浏览
暂无评论
18491字数
分类： php 教程

#  [PHP中的协程](https://blog.p2hp.com/archives/4269) 
之前学习Lua的时候第一次接触到了协程(coroutine)的概念。而PHP5.5版本中也加入了协程的概念，从此PHP编程又有了新的思路和玩法。这里学习一下PHP中协程的相关概念的使用方法。

分成上下两篇文章吧，这篇主要讲一下基础概念。

协程是什么？
在以前的Lua学习笔记三中可以看到，协程与多线程的比较，有自己的堆栈、局部变量、指令指针等，但是协程本身与其他协程共享全局变量。主要不同在于，多处理器下，多线程可以真实的同时运行多个线程。而协程任意时刻只能有一个在真实运行，并且只有在明确要求被挂起时才会挂起。

PHP中协程如何理解？
这里引用知乎赵老师的答案，说的比较好理解。具体来说，一个包含yeild的php函数，就是协程，他有阶段性的结算值 yield $var， 但是代码并不返回，php的调度者接到这个值后，喂给一个generator，generator是个实现了iterator接口的+和协程通讯接口（比如send方法）的实例，所以可以用在for循环里（另个接口负责和协程通讯）。那么generator收到了这个协程的阶段性的值后，他喂给for循环，等for循环下一次循环的时候，他又启动这个协程，协程从上次中断的点继续执行，继续计算，继续yeild值给generator，generator喂for循环，继续循环，直到协程执行完毕。

相关函数
final class Generator implements Iterator {
    public function rewind();     // 返回到迭代器的第一个元素。
    public function valid();      // 返回false如果迭代器已经关闭，否则返回true
    public function current();    // 返回当前yield值.
    public function key();        // 返回当前yield键名.
    public function next();       // 恢复生成器的执行。
    public function PS_UNRESERVE_PREFIX_throw(Exception $exception) {};//抛出异常
    public function send($value); // 将传入的值作为yield表达式的结果并且恢复发生器的执行。
}
简单例子
简单的迭代器给foreach使用

function gen() {
    $ret = (yield 'a');
    echo $ret;
    $ret = (yield 'b');
    echo $ret;
}

$gen = gen();
$ret = $gen->current();
echo $ret;
$ret = $gen->send("c");
echo $ret;
$ret = $gen->send("d");
echo $ret;
//output acbd
带抛出异常的例子

function gen() {
    try{
        $ret = (yield 'a');
        echo $ret;
        $ret = (yield 'b');
        echo $ret;
    } catch (Exception $ex) {
        echo $ex->getMessage();
    }
}

$gen = gen();
$ret = $gen->current();
echo $ret;
$ret = $gen->send("c");
echo $ret;
$ret = $gen->throw(new Exception("d"));
var_dump($ret);

//output acbdNULL
那么能用来干什么呢？
我们来看看，协程可以自己主动出让执行权，把不需要抢占的操作时间(比如socket等待链接)让出来，并且可以和调用方通过yield的方式传递信息。显而易见，他可以用来做多任务调度！

PHP中协程实现多任务调度，鸟哥有一篇翻译的文章里有讲解，网上能找到的大部分资料，都跟这篇相关。但是至少在我看来，理解起来还是蛮复杂的。这里针对那篇文章的前半部分做一个笔记，忽略后面关于独立堆栈协程的部分。

function里使用yield关键字，将生成迭代器。这样调用functionName()时，其实得到的是一个迭代器对象，而并没有实际运行程序。
为什么要走系统调用SystemCall这一层呢？模拟进程和系统会话的方式，控制权限。通过给yield表达式传递信息来与调度器通信，yield既是中断也是传递给调度器的方式。
SystemCall 包含一个回调函数，他自己本身可以被执行。被执行时实际上是调用了这个回调函数，入参是某个task和调度器。
SystemCall其实并没有其他作用，只是在协程函数里面跟在yield后面传给调度器来执行。
注意SplQueue塞进去的对象其实是引用(PHP里对象入参都是引用，不只是SplQueue)！外面对象改了，里面也会变。
为什么忽略协程堆栈？
我打算在第二篇文章中，把有赞的zan framework里关于协程的部分抽出来，针对性的说一下包含子协程额多任务调度。当然主要想偷个懒。

不过个人感觉zan框架里的协程部分，比之前说的那篇文章要好理解一些。

上篇文章里提到PHP中协程的引入，可以使PHP编程有新的玩法，不在遵循原本顺序执行的思路，从而应对大访问量和并发操作。

有赞的zan framework就是基于PHP协程的，提供最简单的方式开发面向C10K+的高并发HTTP服务或SOA服务。我并没有深入的学习这个框架，这里只打算把关于协程的部分抽出来学习一下。

zan框架高并发设计思路
粗看之下(不一定对哦)，框架应该是用swoole_server + 协程解决高并发访问。

比如Web服务中，swoole的http_server只开启了少数的几个worker进程。我们知道，如果worker的onRequest里使用的是异步方法，则worker的响应是异步处理的，反之则是阻塞的。

zan框架在worker进程中大量使用了PHP协程，所以一个worker进程可以响应很多并发的请求(但是本质上正在执行的只有一个)，这应该就是能过达到C10K+的原因吧。

并且协程相对于回调的方式，在PHPer看来更容易接受吧。另外，框架设计里还使用了middleware、连接池、依赖注入等等比较现代的设计，感觉可以更深入的学习一下。

一个类一个类来看
我们把框架里关于协程的部分拆出来看，下面一个类一个类的分析。与鸟哥博客里那个文章的实现相比，有一些相同的地方，更多的是不同。比如那篇文章里，多个任务放到一个schedule里调度，对于后面的实现就比较繁琐。

这里把框架里的代码抽出来，并进行了一定删减，去掉了与其他业务强相关的东西。比如Event、Context、Async等等。实际上Async用于处理MySQL查询的返回值的，框架内部将MySQL的具体操作类封装成了Async的子类，并且yield给调度器来用。

Singal类
Singal类里包含了系统调用所需的信号量。指明了协程在一轮运行之后应该处于的状态。

class Signal
{
    const TASK_SLEEP        = 1;
    const TASK_AWAKE        = 2;
    const TASK_CONTINUE     = 3;
    const TASK_KILLED       = 4;
    const TASK_RUNNING      = 5;
    const TASK_WAIT         = 6;
    const TASK_DONE         = 7;

public static function isSignal($signal) {
        if(!$signal) {
            return false;
        }

if (!is_int($signal)) {
            return false;
        }

if($signal < 1 ) {
            return false;
        }

if($signal > 7) {
            return false;
        }

return true;
    }
}
Task类
Task包装了具体协程函数，并提供相应的get set方法。与网上流行的那篇文章(以下简称那文)不同的是，我们的scheduler是内置于Task里的，在run方法里实现具体的调度。

这里我们省略了Context，并且让taskId自增。

class Task
{
    protected $taskId = 0;
    protected $parentId = 0;
    protected $coroutine = null;
    //这里忽略了context 保存的是当前http请求的相关信息，可以通过系统调用的方式操作
    protected $context = null;

protected $sendValue = null;
    protected $scheduler = null;
    protected $status = 0;

public function __construct(Generator $coroutine, $taskId = 0, $parentId = 0) {
        $this->coroutine = $coroutine;
        if(isset($GLOBALS['stTaskId']) && $taskId == 0){
            global $stTaskId;
            $taskId = $stTaskId ++;
        }
        $this->taskId = $taskId;
        $this->parentId = $parentId;
        $this->scheduler = new Scheduler($this);
    }

/**
     * 静态方法调用
     * @param $coroutine
     * @param int $taskId
     * @param int $parentId
     * @return Task
     */
    public static function execute($coroutine, $taskId = 0, $parentId = 0) {
        if ($coroutine instanceof Generator) {
            if(isset($GLOBALS['stTaskId']) && $taskId == 0){
                global $stTaskId;
                $taskId = $stTaskId ++;
            }
            $task = new Task($coroutine, $taskId, $parentId);
            $task->run();
            return $task;
        }
        return $coroutine;
    }

public function run() {
        while (true) {
            try {
                if ($this->status == Signal::TASK_KILLED){
                    $this->fireTaskDoneEvent();
                    break;
                }
                $this->status = $this->scheduler->schedule();
                //以下几种状态表示信号量，实际上已经从while里跳出来了。如果需要继续的话，会在其他地方重启。
                switch ($this->status) {
                    case Signal::TASK_KILLED:
                    case Signal::TASK_SLEEP:
                    case Signal::TASK_WAIT:
                        return null;
                    case Signal::TASK_DONE:
                        $this->fireTaskDoneEvent();
                        return null;
                }
            } catch (Exception $e) {
                $this->scheduler->throwException($e);
            }
        }
    }

public function send($value) {
        $this->sendValue = $value;
        return $this->coroutine->send($value);
    }

public function getTaskId() {
        return $this->taskId;
    }

public function getContext() {
        return $this->context;
    }

public function getSendValue() {
        return $this->sendValue;
    }

public function getResult() {
        return $this->sendValue;
    }

public function getStatus() {
        return $this->status;
    }

public function setStatus($signal) {
        $this->status = $signal;
    }

public function getCoroutine() {
        return $this->coroutine;
    }

public function setCoroutine(Generator $coroutine) {
        $this->coroutine = $coroutine;
    }

public function fireTaskDoneEvent() {
        echo "Task done $this->taskId\n";
    }
}
Scheduler类
scheduler类负责：

获取Task里的协程函数跑完一轮的返回值
根据返回值的类型采取不同的处理方式，如系统调用、子协程、普通yield值、检查协程栈等等。
在子协程的调用过程中，负责父子协程的进栈出栈，yield值的传递等等。
class Scheduler
{
    private $task = null;
    private $stack = null;

public function __construct(Task $task)
    {
        $this->task = $task;
        $this->stack = new SplStack();
    }

public function schedule()
    {
        $coroutine = $this->task->getCoroutine();

$value = $coroutine->current();

$signal = $this->handleSysCall($value);
        if ($signal !== null) return $signal;

$signal = $this->handleCoroutine($value);
        if ($signal !== null) return $signal;

$signal = $this->handleYieldValue($value);
        if ($signal !== null) return $signal;

$signal = $this->handleTaskStack($value);
        if ($signal !== null) return $signal;

$signal = $this->checkTaskDone($value);
        if ($signal !== null) return $signal;

return Signal::TASK_DONE;
    }

public function isStackEmpty()
    {
        return $this->stack->isEmpty();
    }

public function throwException($e, $isFirstCall = false)
    {
        if ($this->isStackEmpty()) {
            $this->task->getCoroutine()->throw($e);
            return;
        }

try{
            if ($isFirstCall) {
                $coroutine = $this->task->getCoroutine();
            } else {
                $coroutine = $this->stack->pop();
            }

$this->task->setCoroutine($coroutine);
            $coroutine->throw($e);

$this->task->run();
        }catch (Exception $e){
            $this->throwException($e);
        }
    }

/**
     * 处理系统调用
     * @param $value
     * @return mixed|null
     */
    private function handleSysCall($value)
    {
        if (!($value instanceof SysCall)
            && !is_subclass_of($value, SysCall::class)
        ) {
            return null;
        }
        echo $this->task->getTaskId()."| SYSCALL\n";
        //走系统调用 实际上因为__invoke 走的是 $value($this->task);
        $signal = call_user_func($value, $this->task);
        if (Signal::isSignal($signal)) {
            return $signal;
        }

return null;
    }

/**
     * 处理子协程
     * @param $value
     * @return int|null
     */
    private function handleCoroutine($value)
    {
        if (!($value instanceof Generator)) {
            return null;
        }
        echo $this->task->getTaskId()."| COROUTINE\n";
        //获取当前的协程 入栈
        $coroutine = $this->task->getCoroutine();
        $this->stack->push($coroutine);
        //将新的协程设为当前的协程
        $this->task->setCoroutine($value);

return Signal::TASK_CONTINUE;
    }

/**
     * 处理协程栈
     * @param $value
     * @return int|null
     */
    private function handleTaskStack($value)
    {
        //能够跑到这里说明当前协程已经跑完了 valid()==false了 需要看下栈里是否还有以前的协程
        if ($this->isStackEmpty()) {
            return null;
        }

echo $this->task->getTaskId()."| TASKSTACK\n";
        //出栈 设置为当前运行的协程
        $coroutine = $this->stack->pop();
        $this->task->setCoroutine($coroutine);

//这个sendvalue可能是从刚跑完的协程那里得到的 把它当做send值传给老协程 让他继续跑
        $value = $this->task->getSendValue();
        $this->task->send($value);

return Signal::TASK_CONTINUE;
    }

/**
     * 处理普通的yield值
     * @param $value
     * @return int|null
     */
    private function handleYieldValue($value)
    {
        $coroutine = $this->task->getCoroutine();
        if (!$coroutine->valid()) {
            return null;
        }
//        if($this->task->getTaskId() == 2){
//
//        }else{
            echo $this->task->getTaskId()."| YIELD VALUE\n";
//        }
        //如果协程后面没有yield了 这里发出send以后valid就变成false了 并且current变成NULL
        $status = $this->task->send($value);
        return Signal::TASK_CONTINUE;
    }

private function checkTaskDone($value)
    {
        $coroutine = $this->task->getCoroutine();
        if ($coroutine->valid()) {
            return null;
        }
        echo $this->task->getTaskId()."| CHECKDONE\n";

return Signal::TASK_DONE;
    }
}
SysCall类
与那文的思路相同，系统调用类一般作为yield后面跟着的值吐给外层的调用方来执行，并且可能返回响应的信号量，标识这个Task是继续运行还是进入等待状态中。

不同的是这里的__invoke入参不需要Scheduler。

class SysCall
{
    protected $callback = null;

public function __construct(\Closure $callback)
    {
        $this->callback = $callback;
    }

public function __invoke(Task $task)
    {
        return call_user_func($this->callback, $task);
    }
}
组装起来！
基本的组件就是上面的几个类了，下面举一些实际的例子，说明如何利用这几个看似简陋的组件来搞大新闻。

延迟执行任务
function taskSleep($ms)
{
    return new SysCall(function (Task $task) use ($ms) {
        swoole_timer_after($ms, function() use($task){
            $task->send("this is send value in sleep function.");
            $task->run();
        });
        return Signal::TASK_SLEEP;
    });
}

function delay(){
    yield taskSleep(2000);
}

function gen(){
    echo "gen1\n";
    yield 1;
    echo "gen2\n";
    yield 2;
    echo "gen3\n";
    yield 3;
}

//Task::execute(delay(), 1); 亦可
(new Task(delay(), 1))->run();
(new Task(gen(), 2))->run();

/** output
1| SYSCALL
gen1
2| YIELD VALUE
gen2
2| YIELD VALUE
gen3
2| YIELD VALUE
2| CHECKDONE
Task done 2
//2秒以后//
1| CHECKDONE
Task done 1
**/
taskSleep是个系统调用，告诉调度器我要睡眠了(传递给他一个Signal::TASK_SLEEP)。具体说明时候唤醒呢，要等swoole_timer_after2秒以后将它唤醒。

我们这里同时跑了两个任务，从输出来看第一个任务的延时执行，并不会阻塞第二个任务。可以清楚地看到，我们的协程是可以实现多任务并行处理的(当然实际上并不是并行)。

独立堆栈的子协程
function justReturnValue(){
    yield (delay());
    yield 'yield value 2';
}

function gen2(){
    $ret1 = (yield "yield value 1");
    echo "[ret] $ret1\n";
    $ret2 = (yield justReturnValue());
    echo "[ret] $ret2\n";
}

(new Task(gen2(), 1))->run();

/** output
1| YIELD VALUE
[ret] yield value 1
1| COROUTINE
1| COROUTINE
1| SYSCALL
// 2秒以后 //
1| TASKSTACK
1| YIELD VALUE
1| TASKSTACK
[ret] yield value 2
1| CHECKDONE
Task done 1
**/
gen2里有一个子协程justReturnValue的调用，而justReturnValue里也有delay的子协程调用。通过输出可以清楚的看到，父子协程进栈出栈的顺序，以及出栈的协程会将吐出来的值交给原先的协程。

实现一个非阻塞IO的Web服务
参照那文里的实现，我们也可以写一个自己的Web服务。首先还是来说明一下要做什么，以及思路。

直接引用那文的说法:

有一个任务是在套接字上侦听是否有新连接，当有新连接要建立的时候，它创建一个新任务来处理新连接。

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

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

传统的做法中，创建一个套接字，等待新连接，然后读取、发送、关闭。这些都是阻塞的，会花时间在这些抢占资源的步骤上。如果我们使用协程的方式，可以先将等待操作的任务yield掉，之后结合stream_select方法，选择出可以继续操作的任务将其resume。

通俗的说，可以理解为大家一起挤公交车，原先必须一个一个上，但是上车以后要找公交卡，要刷卡或者投币，操作完了以后下一个乘客才能继续。如果使用协程的话，第一个乘客A上车以后，挂起到一边找公交卡，不影响第二个乘客B上车。等到A掏出公交卡以后，直接插队刷卡上车。虽然还是一个一个排队上车，但是找卡的时间里其他乘客不会干等了。

socket的状态
首先定义2个全局数组保存所有用到的socket。再定义2个系统调用将socket设置进数组里，并且返回等待信号量让Task挂起。

$waitingForRead = [];
$waitingForWrite = [];

function ioPoll($timeout) {
    global $waitingForRead;
    global $waitingForWrite;

$rSocks = [];
    foreach ($waitingForRead as list($socket)) {
        $rSocks[] = $socket;
    }

$wSocks = [];
    foreach ($waitingForWrite as list($socket)) {
        $wSocks[] = $socket;
    }

$eSocks = []; // dummy

//stream_select 方法会直接修改入参 只保留就绪的socket数组
    if (false === stream_select($rSocks, $wSocks, $eSocks, $timeout)) {
        return;
    }

foreach ($rSocks as $socket) {
        list(, $tasks) = $waitingForRead[(int) $socket];
        unset($waitingForRead[(int) $socket]);

foreach ($tasks as $task) {
            $task->send("ready for read");
            $task->run();
        }
    }

foreach ($wSocks as $socket) {
        list(, $tasks) = $waitingForWrite[(int) $socket];
        unset($waitingForWrite[(int) $socket]);

foreach ($tasks as $task) {
            $task->send("ready for write");
            $task->run();
        }
    }
}

class CoSocket {
    protected $socket;

public function __construct($socket) {
        $this->socket = $socket;
    }

public function accept() {
        //等待本socket就绪
        yield waitForRead($this->socket);
        //就绪以后会继续走到这里 返回给外层一个客户端连接socket
        yield stream_socket_accept($this->socket, 0);
    }

public function read($size) {
        //等待本socket就绪
        yield waitForRead($this->socket);
        //就绪以后回把读取到的内容 返回给外层
        yield fread($this->socket, $size);
    }

public function write($string) {
        //等待本socket就绪
        yield waitForWrite($this->socket);
        //就绪以后把响应写给客户端
        fwrite($this->socket, $string);
    }

public function close() {
        @fclose($this->socket);
    }
}
处理客户端新连接
服务端socket接受到新的连接以后，创建新的任务。下面是这个任务里实际运行的协程。

function handleClient(CoSocket $socket) {
    $data = (yield $socket->read(8192));
    $msg = "Received following request:\n\n$data";
    $msgLength = strlen($msg);

//响应报文由状态行(HTTP版本、状态码)+HTTP首部字段(响应首部字段、通用首部字段、实体首部字段)组成。
    //空行(CR+LF)分隔首部与报文主体。所以这里留个空行在打印$msg
    $response = <<<RES
HTTP/1.1 200 OK\r
Content-Type: text/plain\r
Content-Length: $msgLength\r
Connection: close\r
\r
$msg
RES;

yield $socket->write($response);
    yield $socket->close();
}
开启服务！
直接创建一个包含server协程的任务，和一个不断刷新stream_select的任务。之后的流程都交给ioPollTask来调度了。

//定义全局的taskId自增用
static $stTaskId = 1;

//创建服务端socket的task 1
Task::execute(server(8000));
//不断刷新socket_select的task 2
Task::execute(ioPollTask());
运行效果
开启服务后，我们先直接用curl访问，观测一下得到的结果。

➜  ~ curl -d "a=123&b=456"  http://localhost:8000
Received following request:

POST / HTTP/1.1
Host: localhost:8000
User-Agent: curl/7.51.0
Accept: */*
Content-Length: 11
Content-Type: application/x-www-form-urlencoded

a=123&b=456
可以看到服务端吐出了我们发送给他的信息，包括HTTP请求行、首部字段和正文。如果我们在浏览器里访问的话，正文内容会丰富许多，会有Cookie,UA等等，如下：

Received following request:

GET / HTTP/1.1
Host: localhost:8000
Connection: keep-alive
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8
DontTrackMeHere: gzip, deflate, sdch, br
Accept-Language: zh-CN,zh;q=0.8,en;q=0.6,zh-TW;q=0.4
Cookie: Phpstorm-f86ac615=34137ba0-5113-4922-b809-b6fa20dbf937
不足的地方
zan framework里的协程调度，并没有采用任务队列的方式。可能是因为他只是针对单独的http或者tcp请求来设计的吧，一般是链式调用。由于这个原因，所以没法设置具体某个任务的执行顺序。当然实际效果跟那文里是相同的，说到底还是由ioPollTask来驱动。

小结
我们通过几个例子加深了对PHP中协程用法的理解。需要注意的是，在协程中(本文构造的这种结构)我们要避免使用死循环，除非循环里yield的结果可以将其挂起并出让控制权给其他协程。

比如上文的Web服务器中，因为有不能主动挂起的ioPollTask，所以不能实现在响应时延迟几秒的效果。因为即使使用了taskSleep这种系统调用，也会因为ioPollTask死循环导致不能获取控制权无法执行

最后修改：2023 年 08 月 10 日

如果觉得我的文章对你有用，请随意赞赏

发表评论取消回复
使用cookie技术保留您的个人信息以便您下次快速评论，继续评论表示您已同意该条款

评论 *

私密评论

名称 *

🎲

邮箱 *

地址

zodhzrusdp
《遥遥乡愁》剧情片高清在线免费观看：https://www.j...
jprfffedhg
你的才华让人瞩目，期待你的更多文章。 https://w...
pkrqcpalxw
你的文章让我感受到了无尽的欢乐，谢谢分享。 http:/...
urjitfqbzy
你的文章让我感受到了生活的美好，谢谢！ https://...
bfgsfrpbas
你的文章让我心情愉悦，每天都要来看一看。 https:/...

浅谈php web安全
评论数： 0
数据库架构设计
评论数： 0
php 目录
评论数： 0
php性能分析
评论数： 0
使用workerman加速任意项目
评论数： 0

PHP中的协程

dafenqi • 2023 年 08 月 10 日