在PHP7中實現協程
- Guanhui轉載
- 2020-05-02 11:08:202917瀏覽
前言
相信大家都聽過『協程』這個概念吧。
但有些同學對這個概念似懂非懂，不知道怎麼實現，怎麼用，用在哪，甚至有些人認為yield就是協程！
我始終相信，如果你無法準確地表達出一個知識點的話，我可以認為你就是不懂。
如果你之前了解過利用PHP實現協程的話，你肯定看過鳥哥的那篇文章：在PHP中使用協程實現多任務調度| 風雪之隅
#鳥哥這篇文章是從國外的作者翻譯來的，翻譯的簡潔明了，也給了具體的例子了。
我寫這篇文章的目的，是想對鳥哥文章做更加充足的補充，畢竟有部分同學的基礎還是不夠好，看得也是雲頭霧裡的。
什麼是協程
先搞清楚，什麼是協程。
你可能已經聽過『進程』和『執行緒』這兩個概念。
進程就是二進位執行檔在電腦記憶體裡的一個運行實例，就好比你的.exe檔是個類，進程就是new出來的那個實例。
行程是電腦系統進行資源分配與調度的基本單位（調度單位這裡別糾結執行緒進程的），每個CPU下同一時刻只能處理一個行程。
所謂的並發，只不過是看起來CPU好像同時能處理幾件事情一樣，對於單核CPU事實上在用很快的速度切換不同的進程。
進程的切換需要進行系統調用，CPU要保存目前進程的各個信息，同時也會使CPUCache被廢掉。
所以進程切換不到非不得已就不做。
那麼怎麼實現『進程切換不到非不得已就不做』？
首先行程被切換的條件是：行程執行完畢、分配給行程的CPU時間片結束，系統發生中斷需要處理，或是行程等待必要的資源（行程阻塞）等。你想下，前面幾種情況自然沒有什麼話可說，但是如果是在阻塞等待，是不是就浪費了。
其實阻塞的話我們的程式還有其他可執行的地方可以執行，不一定要傻傻的等！
所以就有了線程。
線程簡單理解就是一個‘微進程’，專門跑一個函數（邏輯流）。
所以我們就可以在寫程式的過程中將可以同時執行的函數用執行緒來體現了。
執行緒有兩種類型，一種是由核心來管理和調度。
我們說，只要涉及需要核心參與管理調度的，代價都是很大的。這種執行緒其實也解決了當一個行程中，某個正在執行的執行緒遇到阻塞，我們可以調度另外一個可運行的執行緒來跑，但是還是在同一個行程裡，所以沒有了行程切換。
還有另一個線程，他的調度是由程式設計師自己寫程式來管理的，對核心來說不可見。這種線程叫做『用戶空間線程』。
協程可以理解就是一種使用者空間執行緒。
協程，有幾個特點：
協同，因為是由程式設計師自己寫的調度策略，其透過協作而不是搶佔來進行切換
在用戶態完成創建，切換和銷毀
⚠️ 從程式角度上看，協程的思想本質上就是控制流的主動讓出（yield）和恢復（resume）機制
generator經常用來實現協程
說到這裡，你應該明白協程的基本概念了吧？
PHP實作協程
一步一步來，從解釋概念說起！
可迭代物件
PHP5提供了一種定義物件的方法使其可以透過單元列表來遍歷，例如用foreach語句。
你如果要實作一個可迭代對象，你就要實作Iterator介面：
<?php
class MyIterator implements Iterator
{
private $var = array();
public function __construct($array)
{
if (is_array($array)) {
$this->var = $array;
}
}
public function rewind() {
echo "rewinding\n";
reset($this->var);
}
public function current() {
$var = current($this->var);
echo "current: $var\n";
return $var;
}
public function key() {
$var = key($this->var);
echo "key: $var\n";
return $var;
}
public function next() {
$var = next($this->var);
echo "next: $var\n";
return $var;
}
public function valid() {
$var = $this->current() !== false;
echo "valid: {$var}\n";
return $var;
}
}
$values = array(1,2,3);
$it = new MyIterator($values);
foreach ($it as $a => $b) {
print "$a: $b\n";
}產生器
可以說之前為了擁有一個能夠被foreach遍歷的對象，你不得不去實作一堆的方法，yield關鍵字就是為了簡化這個過程。
生成器提供了一種更容易的方法來實現簡單的物件迭代，相比較定義類別實作Iterator介面的方式，效能開銷和複雜性大大降低。
<?php
function xrange($start, $end, $step = 1) {
for ($i = $start; $i <= $end; $i += $step) {
yield $i;
}
}
foreach (xrange(1, 1000000) as $num) {
echo $num, "\n";
}記住，一個函數中如果用了yield，他就是一個生成器，直接呼叫他是沒有用的，不能等同於一個函數那樣去執行！
所以，yield就是yield，下次誰再說yield是協程，我一定把你xxxx。
PHP協程
前面介紹協程的時候說了，協程需要程式設計師自己去寫調度機制，下面我們來看這個機制怎麼寫。
生成器正確使用
既然生成器不能像函數一樣直接調用，那麼怎麼能調用呢？
方法如下：
foreach他
send($value)
current / next...
Task實作
Task就是一個任務的抽象，剛剛我們說了協程就是使用者空間線程，線程可以理解就是跑函數。
所以Task的建構子中就是接收一個閉包函數，我們命名為coroutine。
/**
* Task任务类
*/
class Task
{
protected $taskId;
protected $coroutine;
protected $beforeFirstYield = true;
protected $sendValue;
/**
* Task constructor.
* @param $taskId
* @param Generator $coroutine
*/
public function __construct($taskId, Generator $coroutine)
{
$this->taskId = $taskId;
$this->coroutine = $coroutine;
}
/**
* 获取当前的Task的ID
*
* @return mixed
*/
public function getTaskId()
{
return $this->taskId;
}
/**
* 判断Task执行完毕了没有
*
* @return bool
*/
public function isFinished()
{
return !$this->coroutine->valid();
}
/**
* 设置下次要传给协程的值，比如 $id = (yield $xxxx)，这个值就给了$id了
*
* @param $value
*/
public function setSendValue($value)
{
$this->sendValue = $value;
}
/**
* 运行任务
*
* @return mixed
*/
public function run()
{
// 这里要注意，生成器的开始会reset，所以第一个值要用current获取
if ($this->beforeFirstYield) {
$this->beforeFirstYield = false;
return $this->coroutine->current();
} else {
// 我们说过了，用send去调用一个生成器
$retval = $this->coroutine->send($this->sendValue);
$this->sendValue = null;
return $retval;
}
}
}Scheduler实现
接下来就是Scheduler这个重点核心部分，他扮演着调度员的角色。
/**
* Class Scheduler
*/
Class Scheduler
{
/**
* @var SplQueue
*/
protected $taskQueue;
/**
* @var int
*/
protected $tid = 0;
/**
* Scheduler constructor.
*/
public function __construct()
{
/* 原理就是维护了一个队列，
* 前面说过，从编程角度上看，协程的思想本质上就是控制流的主动让出（yield）和恢复（resume）机制
* */
$this->taskQueue = new SplQueue();
}
/**
* 增加一个任务
*
* @param Generator $task
* @return int
*/
public function addTask(Generator $task)
{
$tid = $this->tid;
$task = new Task($tid, $task);
$this->taskQueue->enqueue($task);
$this->tid++;
return $tid;
}
/**
* 把任务进入队列
*
* @param Task $task
*/
public function schedule(Task $task)
{
$this->taskQueue->enqueue($task);
}
/**
* 运行调度器
*/
public function run()
{
while (!$this->taskQueue->isEmpty()) {
// 任务出队
$task = $this->taskQueue->dequeue();
$res = $task->run(); // 运行任务直到 yield
if (!$task->isFinished()) {
$this->schedule($task); // 任务如果还没完全执行完毕，入队等下次执行
}
}
}
}这样我们基本就实现了一个协程调度器。
你可以使用下面的代码来测试：
<?php
function task1() {
for ($i = 1; $i <= 10; ++$i) {
echo "This is task 1 iteration $i.\n";
yield; // 主动让出CPU的执行权
}
}
function task2() {
for ($i = 1; $i <= 5; ++$i) {
echo "This is task 2 iteration $i.\n";
yield; // 主动让出CPU的执行权
}
}
$scheduler = new Scheduler; // 实例化一个调度器
$scheduler->addTask(task1()); // 添加不同的闭包函数作为任务
$scheduler->addTask(task2());
$scheduler->run();关键说下在哪里能用得到PHP协程。
function task1() {
/* 这里有一个远程任务，需要耗时10s，可能是一个远程机器抓取分析远程网址的任务，我们只要提交最后去远程机器拿结果就行了 */
remote_task_commit();
// 这时候请求发出后，我们不要在这里等，主动让出CPU的执行权给task2运行，他不依赖这个结果
yield;
yield (remote_task_receive());
...
}
function task2() {
for ($i = 1; $i <= 5; ++$i) {
echo "This is task 2 iteration $i.\n";
yield; // 主动让出CPU的执行权
}
}这样就提高了程序的执行效率。
关于『系统调用』的实现，鸟哥已经讲得很明白，我这里不再说明。
协程堆栈
鸟哥文中还有一个协程堆栈的例子。
我们上面说过了，如果在函数中使用了yield，就不能当做函数使用。
所以你在一个协程函数中嵌套另外一个协程函数：
<?php
function echoTimes($msg, $max) {
for ($i = 1; $i <= $max; ++$i) {
echo "$msg iteration $i\n";
yield;
}
}
function task() {
echoTimes('foo', 10); // print foo ten times
echo "---\n";
echoTimes('bar', 5); // print bar five times
yield; // force it to be a coroutine
}
$scheduler = new Scheduler;
$scheduler->addTask(task());
$scheduler->run();这里的echoTimes是执行不了的！所以就需要协程堆栈。
不过没关系，我们改一改我们刚刚的代码。
把Task中的初始化方法改下，因为我们在运行一个Task的时候，我们要分析出他包含了哪些子协程，然后将子协程用一个堆栈保存。（C语言学的好的同学自然能理解这里，不理解的同学我建议去了解下进程的内存模型是怎么处理函数调用）
/**
* Task constructor.
* @param $taskId
* @param Generator $coroutine
*/
public function __construct($taskId, Generator $coroutine)
{
$this->taskId = $taskId;
// $this->coroutine = $coroutine;
// 换成这个，实际Task->run的就是stackedCoroutine这个函数，不是$coroutine保存的闭包函数了
$this->coroutine = stackedCoroutine($coroutine);
}当Task->run()的时候，一个循环来分析：
/**
* @param Generator $gen
*/
function stackedCoroutine(Generator $gen)
{
$stack = new SplStack;
// 不断遍历这个传进来的生成器
for (; ;) {
// $gen可以理解为指向当前运行的协程闭包函数（生成器）
$value = $gen->current(); // 获取中断点，也就是yield出来的值
if ($value instanceof Generator) {
// 如果是也是一个生成器，这就是子协程了，把当前运行的协程入栈保存
$stack->push($gen);
$gen = $value; // 把子协程函数给gen，继续执行，注意接下来就是执行子协程的流程了
continue;
}
// 我们对子协程返回的结果做了封装，下面讲
$isReturnValue = $value instanceof CoroutineReturnValue; // 子协程返回`$value`需要主协程帮忙处理
if (!$gen->valid() || $isReturnValue) {
if ($stack->isEmpty()) {
return;
}
// 如果是gen已经执行完毕，或者遇到子协程需要返回值给主协程去处理
$gen = $stack->pop(); //出栈，得到之前入栈保存的主协程
$gen->send($isReturnValue ? $value->getValue() : NULL); // 调用主协程处理子协程的输出值
continue;
}
$gen->send(yield $gen->key() => $value); // 继续执行子协程
}
}然后我们增加echoTime的结束标示：
class CoroutineReturnValue {
protected $value;
public function __construct($value) {
$this->value = $value;
}
// 获取能把子协程的输出值给主协程，作为主协程的send参数
public function getValue() {
return $this->value;
}
}
function retval($value) {
return new CoroutineReturnValue($value);
}然后修改echoTimes：
function echoTimes($msg, $max) {
for ($i = 1; $i <= $max; ++$i) {
echo "$msg iteration $i\n";
yield;
}
yield retval(""); // 增加这个作为结束标示
}Task变为：
function task1()
{
yield echoTimes('bar', 5);
}这样就实现了一个协程堆栈，现在你可以举一反三了。
PHP7中yield from关键字
PHP7中增加了yield from，所以我们不需要自己实现携程堆栈，真是太好了。
把Task的构造函数改回去：
public function __construct($taskId, Generator $coroutine)
{
$this->taskId = $taskId;
$this->coroutine = $coroutine;
// $this->coroutine = stackedCoroutine($coroutine); //不需要自己实现了，改回之前的
}echoTimes函数：
function echoTimes($msg, $max) {
for ($i = 1; $i <= $max; ++$i) {
echo "$msg iteration $i\n";
yield;
}
}task1生成器：
function task1()
{
yield from echoTimes('bar', 5);
}这样，轻松调用子协程。
建议不要使用PHP的Yield来实现协程，推荐使用swoole，2.0已经支持了协程，并附带了部分案例。
推荐教程：《PHP7教程》
以上是在PHP7中實現協程的詳細內容。更多資訊請關注PHP中文網其他相關文章！