在 PHP 中异步调用多个函数

Question

我正在使用 Symfony 3 和 ReactPHP 库来控制我的所有功能，我需要对同一函数执行多次调用（subFunction() 在代码中）异步。我有 2 个项目（project1 和 project2）：

项目 1 代码：

/**
* Loop an array of urls and call sub function.
**/
public function startFunction() {
  $finalResponse = [];
  $urls = ['www.google.es', 'www.github.com', 'www.bitbucket.org'];

  foreach ($urls as $url) {
    $res = $this->subFunction($url);    // subfunction call ( **IT MAY TAKE A LONG TIME !!** )
    $finalResponse[] = $res;
 }

 return $finalResponse;
}

/**
* Uses Factory loop to get the Promise returned by finalRequest function.
**/
private function subFunction($url) {
  $loop = \React\EventLoop\Factory::create();
  $classA = new Project2\ClassA();
  $finalResponse = null;

  // project 2 function call
  $classA->finalRequest($url)->then(function($response) use(   
     &$finalResponse
  ) {
     $finalResponse = $response;
  })

  return $finalResponse;
}

项目 2 代码：

classA {

/**
* Makes an React\HttpClient request (GET) to sent url and return his value inside a Promise.
**/
public function finalRequest($url) {
   $generalDeferred = new Deferred();
   $generalPromise = $generalDeferred->promise();

   // make React\HttpClient request
   $request = $client->request('GET', $url);
   $request->on('response', function ($response) use($generalDeferred) {
   $response->on('data', function ($response) {
        $generalDeferred->resolve($response);
      });
   });
  $request->end();

  return $generalPromise;
}

}

问题是在每次 subFunction($url) 调用时，程序都会停止，直到子函数获得响应，但我需要异步执行此操作，因为此子函数可能需要很多秒。所以我想同时启动所有 subFunction($url) 调用，并异步获取所有响应。

有可能解决这个问题吗？谢谢。

Answer 1

首先，您只能在一个应用程序中有一个循环运行ning。其次你必须使循环运行: https://reactphp.org/event-loop/

您应该创建应用程序，然后注册所有服务和事件，启动循环并将其运行宁作为服务器。

$loop = React\EventLoop\Factory::create();

$server = stream_socket_server('tcp://127.0.0.1:8080');
stream_set_blocking($server, 0);

$loop->addReadStream($server, function ($server) use ($loop) {
[...]
});

$loop->addPeriodicTimer(5, function () {
[...]
});

$loop->run(); <---- you will not execute anything behind this point.

为什么？ https://github.com/reactphp/event-loop/blob/master/src/ExtLibeventLoop.php#L196

public function run()
{
    $this->running = true;
    while ($this->running) { <------------------------------
        $this->futureTickQueue->tick();
        $flags = EVLOOP_ONCE;
        if (!$this->running || !$this->futureTickQueue->isEmpty()) {
            $flags |= EVLOOP_NONBLOCK;
        } elseif (!$this->streamEvents && !$this->timerEvents->count()) {
            break;
        }
        event_base_loop($this->eventBase, $flags);
    }
}

对于循环的使用，我建议使用 Guzzle Async：http://docs.guzzlephp.org/en/stable/faq.html

$finalResponse = [];
$promises = [];
$urls = ['www.google.es', 'www.github.com', 'www.bitbucket.org'];
foreach ($urls as $index => $url) {
    $promise = $client->requestAsync('GET', $url);
    $promise->then(function ($response) use ($index, &$finalResponse) {
        $finalResponse[$index]=$response;
    });
    $promises[$index]=$promise;
}
foreach ($promises as $promise) {
    $promise->wait();
}
return $finalResponse;

Answer 2

如果您的 subFunction 是阻塞函数调用，则有多种选择。

重写subFunction以使用非阻塞I/O

这可能有效，也可能无效，具体取决于函数的作用。如果函数主要是等待 I/O 比如等待 HTTP 响应，那么这可能是更简单的方法，但这完全取决于需要重写的代码的复杂程度。

ReactPHP 提供了一个 HTTP 客户端，它使用非阻塞 I/O 来执行 HTTP 请求并接收您已经在使用的响应。

使用多处理

如果您的 subFunction 是 CPU 绑定的或者您尝试使用的协议不存在非阻塞驱动程序，您可以使用多处理来执行该函数在另一个进程中而不阻塞主事件循环。

有一个用于 ReactPHP 的子进程库，但我目前不知道有一个 ReactPHP 库可以使并行处理变得像 amphp/parallel. While that library is based on Amp, there exists a compatibility adapter 在 Amp 上使用任何 ReactPHP 库一样容易，所以你可以继续使用您当前的库。

使用构建在 amphp/parallel 之上的 amphp/parallel-functions，您只需使用几行代码即可在另一个进程中执行您的函数。唯一的限制是您传递的数据（参数和 return 值）必须是可序列化的。

<?php

use Amp\Promise;
use function Amp\ParallelFunctions\parallel;

$callable = parallel('subFunction');

$promise = $callable($arg1, $arg2, $arg3);

// either yield $promise in a coroutine,
// or adapt it to a ReactPHP promise,
// or use Promise\wait for a blocking wait.

var_dump(Promise\wait($promise));

在 PHP 中异步调用多个函数

Multiple function calls asynchronously in PHP

php

symfony

asynchronous

parallel-processing

reactphp