php实现Promise.all和Promise.race

  • 测试

    $promise1 = function () {
      msleep(500);
      return 'one';
    };
    $promise2 = function () {
      msleep(100);
      return 'two';
    };
    $promise3 = function () {
      msleep(50);
      throw new \Exception('Reject');
    };
    var_dump(promise_all([$promise1, $promise2]));
    var_dump(promise_race([$promise1, $promise2]));
    var_dump(promise_race([$promise1, $promise2, $promise3]));
  • 结果

    # php promise.php
    array(2) {
    [1]=>
    string(3) "two"
    [0]=>
    string(3) "one"
    }
    string(3) "two"
    object(Exception)#15 (7) {}
  • 实现

    <?php
    
    declare(strict_types=1);
    
    use Swow\Coroutine;
    use Swow\Channel;
    use Swow\Sync\WaitGroup;
    use function Swow\defer;
    
    /**
    * @param array $callbacks 
    * @param int $parallel 并发数量
    * @return array 
    */
    function promise_all(array $callbacks, int $parallel = -1)
    {
      $wg = new WaitGroup();
      $channel = new Channel($parallel);
      $results = [];
      foreach ($callbacks as $key => $callback) {
          $wg->add();
          $channel->push(true);
          Coroutine::run(static function () use ($wg, $channel, $callback, $key, &$results) {
              try {
                  $results[$key] = $callback();
              } catch (\Throwable) {
              } finally {
                  $channel->pop();
                  $wg->done();
              }
          });
      }
      $wg->wait();
      return $results;
    }
    
    /**
    * @param array $callbacks
    * @param int $timeout 超时
    * @param bool $throw 是否抛出异常
    * @return mixed
    */
    function promise_race(array $callbacks, int $timeout = -1, bool $throw = true): mixed
    {
      $coroutines = [];
      defer(static function () use (&$coroutines) {
          foreach ($coroutines as $coroutine) {
              if ($coroutine->isExecuting()) {
                  $coroutine->kill();
              }
          }
      });
      $channel = new Channel();
      foreach ($callbacks as $callback) {
          $coroutines[] = Coroutine::run(static function () use ($channel, $callback) {
              try {
                  $channel->push($callback());
              } catch (\Throwable $e) {
                  $channel->push($e);
              }
          });
      }
      try {
          return $channel->pop($timeout);
      } catch (\Throwable $e) {
          if ($throw) {
              throw $e;
          }
      }
      return false;
    }
本作品采用《CC 协议》,转载必须注明作者和本文链接
讨论数量: 0
(= ̄ω ̄=)··· 暂无内容!

讨论应以学习和精进为目的。请勿发布不友善或者负能量的内容,与人为善,比聪明更重要!