詳談PHP7下的協程實現
前言
相信大家都聽說過『協程』這個概念吧。
但是有些同學對這個概念似懂非懂，不知道怎麼實現，怎麼用，用在哪，甚至有些人認為yield就是協程！
我始終相信，如果你無法準確地表達出一個知識點的話，我可以認為你就是不懂。
如果你之前了解過利用PHP實現協程的話，你肯定看過鳥哥的那篇文章：在PHP中使用協程實現多工排程| 風雪之隅
鳥哥這篇文章是從國外的作者翻譯來的，翻譯的簡潔明瞭，也給出了具體的例子了。
我寫這篇文章的目的，是想對鳥哥文章做更加充足的補充，畢竟有部分同學的基礎還是不夠好，看得也是雲頭霧裡的。
什麼是協程
先搞清楚，什麼是協程。
你可能已經聽過『進程』和『執行緒』這兩個概念。
進程就是二進位制可執行檔案在計算機記憶體裡的一個執行範例，就好比你的.exe檔案是個類，進程就是new出來的那個範例。
進程是計算機系統進行資源分配和排程的基本單位（排程單位這裡別糾結執行緒進程的），每個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 "rewindingn";
reset($this->var);
}
public function current() {
$var = current($this->var);
echo "current: $varn";
return $var;
}
public function key() {
$var = key($this->var);
echo "key: $varn";
return $var;
}
public function next() {
$var = next($this->var);
echo "next: $varn";
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: $bn";
}
生成器
可以說之前為了擁有一個能夠被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協程
前面介紹協程的時候說了，協程需要程式設計師自己去編寫排程機制，下面我們來看這個機制怎麼寫。
0）生成器正確使用
既然生成器不能像函數一樣直接呼叫，那麼怎麼才能呼叫呢？
方法如下：
- foreach他
- send($value)
- current / next...
1）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;
}
}
}
2）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的執行權
}
}
這樣就提高了程式的執行效率。
關於『系統呼叫』的實現，鳥哥已經講得很明白，我這裡不再說明。
3）協程堆疊
鳥哥文中還有一個協程堆疊的例子。
我們上面說過了，如果在函數中使用了yield，就不能當做函數使用。
所以你在一個協程函數中巢狀另外一個協程函數：
<?php
function echoTimes($msg, $max) {
for ($i = 1; $i <= $max; ++$i) {
echo "$msg iteration $in";
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 $in";
yield;
}
yield retval(""); // 增加這個作為結束標示
}
Task變為：
function task1()
{
yield echoTimes('bar', 5);
}
這樣就實現了一個協程堆疊，現在你可以舉一反三了。
4）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 $in";
yield;
}
}
task1生成器：
function task1()
{
yield from echoTimes('bar', 5);
}
這樣，輕鬆呼叫子協程。
總結
這下應該明白怎麼實現PHP協程了吧？
End...
以上就是詳談PHP7下的協程實現的詳細內容，更多請關注TW511.COM其它相關文章！