channel演示

这是我们最终得到的接口：

  1. <?php
  3. function go(...$args)
  4. {
  5.     spawn(...$args);
  6. }
  8. function chan($n = 0)
  9. {
  10.     if ($n === 0) {
  11.         return new Channel();
  12.     } else {
  13.         return new BufferChannel($n);
  14.     }
  15. }

第一个典型例子, PINGPONG。

与golang的channel类似，我们可以在两个semicoroutine之间做同步：

  1. <?php
  3. // 构建两个单向channel, 我们只单向收发数据
  5. $pingCh = chan();
  6. $pongCh = chan();
  8. go(function() use($pingCh, $pongCh) {
  9.     while (true) {
  10.         echo (yield $pingCh->recv());
  11.         yield $pongCh->send("PONG\n");
  13.         // 递归调度器实现，需要引入异步的方法退栈，否则Stack Overflow...
  14.         // 或者考虑将send或者recv以defer方式实现
  15.         yield async_sleep(1);
  16.     }
  17. });
  19. go(function() use($pingCh, $pongCh) {
  20.     while (true) {
  21.         echo (yield $pongCh->recv());
  22.         yield $pingCh->send("PING\n");
  24.         yield async_sleep(1);
  25.     }
  26. });
  28. // start up
  29. go(function() use($pingCh) {
  30.     echo "start up\n";
  31.     yield $pingCh->send("PING");
  32. });
  34. // output:
  35. /*
  36. start up
  37. PING
  38. PONG
  39. PING
  40. PONG
  41. PING
  42. PONG
  43. PING
  44. ...
  45. */

当然，我们可以很轻易构建一个生产者-消费者模型：

  1. <?php
  3. $ch = chan();
  5. // 生产者1
  6. go(function() use($ch) {
  7.     while (true) {
  8.         yield $ch->send("producer 1");
  9.         yield async_sleep(1000);
  10.     }
  11. });
  13. // 生产者2
  14. go(function() use($ch) {
  15.     while (true) {
  16.         yield $ch->send("producer 2");
  17.         yield async_sleep(1000);
  18.     }
  19. });
  21. // 消费者1
  22. go(function() use($ch) {
  23.     while (true) {
  24.         $recv = (yield $ch->recv());
  25.         echo "consumer1: $recv\n";
  26.     }
  27. });
  29. // 消费者2
  30. go(function() use($ch) {
  31.     while (true) {
  32.         $recv = (yield $ch->recv());
  33.         echo "consumer2: $recv\n";
  34.     }
  35. });
  37. // output:
  38. /*
  39. consumer1 recv from producer 1
  40. consumer1 recv from producer 2
  41. consumer1 recv from producer 1
  42. consumer2 recv from producer 2
  43. consumer1 recv from producer 2
  44. consumer2 recv from producer 1
  45. consumer1 recv from producer 1
  46. consumer2 recv from producer 2
  47. consumer1 recv from producer 2
  48. consumer2 recv from producer 1
  49. consumer1 recv from producer 1
  50. consumer2 recv from producer 2
  51. ......
  52. */

channel 自身是first-class value, 所以可传递：

  1. <?php
  3. // 我们通过一个chan来发送另一个chan
  4. // 然后等待接收到这个chan的semicoroutine回送数据
  6. $ch = chan();
  8. go(function() use ($ch) {
  9.     $anotherCh = chan();
  10.     yield $ch->send($anotherCh);
  11.     echo "send another channel\n";
  12.     yield $anotherCh->send("HELLO");
  13.     echo "send hello through another channel\n";
  14. });
  16. go(function() use($ch) {
  17.     /** @var Channel $anotherCh */
  18.     $anotherCh = (yield $ch->recv());
  19.     echo "recv another channel\n";
  20.     $val = (yield $anotherCh->recv());
  21.     echo $val, "\n";
  22. });
  24. // output:
  25. /*
  26. send another channel
  27. recv another channel
  28. send hello through another channel
  29. HELLO
  30. */

我们通过控制channel缓存大小 观察输出结果：

  1. <?php
  2. $ch = chan($n);
  4. go(function() use($ch) {
  5.     $recv = (yield $ch->recv());
  6.     echo "recv $recv\n";
  7.     $recv = (yield $ch->recv());
  8.     echo "recv $recv\n";
  9.     $recv = (yield $ch->recv());
  10.     echo "recv $recv\n";
  11.     $recv = (yield $ch->recv());
  12.     echo "recv $recv\n";
  13. });
  15. go(function() use($ch) {
  16.     yield $ch->send(1);
  17.     echo "send 1\n";
  18.     yield $ch->send(2);
  19.     echo "send 2\n";
  20.     yield $ch->send(3);
  21.     echo "send 3\n";
  22.     yield $ch->send(4);
  23.     echo "send 4\n";
  24. });
  26. // $n = 1;
  27. // output:
  28. /*
  29. send 1
  30. recv 1
  31. send 2
  32. recv 2
  33. send 3
  34. recv 3
  35. send 4
  36. recv 4
  37. */
  39. // $n = 2;
  40. // output:
  41. /*
  42. send 1
  43. send 2
  44. recv 1
  45. recv 2
  46. send 3
  47. send 4
  48. recv 3
  49. recv 4
  50. */
  52. // $n = 3;
  53. // output:
  54. /*
  55. send 1
  56. send 2
  57. send 3
  58. recv 1
  59. recv 2
  60. recv 3
  61. send 4
  62. recv 4
  63. */

一个更具体的生产者消费者的例子：

  1. <?php
  3. // 缓存两个结果
  4. $ch = chan(2);
  6. // 从channel接口请求写过写文件
  7. go(function() use($ch) {
  8.     $file = new AsyncFile("path/to/save");
  9.     while (true) {
  10.         list($host, $status) = (yield $ch->recv());
  11.         yield $file->write("$host: $status\n");
  12.     }
  13. });
  15. // 请求并写入chan
  16. go(function() use($ch) {
  17.     while (true) {
  18.         $host = "www.baidu.com";
  19.         $resp = (yield async_curl_get($host));
  20.         yield $ch->send([$host, $resp->statusCode]);
  21.     }
  22. });
  24. // 请求并写入chan
  25. go(function() use($ch) {
  26.     while (true) {
  27.         $host = "www.bing.com";
  28.         $resp = (yield async_curl_get($host));
  29.         yield $ch->send([$host, $resp->statusCode]);
  30.     }
  31. });
  33. // output:

channel的发送与接受没有超时机制，Golang可以select多个chan实现超时处理，我们可以做一个select设施，或者在send于recv接受直接添加超时参数，扩展接口功能，留待读者自行实现。