生活的道路一旦选定，就要勇敢地走到底，决不回头。

发掘积累过程的快感

首页 » BIBLE模型 » PHP » PHP协程

PHP协程

你左我右啊在 那天发布在PHP栏目， 那天又修改过， 到今天为止已经被浏览过67619次了。

对于单核处理器，多进程实现多任务的原理是让操作系统给一个任务每次分配一定的 CPU 时间片，然后中断、让下一个任务执行一定的时间片接着再中断并继续执行下一个，如此反复。由于切换执行任务的速度非常快，给外部用户的感受就是多个任务的执行是同时进行的。

多进程的调度是由操作系统来实现的，进程自身不能控制自己何时被调度，也就是说：

进程的调度是由外层调度器抢占式实现的

而协程要求当前正在运行的任务自动把控制权回传给调度器，这样就可以继续运行其他任务。这与『抢占式』的多任务正好相反, 抢占多任务的调度器可以强制中断正在运行的任务, 不管它自己有没有意愿。『协作式多任务』在 Windows 的早期版本 (windows95) 和 Mac OS 中有使用, 不过它们后来都切换到『抢占式多任务』了。理由相当明确：如果仅依靠程序自动交出控制的话，那么一些恶意程序将会很容易占用全部 CPU 时间而不与其他任务共享。

协程的调度是由协程自身主动让出控制权到外层调度器实现的

回到刚才生成器实现 xrange 函数的例子，整个执行过程的交替可以用下图来表示：

46863831fcd18d5d921f28f.png

协程可以理解为纯用户态的线程，通过协作而不是抢占来进行任务切换。相对于进程或者线程，协程所有的操作都可以在用户态而非操作系统内核态完成，创建和切换的消耗非常低。

简单的说 Coroutine（协程） 就是提供一种方法来中断当前任务的执行，保存当前的局部变量，下次再过来又可以恢复当前局部变量继续执行。

我们可以把大任务拆分成多个小任务轮流执行，如果有某个小任务在等待系统 IO，就跳过它，执行下一个小任务，这样往复调度，实现了 IO 操作和 CPU 计算的并行执行，总体上就提升了任务的执行效率，这也便是协程的意义。

PHP 协程和 yield

PHP 从 5.5 开始支持生成器及 yield 关键字，而 PHP 协程则由 yield 来实现。

要理解协程，首先要理解：代码是代码，函数是函数。函数包裹的代码赋予了这段代码附加的意义：不管是否显式的指明返回值，当函数内的代码块执行完后都会返回到调用层。而当调用层调用某个函数的时候，必须等这个函数返回，当前函数才能继续执行，这就构成了后进先出，也就是 Stack。

而协程包裹的代码，不是函数，不完全遵守函数的附加意义，协程执行到某个点，协会协程会 yield 返回一个值然后挂起，而不是 return 一个值然后结束，当再次调用协程的时候，会在上次 yield 的点继续执行。

所以协程违背了通常操作系统和 x86 的 CPU 认定的代码执行方式，也就是 Stack 的这种执行方式，需要运行环境（比如 php，python 的 yield 和 golang 的 goroutine）自己调度，来实现任务的中断和恢复，具体到 PHP，就是靠 yield 来实现。

堆栈式调用 和 协程调用的对比：

4686383d4c2f201464868bb.png

结合之前的例子，可以总结一下 yield 能做的就是：

  • 实现不同任务间的主动让位、让行，把控制权交回给任务调度器。
  • 通过 send() 实现不同任务间的双向通信，也就可以实现任务和调度器之间的通信。

yield 就是 PHP 实现协程的方式。

协程多任务调度

下面是雄文 Cooperative multitasking using coroutines (in PHP!) 里一个简单但完整的例子，来展示如何具体的在 PHP 里实现协程任务的调度。

首先是一个任务类：

Task

class Task
{
    // 任务 ID
    protected $taskId;
    // 协程对象
    protected $coroutine;
    // send() 值
    protected $sendVal = null;
    // 是否首次 yield
    protected $beforeFirstYield = true;

    public function __construct($taskId, Generator $coroutine) {
        $this->taskId = $taskId;
        $this->coroutine = $coroutine;
    }
  
    public function getTaskId() {
        return $this->taskId;
    }

    public function setSendValue($sendVal) {
        $this->sendVal = $sendVal;
    }

    public function run() {
        // 如之前提到的在send之前, 当迭代器被创建后第一次 yield 之前，一个 renwind() 方法会被隐式调用
        // 所以实际上发生的应该类似:
        // $this->coroutine->rewind();
        // $this->coroutine->send();
       
        // 这样 renwind 的执行将会导致第一个 yield 被执行, 并且忽略了他的返回值.
        // 真正当我们调用 yield 的时候, 我们得到的是第二个yield的值，导致第一个yield的值被忽略。
        // 所以这个加上一个是否第一次 yield 的判断来避免这个问题
        if ($this->beforeFirstYield) {
            $this->beforeFirstYield = false;
            return $this->coroutine->current();
        } else {
            $retval = $this->coroutine->send($this->sendVal);
            $this->sendVal = null;
            return $retval;
        }
    }

