PHP7下協(xié)程的實(shí)現(xiàn)方法詳解

?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";
}

?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";
}

/**
 * Task任務(wù)類
 */
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;
 }
 /**
  * 獲取當(dāng)前的Task的ID
  * 
  * @return mixed
  */
 public function getTaskId()
 {
  return $this->taskId;
 }
 /**
  * 判斷Task執(zhí)行完畢了沒(méi)有
  * 
  * @return bool
  */
 public function isFinished()
 {
  return !$this->coroutine->valid();
 }
 /**
  * 設(shè)置下次要傳給協(xié)程的值，比如 $id = (yield $xxxx)，這個(gè)值就給了$id了
  * 
  * @param $value
  */
 public function setSendValue($value)
 {
  $this->sendValue = $value;
 }
 /**
  * 運(yùn)行任務(wù)
  * 
  * @return mixed
  */
 public function run()
 {
  // 這里要注意，生成器的開始會(huì)reset，所以第一個(gè)值要用current獲取
  if ($this->beforeFirstYield) {
   $this->beforeFirstYield = false;
   return $this->coroutine->current();
  } else {
   // 我們說(shuō)過(guò)了，用send去調(diào)用一個(gè)生成器
   $retval = $this->coroutine->send($this->sendValue);
   $this->sendValue = null;
   return $retval;
  }
 }
}

/**
 * Class Scheduler
 */
Class Scheduler
{
 /**
  * @var SplQueue
  */
 protected $taskQueue;
 /**
  * @var int
  */
 protected $tid = 0;

 /**
  * Scheduler constructor.
  */
 public function __construct()
 {
  /* 原理就是維護(hù)了一個(gè)隊(duì)列，
   * 前面說(shuō)過(guò)，從編程角度上看，協(xié)程的思想本質(zhì)上就是控制流的主動(dòng)讓出（yield）和恢復(fù)（resume）機(jī)制
   * */
  $this->taskQueue = new SplQueue();
 }
 /**
  * 增加一個(gè)任務(wù)
  *
  * @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;
 }
 /**
  * 把任務(wù)進(jìn)入隊(duì)列
  *
  * @param Task $task
  */
 public function schedule(Task $task)
 {
  $this->taskQueue->enqueue($task);
 }
 /**
  * 運(yùn)行調(diào)度器
  */
 public function run()
 {
  while (!$this->taskQueue->isEmpty()) {
   // 任務(wù)出隊(duì)
   $task = $this->taskQueue->dequeue();
   $res = $task->run(); // 運(yùn)行任務(wù)直到 yield

   if (!$task->isFinished()) {
    $this->schedule($task); // 任務(wù)如果還沒(méi)完全執(zhí)行完畢，入隊(duì)等下次執(zhí)行
   }
  }
 }
}

?php
function task1() {
 for ($i = 1; $i = 10; ++$i) {
  echo "This is task 1 iteration $i.\n";
  yield; // 主動(dòng)讓出CPU的執(zhí)行權(quán)
 }
}
function task2() {
 for ($i = 1; $i = 5; ++$i) {
  echo "This is task 2 iteration $i.\n";
  yield; // 主動(dòng)讓出CPU的執(zhí)行權(quán)
 }
}
$scheduler = new Scheduler; // 實(shí)例化一個(gè)調(diào)度器
$scheduler->newTask(task1()); // 添加不同的閉包函數(shù)作為任務(wù)
$scheduler->newTask(task2());
$scheduler->run();

