Guzzleで並列処理（リトライ機能有）

公式ドキュメントに例もありますが、promiseの処理等部分的に散らばっているのでとりあえずPool（不特定のリクエスト数を処理する場合に使用）を使わない方法をまとめたメモ

リトライ処理を入れていなかったのでこれで対応してみましたが、並列数をあげてもパフォーマンスが変わらなかったので調べたところ、同期処理になってしまいました。

requestAsyncはこのリトライ処理といっしょに使うことはできないようです。
requestAsyncのpiromiseを返すところで都度アクセスが発生していました。

composer require guzzlehttp/guzzle:~6.0

コード

use GuzzleHttp\Client;
use GuzzleHttp\Handler\CurlHandler;
use Psr\Http\Message\ResponseInterface;
use GuzzleHttp\Exception\RequestException;
use GuzzleHttp\Exception\ConnectException;
use GuzzleHttp\Promise;
use GuzzleHttp\Psr7\Request;
use GuzzleHttp\Psr7\Response;
use GuzzleHttp\HandlerStack;

class Api {

    const HTTP_STATUS_CODE_OK = 200;

    private $client;

    public function __construct(HandlerStack $mock_handler = null)
    {
        $argument = [];
        $argument['timeout'] = 3.0;
        $argument['headers'] = [
            'User-Agent' => ''
        ];
        if (isset($mock_handler))
        {
            $argument['handler'] = $mock_handler;
        }

        $this->client = new Client($argument);
    }


    /**
     * API並列実行
     *
     * @return array
     */
    public function execute_multiple_api(): array
    {
        $main_api = 'https://example.com/main';
        $sub_api = 'https://example.com/sub';

        $body = [];
        $body['test'] = [];

        $json_body = json_encode($body);

        $promises = [];

        $promises['main'] = $this->execute_api($main_api, $json_body);
        $promises['sub'] = $this->execute_api($sub_api, $json_body);

        $results = Promise\settle($promises)->wait();

        // エラーの場合リトライ処理を行う
        foreach ($results as &$result)
        {
            if ( ! isset($result['value']))
            {
                $result['value'] = $this->execute_retry($api, $json_body);
            }
        }
        // response body
        // var_dump($results['main']['value']->getBody()->getContents());
        // var_dump($results['sub']['value']->getBody()->getContents());
        // 200以外でもbodyは取得できるので成否判定は$results['main']['value']->getStatusCode()で判定する
        // (RequestExceptionの場合は$results['main']['value']が存在しなかったので前提としてチェックは必要)
        // $promise->then()内で$responseに独自で追加したapiプロパティには$results['main']['value']->apiでアクセスできます。

        return $results;
    }

    /**
     * API実行
     *
     * @param string $api
     * @param string $json_body
     * @return Promise\Promise
     */
    private function execute_api(string $api, string $json_body): Promise\Promise
    {
        $promise = $this->client->requestAsync('POST', $api, [
            'http_errors' => false,// 4xxおよび5xx応答で例外を出さないようにする
            'headers' => ['content-type' => 'application/json', 'Accept' => 'application/json'],
            'body' => $json_body,
            //'debug' => true // debug
        ]);
        $promise->then(
            function (ResponseInterface $response) use ($json_body, $api) {
                $response->api = $api;// responseにapiを保持してみる
                if ($response->getStatusCode() !== self::HTTP_STATUS_CODE_OK)
                {
                    // エラーログ処理
                };
            },
            function (RequestException $exception) use ($json_body) {
                // エラーログ処理
            }
        );

        return $promise;
    }

    private function execute_retry(string $api, string $json_body)
    {
        $retry_max = 2;
        foreach (range(1, $retry_max) as $retry_count)
        {
            $promises = [];
            $promises[] = $this->execute_api($api, $json_body);
            $results = Promise\settle($promises)->wait();
            if (isset($results[0]['value']))
            {
                return $results[0]['value'];
            }
        }

        return null;
    }
}

テスト用モック例

$mock = new MockHandler([
    new RequestException('RequestException', new Request('GET', 'test')),// 1回目の呼び出し
    new RequestException('RequestException', new Request('GET', 'test')),// 2回目の呼び出し
    new Response(200),// 3回目の呼び出し
]);

$mock_handler= HandlerStack::create($mock);
$api = new Api($mock_handler);
$api->execute_multiple_api();

参考

http://docs.guzzlephp.org/en/stable/quickstart.html#concurrent-requests
http://docs.guzzlephp.org/en/stable/quickstart.html#using-responses
http://docs.guzzlephp.org/en/stable/request-options.html?highlight=%27http_errors%27%20%3D%3E%20false#http-errors
http://addshore.com/2015/12/guzzle-6-retry-middleware
http://docs.guzzlephp.org/en/stable/testing.html#mock-handler