PHP7下協(xié)程的實現(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;
 }
 /**
  * 獲取當前的Task的ID
  * 
  * @return mixed
  */
 public function getTaskId()
 {
  return $this->taskId;
 }
 /**
  * 判斷Task執(zhí)行完畢了沒有
  * 
  * @return bool
  */
 public function isFinished()
 {
  return !$this->coroutine->valid();
 }
 /**
  * 設(shè)置下次要傳給協(xié)程的值，比如 $id = (yield $xxxx)，這個值就給了$id了
  * 
  * @param $value
  */
 public function setSendValue($value)
 {
  $this->sendValue = $value;
 }
 /**
  * 運行任務(wù)
  * 
  * @return mixed
  */
 public function run()
 {
  // 這里要注意，生成器的開始會reset，所以第一個值要用current獲取
  if ($this->beforeFirstYield) {
   $this->beforeFirstYield = false;
   return $this->coroutine->current();
  } else {
   // 我們說過了，用send去調(diào)用一個生成器
   $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()
 {
  /* 原理就是維護了一個隊列，
   * 前面說過，從編程角度上看，協(xié)程的思想本質(zhì)上就是控制流的主動讓出（yield）和恢復（resume）機制
   * */
  $this->taskQueue = new SplQueue();
 }
 /**
  * 增加一個任務(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ù)進入隊列
  *
  * @param Task $task
  */
 public function schedule(Task $task)
 {
  $this->taskQueue->enqueue($task);
 }
 /**
  * 運行調(diào)度器
  */
 public function run()
 {
  while (!$this->taskQueue->isEmpty()) {
   // 任務(wù)出隊
   $task = $this->taskQueue->dequeue();
   $res = $task->run(); // 運行任務(wù)直到 yield

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

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

function task1() {
  /* 這里有一個遠程任務(wù)，需要耗時10s，可能是一個遠程機器抓取分析遠程網(wǎng)址的任務(wù)，我們只要提交最后去遠程機器拿結(jié)果就行了 */
  remote_task_commit();
  // 這時候請求發(fā)出后，我們不要在這里等，主動讓出CPU的執(zhí)行權(quán)給task2運行，他不依賴這個結(jié)果
  yield;
  yield (remote_task_receive());
  ...
} 
function task2() {
 for ($i = 1; $i = 5; ++$i) {
  echo "This is task 2 iteration $i.\n";
  yield; // 主動讓出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;
  // 換成這個，實際Task->run的就是stackedCoroutine這個函數(shù)，不是$coroutine保存的閉包函數(shù)了
  $this->coroutine = stackedCoroutine($coroutine); 
 }

/**
 * @param Generator $gen
 */
function stackedCoroutine(Generator $gen)
{
 $stack = new SplStack;
 // 不斷遍歷這個傳進來的生成器
 for (; ;) {
  // $gen可以理解為指向當前運行的協(xié)程閉包函數(shù)（生成器）
  $value = $gen->current(); // 獲取中斷點，也就是yield出來的值
  if ($value instanceof Generator) {
   // 如果是也是一個生成器，這就是子協(xié)程了，把當前運行的協(xié)程入棧保存
   $stack->push($gen);
   $gen = $value; // 把子協(xié)程函數(shù)給gen，繼續(xù)執(zhí)行，注意接下來就是執(zhí)行子協(xié)程的流程了
   continue;
  }
  // 我們對子協(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(""); // 增加這個作為結(jié)束標示
}

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

 public function __construct($taskId, Generator $coroutine)
 {
  $this->taskId = $taskId;
  $this->coroutine = $coroutine;
  // $this->coroutine = stackedCoroutine($coroutine); //不需要自己實現(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);
}