协程工作原理

协程  

协程(Coroutine)在执行过程中可中断去执行其他任务,执行完毕后再回来继续原先的操作。可以理解为两个或多个程序协同工作。

协程特点在于单线程执行。
优势一:具有极高的执行效率,因为在任务切换的时候是程序之间的切换(由程序自身控制)而不是线程间的切换,所以没有线程切换导致的额外开销(时间浪费),线程越多,协程性能优势越明显。

优势二:由于是单线程工作,没有多线程需要考虑的同时写变量冲突,所以不需要多线程的锁机制,故执行效率比多线程更高。

常利用多进程(利用多核)+协程来获取更高的性能。

迭代生成器

讲工作原理前先了解下迭代生成器,迭代生成器也是一个函数,不同的是这个函数的返回值是依次返回,而不是只返回一个单独的值。或者,换句话说,生成器使你能更方便的实现了迭代器接口。下面通过实现一个xrange函数来简单说明:

  1. <?php
  2. function xrange($start, $end, $step = 1) {
  3. for ($i = $start; $i <= $end; $i += $step) {
  4. yield $i;
  5. }
  6. }
  7. foreach (xrange(1, 1000000) as $num) {
  8. echo $num, "\n";
  9. }

上面这个xrange()函数提供了和PHP的内建函数range()一样的功能.但是不同的是range()函数返回的是一个包含值从1到100万的数组(注:请查看手册)。而xrange()函数返回的是依次输出这些值的一个迭代器,而不会真正以数组形式返回。

这种方法的优点是显而易见的,它可以让你在处理大数据集合的时候不用一次性的加载到内存中。甚至你可以处理无限大的数据流。

当然,也可以不同通过生成器来实现这个功能,而是可以通过继承Iterator接口实现。但通过使用生成器实现起来会更方便,不用再去实现iterator接口中的5个方法了。

生成器为可中断的函数

要从生成器认识协程, 理解它内部是如何工作是非常重要的: 生成器是一种可中断的函数, 在它里面的yield构成了中断点.

还是看上面的例子, 调用xrange(1,1000000)的时候, xrange()函数里代码其实并没有真正地运行. 它只是返回了一个迭代器:

  1. <?php
  2. $range = xrange(1, 1000000);
  3. var_dump($range); // object(Generator)#1
  4. var_dump($range instanceof Iterator); // bool(true)
  5. ?>

这也解释了为什么xrange叫做迭代生成器,因为它返回一个迭代器,而这个迭代器实现了Iterator接口。

调用迭代器的方法一次,其中的代码运行一次。例如,如果你调用$range->rewind(),那么xrange()里的代码就会运行到控制流第一次出现yield的地方。而函数内传递给yield语句的返回值可以通过$range->current()获取。

为了继续执行生成器中yield后的代码,你就需要调用$range->next()方法。这将再次启动生成器,直到下一次yield语句出现。因此,连续调用next()和current()方法,你就能从生成器里获得所有的值,直到再没有yield语句出现。

对xrange()来说,这种情形出现在$i超过$end时。在这中情况下,控制流将到达函数的终点,因此将不执行任何代码。一旦这种情况发生,vaild()方法将返回假,这时迭代结束。

再说协程

协程的支持是在迭代生成器的基础上,增加了可以回送数据给生成器的功能(调用者发送数据给被调用的生成器函数)。 这就把生成器到调用者的单向通信转变为两者之间的双向通信。

传递数据的功能是通过迭代器的send()方法实现的。下面的logger()协程是这种通信如何运行的例子:

  1. <?php
  2. function logger($fileName) {
  3. $fileHandle = fopen($fileName, 'a');
  4. while (true) {
  5. fwrite($fileHandle, yield . "\n");
  6. }
  7. }
  8. $logger = logger(__DIR__ . '/log');
  9. $logger->send('Foo');
  10. $logger->send('Bar')
  11. ?>