function task1() {
  /* 這里有一個(gè)遠(yuǎn)程任務(wù)，需要耗時(shí)10s，可能是一個(gè)遠(yuǎn)程機(jī)器抓取分析遠(yuǎn)程網(wǎng)址的任務(wù)，我們只要提交最后去遠(yuǎn)程機(jī)器拿結(jié)果就行了 */
  remote_task_commit();
  // 這時(shí)候請(qǐng)求發(fā)出后，我們不要在這里等，主動(dòng)讓出CPU的執(zhí)行權(quán)給task2運(yùn)行，他不依賴這個(gè)結(jié)果
  yield;
  yield (remote_task_receive());
  ...
} 
function task2() {
 for ($i = 1; $i = 5; ++$i) {
  echo "This is task 2 iteration $i.\n";
  yield; // 主動(dòng)讓出CPU的執(zhí)行權(quán)
 }
}

?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->newTask(task());
$scheduler->run();

 /**
  * Task constructor.
  * @param $taskId
  * @param Generator $coroutine
  */
 public function __construct($taskId, Generator $coroutine)
 {
  $this->taskId = $taskId;
  // $this->coroutine = $coroutine;
  // 換成這個(gè)，實(shí)際Task->run的就是stackedCoroutine這個(gè)函數(shù)，不是$coroutine保存的閉包函數(shù)了
  $this->coroutine = stackedCoroutine($coroutine); 
 }

/**
 * @param Generator $gen
 */
function stackedCoroutine(Generator $gen)
{
 $stack = new SplStack;
 // 不斷遍歷這個(gè)傳進(jìn)來(lái)的生成器
 for (; ;) {
  // $gen可以理解為指向當(dāng)前運(yùn)行的協(xié)程閉包函數(shù)（生成器）
  $value = $gen->current(); // 獲取中斷點(diǎn)，也就是yield出來(lái)的值
  if ($value instanceof Generator) {
   // 如果是也是一個(gè)生成器，這就是子協(xié)程了，把當(dāng)前運(yùn)行的協(xié)程入棧保存
   $stack->push($gen);
   $gen = $value; // 把子協(xié)程函數(shù)給gen，繼續(xù)執(zhí)行，注意接下來(lái)就是執(zhí)行子協(xié)程的流程了
   continue;
  }
  // 我們對(duì)子協(xié)程返回的結(jié)果做了封裝，下面講
  $isReturnValue = $value instanceof CoroutineReturnValue; // 子協(xié)程返回`$value`需要主協(xié)程幫忙處理 
  if (!$gen->valid() || $isReturnValue) {
   if ($stack->isEmpty()) {
    return;
   }
   // 如果是gen已經(jīng)執(zhí)行完畢，或者遇到子協(xié)程需要返回值給主協(xié)程去處理
   $gen = $stack->pop(); //出棧，得到之前入棧保存的主協(xié)程
   $gen->send($isReturnValue ? $value->getValue() : NULL); // 調(diào)用主協(xié)程處理子協(xié)程的輸出值
   continue;
  }
  $gen->send(yield $gen->key() => $value); // 繼續(xù)執(zhí)行子協(xié)程
 }
}

class CoroutineReturnValue {
 protected $value;
 
 public function __construct($value) {
  $this->value = $value;
 }
 // 獲取能把子協(xié)程的輸出值給主協(xié)程，作為主協(xié)程的send參數(shù)
 public function getValue() {
  return $this->value;
 }
}
function retval($value) {
 return new CoroutineReturnValue($value);
}

function echoTimes($msg, $max) {
 for ($i = 1; $i = $max; ++$i) {
  echo "$msg iteration $i\n";
  yield;
 }
 yield retval(""); // 增加這個(gè)作為結(jié)束標(biāo)示
}

function task1()
{
 yield echoTimes('bar', 5);
}

 public function __construct($taskId, Generator $coroutine)
 {
  $this->taskId = $taskId;
  $this->coroutine = $coroutine;
  // $this->coroutine = stackedCoroutine($coroutine); //不需要自己實(shí)現(xiàn)了，改回之前的
 }

function echoTimes($msg, $max) {
 for ($i = 1; $i = $max; ++$i) {
  echo "$msg iteration $i\n";
  yield;
 }
}

function task1()
{
 yield from echoTimes('bar', 5);
}