    public function isFinished() {
        return !$this->coroutine->valid();
    }
}

接下来是调度器，比 foreach 是要复杂一点，但好歹也能算个正儿八经的 Scheduler :)

Scheduler

class Scheduler
{
    protected $maxTaskId = 0;
    protected $taskMap = []; // taskId => task
    protected $taskQueue;
 
    public function __construct() {
        $this->taskQueue = new SplQueue();
    }
 
    // （使用下一个空闲的任务id）创建一个新任务,然后把这个任务放入任务map数组里. 接着它通过把任务放入任务队列里来实现对任务的调度. 接着run()方法扫描任务队列, 运行任务.如果一个任务结束了, 那么它将从队列里删除, 否则它将在队列的末尾再次被调度。
    public function newTask(Generator $coroutine) {
        $tid = ++$this->maxTaskId;
        $task = new Task($tid, $coroutine);
        $this->taskMap[$tid] = $task;
        $this->schedule($task);
        return $tid;
    }
 
    public function schedule(Task $task) {
        // 任务入队
        $this->queue->enqueue($task);
    }
 
    public function run() {
        while (!$this->queue->isEmpty()) {
            // 任务出队
            $task = $this->queue->dequeue();
            $task->run();
 
            if ($task->isFinished()) {
                unset($this->taskMap[$task->getTaskId()]);
            } else {
                $this->schedule($task);
            }
        }
    }
}

队列可以使每个任务获得同等的 CPU 使用时间，

Demo

function task1() {
    for ($i = 1; $i <= 10; ++$i) {
        echo "This is task 1 iteration $i.\n";
        yield;
    }
}
 
function task2() {
    for ($i = 1; $i <= 5; ++$i) {
        echo "This is task 2 iteration $i.\n";
        yield;
    }
}
  
$scheduler = new Scheduler;
 
$scheduler->newTask(task1());
$scheduler->newTask(task2());
 
$scheduler->run();

输出：

This is task 1 iteration 1.
This is task 2 iteration 1.
This is task 1 iteration 2.
This is task 2 iteration 2.
This is task 1 iteration 3.
This is task 2 iteration 3.
This is task 1 iteration 4.
This is task 2 iteration 4.
This is task 1 iteration 5.
This is task 2 iteration 5.
This is task 1 iteration 6.
This is task 1 iteration 7.
This is task 1 iteration 8.
This is task 1 iteration 9.
This is task 1 iteration 10.

结果正是我们期待的，最初的 5 次迭代，两个任务是交替进行的，而在第二个任务结束后，只有第一个任务继续执行到结束。

协程非阻塞 IO

若想真正的发挥出协程的作用，那一定是在一些涉及到阻塞 IO 的场景，我们都知道 Web 服务器最耗时的部分通常都是 socket 读取数据等操作上，如果进程对每个请求都挂起的等待 IO 操作，那处理效率就太低了，接下来我们看个支持非阻塞 IO 的 Scheduler：

<?php

class Scheduler
{
    protected $maxTaskId = 0;
    protected $tasks = []; // taskId => task
    protected $queue;

    // resourceID => [socket, tasks]
    protected $waitingForRead = [];
    protected $waitingForWrite = [];
 
    public function __construct() {
        // SPL 队列
        $this->queue = new SplQueue();
    }
 
    public function newTask(Generator $coroutine) {
        $tid = ++$this->maxTaskId;
        $task = new Task($tid, $coroutine);
        $this->tasks[$tid] = $task;
        $this->schedule($task);
        return $tid;
    }
 
    public function schedule(Task $task) {
        // 任务入队
        $this->queue->enqueue($task);
    }
 
    public function run() {
        while (!$this->queue->isEmpty()) {
            // 任务出队
            $task = $this->queue->dequeue();
            $task->run();
 
            if ($task->isFinished()) {
                unset($this->tasks[$task->getTaskId()]);
            } else {
                $this->schedule($task);
            }
        }
    }

    public function waitForRead($socket, Task $task)
    {
        if (isset($this->waitingForRead[(int)$socket])) {
            $this->waitingForRead[(int)$socket][1][] = $task;
        } else {
            $this->waitingForRead[(int)$socket] = [$socket, [$task]];
        }
    }

    public function waitForWrite($socket, Task $task)
    {
        if (isset($this->waitingForWrite[(int)$socket])) {
            $this->waitingForWrite[(int)$socket][1][] = $task;
        } else {
            $this->waitingForWrite[(int)$socket] = [$socket, [$task]];
        }
    }

