PHP協(xié)程實(shí)現(xiàn)過(guò)程詳解
實(shí)現(xiàn) PHP 協(xié)程需要了解的基本內(nèi)容。
多進(jìn)程/線程
最早的服務(wù)器端程序都是通過(guò)多進(jìn)程、多線程來(lái)解決并發(fā)IO的問(wèn)題。進(jìn)程模型出現(xiàn)的最早，從Unix 系統(tǒng)誕生就開(kāi)始有了進(jìn)程的概念。最早的服務(wù)器端程序一般都是 Accept 一個(gè)客戶端連接就創(chuàng)建一個(gè)進(jìn)程，然后子進(jìn)程進(jìn)入循環(huán)同步阻塞地與客戶端連接進(jìn)行交互，收發(fā)處理數(shù)據(jù)。
多線程模式出現(xiàn)要晚一些，線程與進(jìn)程相比更輕量，而且線程之間共享內(nèi)存堆棧，所以不同的線程之間交互非常容易實(shí)現(xiàn)。比如實(shí)現(xiàn)一個(gè)聊天室，客戶端連接之間可以交互，聊天室中的玩家可以任意的其他人發(fā)消息。用多線程模式實(shí)現(xiàn)非常簡(jiǎn)單，線程中可以直接向某一個(gè)客戶端連接發(fā)送數(shù)據(jù)。而多進(jìn)程模式就要用到管道、消息隊(duì)列、共享內(nèi)存等等統(tǒng)稱進(jìn)程間通信（IPC）復(fù)雜的技術(shù)才能實(shí)現(xiàn)。
最簡(jiǎn)單的多進(jìn)程服務(wù)端模型
$serv = stream_socket_server("tcp://0.0.0.0:8000", $errno, $errstr)
or die("Create server failed");
while(1) {
$conn = stream_socket_accept($serv);
if (pcntl_fork() == 0) {
$request = fread($conn);
// do something
// $response = "hello world";
fwrite($response);
fclose($conn);
exit(0);
}
}
多進(jìn)程/線程模型的流程是：
創(chuàng)建一個(gè) socket，綁定服務(wù)器端口（bind），監(jiān)聽(tīng)端口（listen），在 PHP 中用 stream_socket_server 一個(gè)函數(shù)就能完成上面 3 個(gè)步驟，當(dāng)然也可以使用更底層的sockets 擴(kuò)展分別實(shí)現(xiàn)。
進(jìn)入 while 循環(huán)，阻塞在 accept 操作上，等待客戶端連接進(jìn)入。此時(shí)程序會(huì)進(jìn)入睡眠狀態(tài)，直到有新的客戶端發(fā)起 connect 到服務(wù)器，操作系統(tǒng)會(huì)喚醒此進(jìn)程。accept 函數(shù)返回客戶端連接的 socket 主進(jìn)程在多進(jìn)程模型下通過(guò) fork（php: pcntl_fork）創(chuàng)建子進(jìn)程，多線程模型下使用 pthread_create（php: new Thread）創(chuàng)建子線程。
下文如無(wú)特殊聲明將使用進(jìn)程同時(shí)表示進(jìn)程/線程。
子進(jìn)程創(chuàng)建成功后進(jìn)入 while 循環(huán)，阻塞在 recv（php:fread）調(diào)用上，等待客戶端向服務(wù)器發(fā)送數(shù)據(jù)。收到數(shù)據(jù)后服務(wù)器程序進(jìn)行處理然后使用 send（php: fwrite）向客戶端發(fā)送響應(yīng)。長(zhǎng)連接的服務(wù)會(huì)持續(xù)與客戶端交互，而短連接服務(wù)一般收到響應(yīng)就會(huì) close。
當(dāng)客戶端連接關(guān)閉時(shí)，子進(jìn)程退出并銷毀所有資源，主進(jìn)程會(huì)回收掉此子進(jìn)程。
這種模式***的問(wèn)題是，進(jìn)程創(chuàng)建和銷毀的開(kāi)銷很大。所以上面的模式?jīng)]辦法應(yīng)用于非常繁忙的服務(wù)器程序。對(duì)應(yīng)的改進(jìn)版解決了此問(wèn)題，這就是經(jīng)典的 Leader-Follower 模型。
$serv = stream_socket_server("tcp://0.0.0.0:8000", $errno, $errstr)
or die("Create server failed");
for($i = 0; $i < 32; $i++) {
if (pcntl_fork() == 0) {
while(1) {
$conn = stream_socket_accept($serv);
if ($conn == false) continue;
// do something
$request = fread($conn);
// $response = "hello world";
fwrite($response);
fclose($conn);
}
exit(0);
}
}
它的特點(diǎn)是程序啟動(dòng)后就會(huì)創(chuàng)建 N 個(gè)進(jìn)程。每個(gè)子進(jìn)程進(jìn)入 Accept，等待新的連接進(jìn)入。當(dāng)客戶端連接到服務(wù)器時(shí)，其中一個(gè)子進(jìn)程會(huì)被喚醒，開(kāi)始處理客戶端請(qǐng)求，并且不再接受新的 TCP 連接。當(dāng)此連接關(guān)閉時(shí)，子進(jìn)程會(huì)釋放，重新進(jìn)入 Accept，參與處理新的連接。
這個(gè)模型的優(yōu)勢(shì)是完全可以復(fù)用進(jìn)程，沒(méi)有額外消耗，性能非常好。很多常見(jiàn)的服務(wù)器程序都是基于此模型的，比如 Apache、PHP-FPM。
多進(jìn)程模型也有一些缺點(diǎn)。
這種模型嚴(yán)重依賴進(jìn)程的數(shù)量解決并發(fā)問(wèn)題，一個(gè)客戶端連接就需要占用一個(gè)進(jìn)程，工作進(jìn)程的數(shù)量有多少，并發(fā)處理能力就有多少。操作系統(tǒng)可以創(chuàng)建的進(jìn)程數(shù)量是有限的。
啟動(dòng)大量進(jìn)程會(huì)帶來(lái)額外的進(jìn)程調(diào)度消耗。數(shù)百個(gè)進(jìn)程時(shí)可能進(jìn)程上下文切換調(diào)度消耗占 CPU 不到 1% 可以忽略不計(jì)，如果啟動(dòng)數(shù)千甚至數(shù)萬(wàn)個(gè)進(jìn)程，消耗就會(huì)直線上升。調(diào)度消耗可能占到 CPU 的百分之幾十甚至 100%。
并行和并發(fā)
談到多進(jìn)程以及類似同時(shí)執(zhí)行多個(gè)任務(wù)的模型，就不得不先談?wù)劜⑿泻筒l(fā)。
并發(fā)（Concurrency）
是指能處理多個(gè)同時(shí)活動(dòng)的能力，并發(fā)事件之間不一定要同一時(shí)刻發(fā)生。
并行（Parallesim）
是指同時(shí)刻發(fā)生的兩個(gè)并發(fā)事件，具有并發(fā)的含義，但并發(fā)不一定并行。
區(qū)別
- 『并發(fā)』指的是程序的結(jié)構(gòu)，『并行』指的是程序運(yùn)行時(shí)的狀態(tài)
- 『并行』一定是并發(fā)的，『并行』是『并發(fā)』設(shè)計(jì)的一種
- 單線程永遠(yuǎn)無(wú)法達(dá)到『并行』狀態(tài)
正確的并發(fā)設(shè)計(jì)的標(biāo)準(zhǔn)是：
使多個(gè)操作可以在重疊的時(shí)間段內(nèi)進(jìn)行。
two tasks can start, run, and complete in overlapping time periods
參考：
迭代器 & 生成器
在了解 PHP 協(xié)程前，還有 迭代器 和 生成器 這兩個(gè)概念需要先認(rèn)識(shí)一下。
迭代器
PHP5 開(kāi)始內(nèi)置了 Iterator 即迭代器接口，所以如果你定義了一個(gè)類，并實(shí)現(xiàn)了Iterator 接口，那么你的這個(gè)類對(duì)象就是 ZEND_ITER_OBJECT 即可迭代的，否則就是 ZEND_ITER_PLAIN_OBJECT。
對(duì)于 ZEND_ITER_PLAIN_OBJECT 的類，foreach 會(huì)獲取該對(duì)象的默認(rèn)屬性數(shù)組，然后對(duì)該數(shù)組進(jìn)行迭代。
而對(duì)于 ZEND_ITER_OBJECT 的類對(duì)象，則會(huì)通過(guò)調(diào)用對(duì)象實(shí)現(xiàn)的 Iterator 接口相關(guān)函數(shù)來(lái)進(jìn)行迭代。
任何實(shí)現(xiàn)了 Iterator 接口的類都是可迭代的，即都可以用 foreach 語(yǔ)句來(lái)遍歷。
Iterator 接口
interface Iterator extends Traversable {
// 獲取當(dāng)前內(nèi)部標(biāo)量指向的元素的數(shù)據(jù)
public mixed current() // 獲取當(dāng)前標(biāo)量 public scalar key() // 移動(dòng)到下一個(gè)標(biāo)量 public void next() // 重置標(biāo)量 public void rewind() // 檢查當(dāng)前標(biāo)量是否有效 public boolean valid() }
常規(guī)實(shí)現(xiàn) range 函數(shù)
PHP 自帶的 range 函數(shù)原型：
range — 根據(jù)范圍創(chuàng)建數(shù)組，包含指定的元素
array range (mixed $start , mixed $end [, number $step = 1 ])
建立一個(gè)包含指定范圍單元的數(shù)組。
在不使用迭代器的情況要實(shí)現(xiàn)一個(gè)和 PHP 自帶的 range 函數(shù)類似的功能，可能會(huì)這么寫(xiě)：
function range ($start, $end, $step = 1) {
$ret = [];
for ($i = $start; $i <= $end; $i += $step) {
$ret[] = $i;
}
return $ret;
}
需要將生成的所有元素放在內(nèi)存數(shù)組中，如果需要生成一個(gè)非常大的集合，則會(huì)占用巨大的內(nèi)存。
迭代器實(shí)現(xiàn) xrange 函數(shù)
來(lái)看看迭代實(shí)現(xiàn)的 range，我們叫做 xrange，他實(shí)現(xiàn)了 Iterator 接口必須的 5 個(gè)方法：
class Xrange implements Iterator {
protected $start;
protected $limit;
protected $step;
protected $current;
public function __construct($start, $limit, $step = 1) {
$this->start = $start;
$this->limit = $limit;
$this->step = $step;
}
public function rewind() {
$this->current = $this->start;
}
public function next() {
$this->current += $this->step;
}
public function current() {
return $this->current;
}
public function key() {
return $this->current + 1;
}
public function valid() {
return $this->current <= $this->limit;
}
}
使用時(shí)代碼如下：
foreach (new Xrange(0, 9) as $key => $val) {
echo $key, ' ', $val, "\n";
}
輸出：
0 0 1 1 2 2 3 3 4 4 5 5 6 6 7 7 8 8 9 9
看上去功能和 range() 函數(shù)所做的一致，不同點(diǎn)在于迭代的是一個(gè) 對(duì)象(Object) 而不是數(shù)組：
var_dump(new Xrange(0, 9));
輸出：
object(Xrange)#1 (4) {
["start":protected]=>
int(0)
["limit":protected]=>
int(9)
["step":protected]=>
int(1)
["current":protected]=>
NULL
}
另外，內(nèi)存的占用情況也完全不同：
// range
$startMemory = memory_get_usage();
$arr = range(0, 500000);
echo 'range(): ', memory_get_usage() - $startMemory, " bytes\n";
unset($arr);
// xrange
$startMemory = memory_get_usage();
$arr = new Xrange(0, 500000);
echo 'xrange(): ', memory_get_usage() - $startMemory, " bytes\n";
輸出：
xrange(): 624 bytes
range(): 72194784 bytes
range() 函數(shù)在執(zhí)行后占用了 50W 個(gè)元素內(nèi)存空間，而 xrange 對(duì)象在整個(gè)迭代過(guò)程中只占用一個(gè)對(duì)象的內(nèi)存。
Yii2 Query
在喜聞樂(lè)見(jiàn)的各種 PHP 框架里有不少生成器的實(shí)例，比如 Yii2 中用來(lái)構(gòu)建 SQL 語(yǔ)句的 \yii\db\Query類：
$query = (new \yii\db\Query)->from('user');
// yii\db\BatchQueryResult
foreach ($query->batch() as $users) {
// 每次循環(huán)得到多條 user 記錄
}
來(lái)看一下 batch() 做了什么：
/** * Starts a batch query. * * A batch query supports fetching data in batches, which can keep the memory usage under a limit. * This method will return a [[BatchQueryResult]] object which implements the [[\Iterator]] interface * and can be traversed to retrieve the data in batches. * * For example, * * * $query = (new Query)->from('user'); * foreach ($query->batch() as $rows) { * // $rows is an array of 10 or fewer rows from user table * } * * * @param integer $batchSize the number of records to be fetched in each batch. * @param Connection $db the database connection. If not set, the "db" application component will be used. * @return BatchQueryResult the batch query result. It implements the [[\Iterator]] interface * and can be traversed to retrieve the data in batches. */
public function batch($batchSize = 100, $db = null) {
return Yii::createObject([
'class' => BatchQueryResult::className(),
'query' => $this,
'batchSize' => $batchSize,
'db' => $db,
'each' => false,
]);
}
實(shí)際上返回了一個(gè) BatchQueryResult 類，類的源碼實(shí)現(xiàn)了 Iterator 接口 5 個(gè)關(guān)鍵方法：
class BatchQueryResult extends Object implements \Iterator {
public $db;
public $query;
public $batchSize = 100;
public $each = false;
private $_dataReader;
private $_batch;
private $_value;
private $_key;
/** * Destructor. */
public function __destruct() {
// make sure cursor is closed
$this->reset();
}
/** * Resets the batch query. * This method will clean up the existing batch query so that a new batch query can be performed. */
public function reset() {
if ($this->_dataReader !== null) {
$this->_dataReader->close();
}
$this->_dataReader = null;
$this->_batch = null;
$this->_value = null;
$this->_key = null;
}
/** * Resets the iterator to the initial state. * This method is required by the interface [[\Iterator]]. */
public function rewind() {
$this->reset();
$this->next();
}
/** * Moves the internal pointer to the next dataset. * This method is required by the interface [[\Iterator]]. */
public function next() {
if ($this->_batch === null || !$this->each || $this->each && next($this->_batch) === false) {
$this->_batch = $this->fetchData();
reset($this->_batch);
}
if ($this->each) {
$this->_value = current($this->_batch);
if ($this->query->indexBy !== null) {
$this->_key = key($this->_batch);
} elseif (key($this->_batch) !== null) {
$this->_key++;
} else {
$this->_key = null;
}
} else {
$this->_value = $this->_batch;
$this->_key = $this->_key === null ? 0 : $this->_key + 1;
}
}
/** * Fetches the next batch of data. * @return array the data fetched */
protected function fetchData() {
// ...
}
/** * Returns the index of the current dataset. * This method is required by the interface [[\Iterator]]. * @return integer the index of the current row. */
public function key() {
return $this->_key;
}
/** * Returns the current dataset. * This method is required by the interface [[\Iterator]]. * @return mixed the current dataset. */
public function current() {
return $this->_value;
}
/** * Returns whether there is a valid dataset at the current position. * This method is required by the interface [[\Iterator]]. * @return boolean whether there is a valid dataset at the current position. */
public function valid() {
return !empty($this->_batch);
}
}
以迭代器的方式實(shí)現(xiàn)了類似分頁(yè)取的效果，同時(shí)避免了一次性取出所有數(shù)據(jù)占用太多的內(nèi)存空間。
迭代器使用場(chǎng)景
- 使用返回迭代器的包或庫(kù)時(shí)（如 PHP5 中的 SPL 迭代器）
- 無(wú)法在一次調(diào)用獲取所需的所有元素時(shí)
- 要處理數(shù)量巨大的元素時(shí)（數(shù)據(jù)庫(kù)中要處理的結(jié)果集內(nèi)容超過(guò)內(nèi)存）
- …
生成器
雖然迭代器僅需繼承接口即可實(shí)現(xiàn)，但畢竟需要定義一整個(gè)類然后實(shí)現(xiàn)接口的所有方法，實(shí)在是不怎么方便。
生成器則提供了一種更簡(jiǎn)單的方式來(lái)實(shí)現(xiàn)簡(jiǎn)單的對(duì)象迭代，相比定義類來(lái)實(shí)現(xiàn)
Iterator接口的方式，性能開(kāi)銷和復(fù)雜度大大降低。
生成器允許在 foreach 代碼塊中迭代一組數(shù)據(jù)而不需要?jiǎng)?chuàng)建任何數(shù)組。一個(gè)生成器函數(shù)，就像一個(gè)普通的有返回值的自定義函數(shù)類似，但普通函數(shù)只返回一次, 而生成器可以根據(jù)需要通過(guò) yield 關(guān)鍵字返回多次，以便連續(xù)生成需要迭代返回的值。
一個(gè)最簡(jiǎn)單的例子就是使用生成器來(lái)重新實(shí)現(xiàn) xrange() 函數(shù)。效果和上面我們用迭代器實(shí)現(xiàn)的差不多，但實(shí)現(xiàn)起來(lái)要簡(jiǎn)單的多。
生成器實(shí)現(xiàn) xrange 函數(shù)
function xrange($start, $limit, $step = 1) {
for ($i = 0; $i < $limit; $i += $step) {
yield $i + 1 => $i;
}
}
foreach (xrange(0, 9) as $key => $val) {
printf("%d %d \n", $key, $val);
}
// 輸出
// 1 0
// 2 1
// 3 2
// 4 3
// 5 4
// 6 5
// 7 6
// 8 7
// 9 8
實(shí)際上生成器生成的正是一個(gè)迭代器對(duì)象實(shí)例，該迭代器對(duì)象繼承了 Iterator 接口，同時(shí)也包含了生成器對(duì)象自有的接口，具體可以參考 Generator 類的定義以及語(yǔ)法參考。
同時(shí)需要注意的是：
yield 關(guān)鍵字
需要注意的是 yield 關(guān)鍵字，這是生成器的關(guān)鍵。通過(guò)上面的例子可以看出，yield 會(huì)將當(dāng)前產(chǎn)生的值傳遞給 foreach，換句話說(shuō)，foreach 每一次迭代過(guò)程都會(huì)從 yield 處取一個(gè)值，直到整個(gè)遍歷過(guò)程不再能執(zhí)行到 yield 時(shí)遍歷結(jié)束，此時(shí)生成器函數(shù)簡(jiǎn)單的退出，而調(diào)用生成器的上層代碼還可以繼續(xù)執(zhí)行，就像一個(gè)數(shù)組已經(jīng)被遍歷完了。
yield 最簡(jiǎn)單的調(diào)用形式看起來(lái)像一個(gè) return 申明，不同的是 yield 暫停當(dāng)前過(guò)程的執(zhí)行并返回值，而 return 是中斷當(dāng)前過(guò)程并返回值。暫停當(dāng)前過(guò)程，意味著將處理權(quán)轉(zhuǎn)交由上一級(jí)繼續(xù)進(jìn)行，直到上一級(jí)再次調(diào)用被暫停的過(guò)程，該過(guò)程又會(huì)從上一次暫停的位置繼續(xù)執(zhí)行。這像是什么呢？如果之前已經(jīng)在鳥(niǎo)哥的文章中粗略看過(guò)，應(yīng)該知道這很像操作系統(tǒng)的進(jìn)程調(diào)度，多個(gè)進(jìn)程在一個(gè) CPU 核心上執(zhí)行，在系統(tǒng)調(diào)度下每一個(gè)進(jìn)程執(zhí)行一段指令就被暫停，切換到下一個(gè)進(jìn)程，這樣外部用戶看起來(lái)就像是同時(shí)在執(zhí)行多個(gè)任務(wù)。
但僅僅如此還不夠，yield 除了可以返回值以外，還能接收值，也就是可以在兩個(gè)層級(jí)間實(shí)現(xiàn)雙向通信。
來(lái)看看如何傳遞一個(gè)值給 yield：
function printer() {
while (true) {
printf("receive: %s\n", yield);
}
}
$printer = printer();
$printer->send('hello');
$printer->send('world');
// 輸出
receive: hello
receive: world
根據(jù) PHP 官方文檔的描述可以知道 Generator 對(duì)象除了實(shí)現(xiàn) Iterator 接口中的必要方法以外，還有一個(gè) send 方法，這個(gè)方法就是向 yield 語(yǔ)句處傳遞一個(gè)值，同時(shí)從 yield 語(yǔ)句處繼續(xù)執(zhí)行，直至再次遇到 yield 后控制權(quán)回到外部。
既然 yield 可以在其位置中斷并返回或者接收一個(gè)值，那能不能同時(shí)進(jìn)行接收和返回呢？當(dāng)然，這也是實(shí)現(xiàn)協(xié)程的根本。對(duì)上述代碼做出修改：
function printer() {
$i = 0;
while (true) {
printf("receive: %s\n", (yield ++$i));
}
}
$printer = printer();
printf("%d\n", $printer->current());
$printer->send('hello');
printf("%d\n", $printer->current());
$printer->send('world');
printf("%d\n", $printer->current());
// 輸出
1
receive: hello
2
receive: world
3
這是另一個(gè)例子：
function gen() {
$ret = (yield 'yield1');
var_dump($ret);
$ret = (yield 'yield2');
var_dump($ret);
}
$gen = gen();
var_dump($gen->current()); // string(6) "yield1"
var_dump($gen->send('ret1')); // string(4) "ret1" (***個(gè) var_dump)
// string(6) "yield2" (繼續(xù)執(zhí)行到第二個(gè) yield，吐出了返回值)
var_dump($gen->send('ret2')); // string(4) "ret2" (第二個(gè) var_dump)
// NULL (var_dump 之后沒(méi)有其他語(yǔ)句，所以這次 ->send() 的返回值為 null)
current 方法是迭代器 Iterator 接口必要的方法，foreach 語(yǔ)句每一次迭代都會(huì)通過(guò)其獲取當(dāng)前值，而后調(diào)用迭代器的 next 方法。在上述例子里則是手動(dòng)調(diào)用了 current 方法獲取值。
上述例子已經(jīng)足以表示 yield 能夠作為實(shí)現(xiàn)雙向通信的工具，也就是具備了后續(xù)實(shí)現(xiàn)協(xié)程的基本條件。
上面的例子如果***次接觸并稍加思考，不免會(huì)疑惑為什么一個(gè) yield 既是語(yǔ)句又是表達(dá)式，而且這兩種情況還同時(shí)存在：
- 對(duì)于所有在生成器函數(shù)中出現(xiàn)的
yield，首先它都是語(yǔ)句，而跟在yield后面的任何表達(dá)式的值將作為調(diào)用生成器函數(shù)的返回值，如果yield后面沒(méi)有任何表達(dá)式（變量、常量都是表達(dá)式），那么它會(huì)返回NULL，這一點(diǎn)和return語(yǔ)句一致。 yield也是表達(dá)式，它的值就是send函數(shù)傳過(guò)來(lái)的值（相當(dāng)于一個(gè)特殊變量，只不過(guò)賦值是通過(guò)send函數(shù)進(jìn)行的）。只要調(diào)用send方法，并且生成器對(duì)象的迭代并未終結(jié)，那么當(dāng)前位置的yield就會(huì)得到send方法傳遞過(guò)來(lái)的值，這和生成器函數(shù)有沒(méi)有把這個(gè)值賦值給某個(gè)變量沒(méi)有任何關(guān)系。
這個(gè)地方可能需要仔細(xì)品味上面兩個(gè) send() 方法的例子才能理解。但可以簡(jiǎn)單的記?。?/p>
除了 send() 方法，還有一種控制生成器執(zhí)行的方法是 next() 函數(shù)：
Next()，恢復(fù)生成器函數(shù)的執(zhí)行直到下一個(gè)yieldSend()，向生成器傳入一個(gè)值，恢復(fù)執(zhí)行直到下一個(gè)yield
協(xié)程
對(duì)于單核處理器，多進(jìn)程實(shí)現(xiàn)多任務(wù)的原理是讓操作系統(tǒng)給一個(gè)任務(wù)每次分配一定的 CPU 時(shí)間片，然后中斷、讓下一個(gè)任務(wù)執(zhí)行一定的時(shí)間片接著再中斷并繼續(xù)執(zhí)行下一個(gè)，如此反復(fù)。由于切換執(zhí)行任務(wù)的速度非常快，給外部用戶的感受就是多個(gè)任務(wù)的執(zhí)行是同時(shí)進(jìn)行的。
多進(jìn)程的調(diào)度是由操作系統(tǒng)來(lái)實(shí)現(xiàn)的，進(jìn)程自身不能控制自己何時(shí)被調(diào)度，也就是說(shuō)：
進(jìn)程的調(diào)度是由外層調(diào)度器搶占式實(shí)現(xiàn)的
而協(xié)程要求當(dāng)前正在運(yùn)行的任務(wù)自動(dòng)把控制權(quán)回傳給調(diào)度器，這樣就可以繼續(xù)運(yùn)行其他任務(wù)。這與『搶占式』的多任務(wù)正好相反, 搶占多任務(wù)的調(diào)度器可以強(qiáng)制中斷正在運(yùn)行的任務(wù), 不管它自己有沒(méi)有意愿?！簠f(xié)作式多任務(wù)』在 Windows 的早期版本 (windows95) 和 Mac OS 中有使用, 不過(guò)它們后來(lái)都切換到『搶占式多任務(wù)』了。理由相當(dāng)明確：如果僅依靠程序自動(dòng)交出控制的話，那么一些惡意程序?qū)?huì)很容易占用全部 CPU 時(shí)間而不與其他任務(wù)共享。
協(xié)程的調(diào)度是由協(xié)程自身主動(dòng)讓出控制權(quán)到外層調(diào)度器實(shí)現(xiàn)的
回到剛才生成器實(shí)現(xiàn) xrange 函數(shù)的例子，整個(gè)執(zhí)行過(guò)程的交替可以用下圖來(lái)表示：
協(xié)程可以理解為純用戶態(tài)的線程，通過(guò)協(xié)作而不是搶占來(lái)進(jìn)行任務(wù)切換。相對(duì)于進(jìn)程或者線程，協(xié)程所有的操作都可以在用戶態(tài)而非操作系統(tǒng)內(nèi)核態(tài)完成，創(chuàng)建和切換的消耗非常低。
簡(jiǎn)單的說(shuō) Coroutine（協(xié)程） 就是提供一種方法來(lái)中斷當(dāng)前任務(wù)的執(zhí)行，保存當(dāng)前的局部變量，下次再過(guò)來(lái)又可以恢復(fù)當(dāng)前局部變量繼續(xù)執(zhí)行。
我們可以把大任務(wù)拆分成多個(gè)小任務(wù)輪流執(zhí)行，如果有某個(gè)小任務(wù)在等待系統(tǒng) IO，就跳過(guò)它，執(zhí)行下一個(gè)小任務(wù)，這樣往復(fù)調(diào)度，實(shí)現(xiàn)了 IO 操作和 CPU 計(jì)算的并行執(zhí)行，總體上就提升了任務(wù)的執(zhí)行效率，這也便是協(xié)程的意義。
PHP 協(xié)程和 yield
PHP 從 5.5 開(kāi)始支持生成器及 yield 關(guān)鍵字，而 PHP 協(xié)程則由 yield 來(lái)實(shí)現(xiàn)。
要理解協(xié)程，首先要理解：代碼是代碼，函數(shù)是函數(shù)。函數(shù)包裹的代碼賦予了這段代碼附加的意義：不管是否顯式的指明返回值，當(dāng)函數(shù)內(nèi)的代碼塊執(zhí)行完后都會(huì)返回到調(diào)用層。而當(dāng)調(diào)用層調(diào)用某個(gè)函數(shù)的時(shí)候，必須等這個(gè)函數(shù)返回，當(dāng)前函數(shù)才能繼續(xù)執(zhí)行，這就構(gòu)成了后進(jìn)先出，也就是 Stack。
而協(xié)程包裹的代碼，不是函數(shù)，不完全遵守函數(shù)的附加意義，協(xié)程執(zhí)行到某個(gè)點(diǎn)，協(xié)會(huì)協(xié)程會(huì) yield返回一個(gè)值然后掛起，而不是 return 一個(gè)值然后結(jié)束，當(dāng)再次調(diào)用協(xié)程的時(shí)候，會(huì)在上次 yield 的點(diǎn)繼續(xù)執(zhí)行。
所以協(xié)程違背了通常操作系統(tǒng)和 x86 的 CPU 認(rèn)定的代碼執(zhí)行方式，也就是 Stack 的這種執(zhí)行方式，需要運(yùn)行環(huán)境（比如 php，python 的 yield 和 golang 的 goroutine）自己調(diào)度，來(lái)實(shí)現(xiàn)任務(wù)的中斷和恢復(fù)，具體到 PHP，就是靠 yield 來(lái)實(shí)現(xiàn)。
堆棧式調(diào)用 和 協(xié)程調(diào)用的對(duì)比：
結(jié)合之前的例子，可以總結(jié)一下 yield 能做的就是：
- 實(shí)現(xiàn)不同任務(wù)間的主動(dòng)讓位、讓行，把控制權(quán)交回給任務(wù)調(diào)度器。
- 通過(guò)
send()實(shí)現(xiàn)不同任務(wù)間的雙向通信，也就可以實(shí)現(xiàn)任務(wù)和調(diào)度器之間的通信。
yield 就是 PHP 實(shí)現(xiàn)協(xié)程的方式。
協(xié)程多任務(wù)調(diào)度
下面是雄文 Cooperative multitasking using coroutines (in PHP!) 里一個(gè)簡(jiǎn)單但完整的例子，來(lái)展示如何具體的在 PHP 里實(shí)現(xiàn)協(xié)程任務(wù)的調(diào)度。
首先是一個(gè)任務(wù)類：
Task
class Task {
// 任務(wù) ID
protected $taskId;
// 協(xié)程對(duì)象
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之前, 當(dāng)?shù)鞅粍?chuàng)建后***次 yield 之前，一個(gè) renwind() 方法會(huì)被隱式調(diào)用
// 所以實(shí)際上發(fā)生的應(yīng)該類似:
// $this->coroutine->rewind();
// $this->coroutine->send();
// 這樣 renwind 的執(zhí)行將會(huì)導(dǎo)致***個(gè) yield 被執(zhí)行, 并且忽略了他的返回值.
// 真正當(dāng)我們調(diào)用 yield 的時(shí)候, 我們得到的是第二個(gè)yield的值，導(dǎo)致***個(gè)yield的值被忽略。
// 所以這個(gè)加上一個(gè)是否***次 yield 的判斷來(lái)避免這個(gè)問(wèn)題
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();
}
}
接下來(lái)是調(diào)度器，比 foreach 是要復(fù)雜一點(diǎn)，但好歹也能算個(gè)正兒八經(jīng)的 Scheduler :)
Scheduler
class Scheduler {
protected $maxTaskId = 0;
protected $taskMap = []; // taskId => task
protected $taskQueue;
public function __construct() {
$this->taskQueue = new SplQueue();
}
// （使用下一個(gè)空閑的任務(wù)id）創(chuàng)建一個(gè)新任務(wù),然后把這個(gè)任務(wù)放入任務(wù)map數(shù)組里. 接著它通過(guò)把任務(wù)放入任務(wù)隊(duì)列里來(lái)實(shí)現(xiàn)對(duì)任務(wù)的調(diào)度. 接著run()方法掃描任務(wù)隊(duì)列, 運(yùn)行任務(wù).如果一個(gè)任務(wù)結(jié)束了, 那么它將從隊(duì)列里刪除, 否則它將在隊(duì)列的末尾再次被調(diào)度。
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) {
// 任務(wù)入隊(duì)
$this->queue->enqueue($task);
}
public function run() {
while (!$this->queue->isEmpty()) {
// 任務(wù)出隊(duì)
$task = $this->queue->dequeue();
$task->run();
if ($task->isFinished()) {
unset($this->taskMap[$task->getTaskId()]);
} else {
$this->schedule($task);
}
}
}
}
隊(duì)列可以使每個(gè)任務(wù)獲得同等的 CPU 使用時(shí)間，
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.
結(jié)果正是我們期待的，最初的 5 次迭代，兩個(gè)任務(wù)是交替進(jìn)行的，而在第二個(gè)任務(wù)結(jié)束后，只有***個(gè)任務(wù)繼續(xù)執(zhí)行到結(jié)束。
協(xié)程非阻塞 IO
若想真正的發(fā)揮出協(xié)程的作用，那一定是在一些涉及到阻塞 IO 的場(chǎng)景，我們都知道 Web 服務(wù)器最耗時(shí)的部分通常都是 socket 讀取數(shù)據(jù)等操作上，如果進(jìn)程對(duì)每個(gè)請(qǐng)求都掛起的等待 IO 操作，那處理效率就太低了，接下來(lái)我們看個(gè)支持非阻塞 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 隊(duì)列
$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) {
// 任務(wù)入隊(duì)
$this->queue->enqueue($task);
}
public function run() {
while (!$this->queue->isEmpty()) {
// 任務(wù)出隊(duì)
$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 時(shí), stream_select 為立即返回，為 null 時(shí)則會(huì)阻塞的等，見(jiàn) 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);
}
}
}
/** * 檢查隊(duì)列是否為空，若為空則掛起的執(zhí)行 stream_select，否則檢查完 IO 狀態(tài)立即返回，詳見(jiàn) ioPoll() * 作為任務(wù)加入隊(duì)列后，由于 while true，會(huì)被一直重復(fù)的加入任務(wù)隊(duì)列，實(shí)現(xiàn)每次任務(wù)前檢查 IO 狀態(tài) * @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 任務(wù)后先執(zhí)行 withIoPoll() 將 ioPollTask() 作為任務(wù)入隊(duì) * * @return $this */
public function withIoPoll() {
$this->newTask($this->ioPollTask());
return $this;
}
}
這個(gè)版本的 Scheduler 里加入一個(gè)永不退出的任務(wù)，并且通過(guò) stream_select 支持的特性來(lái)實(shí)現(xiàn)快速的來(lái)回檢查各個(gè)任務(wù)的 IO 狀態(tài)，只有 IO 完成的任務(wù)才會(huì)繼續(xù)執(zhí)行，而 IO 還未完成的任務(wù)則會(huì)跳過(guò)，完整的代碼和例子可以戳這里。
也就是說(shuō)任務(wù)交替執(zhí)行的過(guò)程中，一旦遇到需要 IO 的部分，調(diào)度器就會(huì)把 CPU 時(shí)間分配給不需要 IO 的任務(wù)，等到當(dāng)前任務(wù)遇到 IO 或者之前的任務(wù) IO 結(jié)束才再次調(diào)度 CPU 時(shí)間，以此實(shí)現(xiàn) CPU 和 IO 并行來(lái)提升執(zhí)行效率，類似下圖：
單任務(wù)改造
如果想將一個(gè)單進(jìn)程任務(wù)改造成并發(fā)執(zhí)行，我們可以選擇改造成多進(jìn)程或者協(xié)程：
- 多進(jìn)程，不改變?nèi)蝿?wù)執(zhí)行的整體過(guò)程，在一個(gè)時(shí)間段內(nèi)同時(shí)執(zhí)行多個(gè)相同的代碼段，調(diào)度權(quán)在 CPU，如果一個(gè)任務(wù)能獨(dú)占一個(gè) CPU 則可以實(shí)現(xiàn)并行。
- 協(xié)程，把原有任務(wù)拆分成多個(gè)小任務(wù)，原有任務(wù)的執(zhí)行流程被改變，調(diào)度權(quán)在進(jìn)程自己，如果有 IO 并且可以實(shí)現(xiàn)異步，則可以實(shí)現(xiàn)并行。
多進(jìn)程改造
協(xié)程改造
協(xié)程（Coroutines）和 Go 協(xié)程（Goroutines）
PHP 的協(xié)程或者其他語(yǔ)言中，比如 Python、Lua 等都有協(xié)程的概念，和 Go 協(xié)程有些相似，不過(guò)有兩點(diǎn)不同：
- Go 協(xié)程意味著并行（或者可以以并行的方式部署，可以用
runtime.GOMAXPROCS()指定可同時(shí)使用的 CPU 個(gè)數(shù)），協(xié)程一般來(lái)說(shuō)只是并發(fā)。 - Go 協(xié)程通過(guò)通道
channel來(lái)通信；協(xié)程通過(guò)yield讓出和恢復(fù)操作來(lái)通信。
Go 協(xié)程比普通協(xié)程更強(qiáng)大，也很容易從協(xié)程的邏輯復(fù)用到 Go 協(xié)程，而且在 Go 的開(kāi)發(fā)中也使用的極為普遍，有興趣的話可以了解一下作為對(duì)比。
結(jié)束
個(gè)人感覺(jué) PHP 的協(xié)程在實(shí)際使用中想要徒手實(shí)現(xiàn)和應(yīng)用并不方便而且場(chǎng)景有限，但了解其概念及實(shí)現(xiàn)原理對(duì)更好的理解并發(fā)不無(wú)裨益。
如果想更多的了解協(xié)程的實(shí)際應(yīng)用場(chǎng)景不妨試試已經(jīng)大名鼎鼎的 Swoole，其對(duì)多種協(xié)議的 client 做了底層的協(xié)程封裝，幾乎可以做到以同步編程的寫(xiě)法實(shí)現(xiàn)協(xié)程異步 IO 的效果。