正如你能看到,这儿yield没有作为一个语句来使用,而是用作一个表达式,即它能被演化成一个值。这个值就是调用者传递给send()方法的值。在这个例子里,yield表达式将首先被Foo替代写入Log,然后被Bar替代写入Log。

上面的例子里演示了yield作为接受者,接下来我们看如何同时进行接收和发送的例子:

  1. <?php
  2. function gen() {
  3. $ret = (yield 'yield1');
  4. var_dump($ret);
  5. $ret = (yield 'yield2');
  6. var_dump($ret);
  7. }
  8. $gen = gen();
  9. var_dump($gen->current()); // string(6) "yield1"
  10. var_dump($gen->send('ret1')); // string(4) "ret1" (the first var_dump in gen)
  11. // string(6) "yield2" (the var_dump of the ->send() return value)
  12. var_dump($gen->send('ret2')); // string(4) "ret2" (again from within gen)
  13. // NULL (the return value of ->send())
  14. ?>

要很快的理解输出的精确顺序可能稍微有点困难,但你确定要搞清楚为什按照这种方式输出。以便后续继续阅读。

另外,我要特别指出的有两点:

第一点,yield表达式两边的括号在PHP7以前不是可选的,也就是说在PHP5.5和PHP5.6中圆括号是必须的。

第二点,你可能已经注意到调用current()之前没有调用rewind()。这是因为生成迭代对象的时候已经隐含地执行了rewind操作。

多任务协作

如果阅读了上面的logger()例子,你也许会疑惑“为了双向通信我为什么要使用协程呢?我完全可以使用其他非协程方法实现同样的功能啊?”,是的,你是对的,但上面的例子只是为了演示了基本用法,这个例子其实并没有真正的展示出使用协程的优点。

正如上面介绍里提到的,协程是非常强大的概念,不过却应用的很稀少而且常常十分复杂。要给出一些简单而真实的例子很难。

在这篇文章里,我决定去做的是使用协程实现多任务协作。我们要解决的问题是你想并发地运行多任务(或者 程序 ),不过我们都知道CPU在一个时刻只能运行一个任务(不考虑多核的情况)。因此处理器需要在不同的任务之间进行切换,而且总是让每个任务运行 一小会儿

多任务协作这个术语中的 协作 很好的说明了如何进行这种切换的:它要求当前正在运行的任务自动把控制传回给调度器,这样就可以运行其他任务了。这与 抢占 多任务相反,抢占多任务是这样的:调度器可以中断运行了一段时间的任务,不管它喜欢还是不喜欢。协作多任务在Windows的早期版本(windows95)和Mac OS中有使用,不过它们后来都切换到使用抢先多任务了。理由相当明确:如果你依靠程序自动交出控制的话,那么一些恶意的程序将很容易占用整个CPU,不与其他任务共享。

现在你应当明白协程和任务调度之间的关系:yield指令提供了任务中断自身的一种方法,然后把控制交回给任务调度器。因此协程可以运行多个其他任务。更进一步来说,yield还可以用来在任务和调度器之间进行通信。

为了实现我们的多任务调度,首先实现 任务 , 一个用轻量级的包装的协程函数:

  1. <?php
  2. class Task {
  3. protected $taskId;
  4. protected $coroutine;
  5. protected $sendValue = null;
  6. protected $beforeFirstYield = true;
  7. public function __construct($taskId, Generator $coroutine) {
  8. $this->taskId = $taskId;
  9. $this->coroutine = $coroutine;
  10. }
  11. public function getTaskId() {
  12. return $this->taskId;
  13. }
  14. public function setSendValue($sendValue) {
  15. $this->sendValue = $sendValue;
  16. }
  17. public function run() {
  18. if ($this->beforeFirstYield) {
  19. $this->beforeFirstYield = false;
  20. return $this->coroutine->current();
  21. } else {
  22. $retval = $this->coroutine->send($this->sendValue);
  23. $this->sendValue = null;
  24. return $retval;
  25. }
  26. }
  27. public function isFinished() {
  28. return !$this->coroutine->valid();
  29. }
  30. }