    /**
     * @param $timeout 0 represent
     */
    protected function ioPoll($timeout)
    {
        $rSocks = [];
        foreach ($this->waitingForRead as list($socket)) {
            $rSocks[] = $socket;
        }

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

        $eSocks = [];
        // $timeout 为 0 时, stream_select 为立即返回，为 null 时则会阻塞的等，见 http://php.net/manual/zh/function.stream-select.php
        if (!@stream_select($rSocks, $wSocks, $eSocks, $timeout)) {
            return;
        }

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

            foreach ($tasks as $task) {
                $this->schedule($task);
            }
        }

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

            foreach ($tasks as $task) {
                $this->schedule($task);
            }
        }
    }

    /**
     * 检查队列是否为空，若为空则挂起的执行 stream_select，否则检查完 IO 状态立即返回，详见 ioPoll()
     * 作为任务加入队列后，由于 while true，会被一直重复的加入任务队列，实现每次任务前检查 IO 状态
     * @return Generator object for newTask
     *
     */
    protected function ioPollTask()
    {
        while (true) {
            if ($this->taskQueue->isEmpty()) {
                $this->ioPoll(null);
            } else {
                $this->ioPoll(0);
            }
            yield;
        }
    }

    /**
     * $scheduler = new Scheduler;
     * $scheduler->newTask(Web Server Generator);
     * $scheduler->withIoPoll()->run();
     *
     * 新建 Web Server 任务后先执行 withIoPoll() 将 ioPollTask() 作为任务入队
     * 
     * @return $this
     */
    public function withIoPoll()
    {
        $this->newTask($this->ioPollTask());
        return $this;
    }
}

这个版本的 Scheduler 里加入一个永不退出的任务，并且通过 stream_select 支持的特性来实现快速的来回检查各个任务的 IO 状态，只有 IO 完成的任务才会继续执行，而 IO 还未完成的任务则会跳过，完整的代码和例子可以戳这里。

也就是说任务交替执行的过程中，一旦遇到需要 IO 的部分，调度器就会把 CPU 时间分配给不需要 IO 的任务，等到当前任务遇到 IO 或者之前的任务 IO 结束才再次调度 CPU 时间，以此实现 CPU 和 IO 并行来提升执行效率，类似下图：

4686383fb6b7c133914d538.png

单任务改造

如果想将一个单进程任务改造成并发执行，我们可以选择改造成多进程或者协程：

  • 多进程 ，不改变任务执行的整体过程，在一个时间段内同时执行多个相同的代码段，调度权在 CPU，如果一个任务能独占一个 CPU 则可以实现并行。
  • 协程 ，把原有任务拆分成多个小任务，原有任务的执行流程被改变，调度权在进程自己，如果有 IO 并且可以实现异步，则可以实现并行。

多进程改造

468638339140a5747e4e3f5.png

协程改造

4686383efcbd88f282e9a5a.png

协程（Coroutines）和 Go 协程（Goroutines）

PHP 的协程或者其他语言中，比如 Python、Lua 等都有协程的概念，和 Go 协程有些相似，不过有两点不同：

  • Go 协程意味着并行（或者可以以并行的方式部署，可以用 runtime.GOMAXPROCS() 指定可同时使用的 CPU 个数），协程一般来说只是并发。
  • Go 协程通过通道 channel 来通信；协程通过 yield 让出和恢复操作来通信。

Go 协程比普通协程更强大，也很容易从协程的逻辑复用到 Go 协程，而且在 Go 的开发中也使用的极为普遍，有兴趣的话可以了解一下作为对比。

结束

个人感觉 PHP 的协程在实际使用中想要徒手实现和应用并不方便而且场景有限，但了解其概念及实现原理对更好的理解并发不无裨益。

如果想更多的了解协程的实际应用场景不妨试试已经大名鼎鼎的 Swoole，其对多种协议的 client 做了底层的协程封装，几乎可以做到以同步编程的写法实现协程异步 IO 的效果。

PHP协程使用指南专题
编程语言
前端设计
计算机基础
数据魅力
开源
程序人生
互联网信息太多太杂，各互联网公司不断推送娱乐花边新闻，SNS，微博不断转移我们的注意力。但是，我们的时间和精力却是有限的。这里是互联网浩瀚的海洋中的一座宁静与美丽的小岛，供开发者歇息与静心潜心修炼。 “Bible”是圣经，有权威的书，我们的本意就是为开发者提供真正有用的的资料。 我的电子邮件 1217179982@qq.com，您在开发过程中遇到任何问题，欢迎与我联系。
Copyright © 2024. All rights reserved. 本站由 Helay 纯手工打造. 蜀ICP备15017444号