如代码,一个任务就是用任务ID标记的一个协程(函数)。使用 setSendValue() 方法,你可以指定哪些值将被发送到下次的恢复(在之后你会了解到我们需要这个),run() 函数确实没有做什么,除了调用 send() 方法的协同程序,要理解为什么添加了一个 beforeFirstYieldflag 变量,需要考虑下面的代码片段:

  1. <?php
  2. function gen() {
  3. yield 'foo';
  4. yield 'bar';
  5. }
  6. $gen = gen();
  7. var_dump($gen->send('something'));
  8. // 如之前提到的在send之前,当$gen迭代器被创建的时候一个renwind()方法已经被隐式调用
  9. // 所以实际上发生的应该类似:
  10. //$gen->rewind();
  11. //var_dump($gen->send('something'));
  12. //这样renwind的执行将会导致第一个yield被执行, 并且忽略了他的返回值.
  13. //真正当我们调用yield的时候, 我们得到的是第二个yield的值! 导致第一个yield的值被忽略.
  14. //string(3) "bar"

通过添加 beforeFirstYieldcondition 我们可以确定第一个yield的值能被正确返回。

调度器现在不得不比多任务循环要做稍微多点了,然后才运行多任务:

  1. <?php
  2. class Scheduler {
  3. protected $maxTaskId = 0;
  4. protected $taskMap = []; // taskId => task
  5. protected $taskQueue;
  6. public function __construct() {
  7. $this->taskQueue = new SplQueue();
  8. }
  9. public function newTask(Generator $coroutine) {
  10. $tid = ++$this->maxTaskId;
  11. $task = new Task($tid, $coroutine);
  12. $this->taskMap[$tid] = $task;
  13. $this->schedule($task);
  14. return $tid;
  15. }
  16. public function schedule(Task $task) {
  17. $this->taskQueue->enqueue($task);
  18. }
  19. public function run() {
  20. while (!$this->taskQueue->isEmpty()) {
  21. $task = $this->taskQueue->dequeue();
  22. $task->run();
  23. if ($task->isFinished()) {
  24. unset($this->taskMap[$task->getTaskId()]);
  25. } else {
  26. $this->schedule($task);
  27. }
  28. }
  29. }
  30. }
  31. ?>

newTask()方法(使用下一个空闲的任务id)创建一个新任务,然后把这个任务放入任务map数组里。接着它通过把任务放入任务队列里来实现对任务的调度。接着run()方法扫描任务队列,运行任务。如果一个任务结束了,那么它将从队列里删除,否则它将在队列的末尾再次被调度。

让我们看看下面具有两个简单(没有什么意义)任务的调度器:

  1. <?php
  2. function task1() {
  3. for ($i = 1; $i <= 10; ++$i) {
  4. echo "This is task 1 iteration $i.\n";
  5. yield;
  6. }
  7. }
  8. function task2() {
  9. for ($i = 1; $i <= 5; ++$i) {
  10. echo "This is task 2 iteration $i.\n";
  11. yield;
  12. }
  13. }
  14. $scheduler = new Scheduler;
  15. $scheduler->newTask(task1());
  16. $scheduler->newTask(task2());
  17. $scheduler->run();

两个任务都仅仅回显一条信息,然后使用yield把控制回传给调度器。输出结果如下:

  1. This is task 1 iteration 1.
  2. This is task 2 iteration 1.
  3. This is task 1 iteration 2.
  4. This is task 2 iteration 2.
  5. This is task 1 iteration 3.
  6. This is task 2 iteration 3.
  7. This is task 1 iteration 4.
  8. This is task 2 iteration 4.
  9. This is task 1 iteration 5.
  10. This is task 2 iteration 5.
  11. This is task 1 iteration 6.
  12. This is task 1 iteration 7.
  13. This is task 1 iteration 8.
  14. This is task 1 iteration 9.
  15. This is task 1 iteration 10.

输出确实如我们所期望的:对前五个迭代来说,两个任务是交替运行的,而在第二个任务结束后,只有第一个任务继续运行。



评论 0

发表评论

Top