Laravel HTTP 客户端池捕获 ConnectException

Question

如何处理 Laravel 的并发请求失败？例如，我在一个池中有 10 个请求，其中一半返回 GuzzleHttp\Exception\ConnectException。我想捕获它们并使用不同的代理服务器重试。

<?php

$responses = Illuminate\Support\Facades\Http::pool(
    function (Illuminate\Http\Client\Pool $pool) use ($params) {
        $url = "https://example.com/api/endpoint?foo=%s";
        $return = [];
        $proxy = $this->proxyManager->get();
    
        foreach($params as $param) {
            // Number of retries
            for($i = 0; $i < 10; $i++) {
                try {
                    // This returns GuzzleHttp\Promise\Promise
                    $return[] = $pool->as($param)
                        ->withOptions(['proxy' => $proxy])
                        ->get(\sprintf($url, $param))
                        ->otherwise(function ($e) {
                            // This never happens
                            dd( $e );
                        });
                    
                    break;
                } catch(\Exception $e) {
                    $proxy = $this->proxyManager->get();
                    
                    continue;
                }
            }
        }
        
        return $return;
    }
);

    
// Some fail, some ok
dd( $responses );

Answer 1

必须使用 Guzzle 和自定义中间件来归档我想要的内容：

app/Http/Client/Middleware.php

<?php

namespace App\Http\Client;

final class Middleware
{
    public static function retry( callable $decider, callable $delay = null ): callable
    {
        return static function( callable $handler ) use ( $decider, $delay ): RetryMiddleware
        {
            return new RetryMiddleware( $decider, $handler, $delay );
        };
    }
}

app/Http/Client/RetryMiddleware.php

<?php

namespace App\Http\Client;

use GuzzleHttp\Promise as P;
use GuzzleHttp\Promise\PromiseInterface;
use Psr\Http\Message\RequestInterface;
use Psr\Http\Message\ResponseInterface;

class RetryMiddleware
{
    private $nextHandler;

    private $decider;

    private $delay;

    public function __construct(callable $decider, callable $nextHandler, callable $delay = null)
    {
        $this->decider = $decider;
        $this->nextHandler = $nextHandler;
        $this->delay = $delay ?: __CLASS__ . '::defaultDelay';
    }

    public static function defaultDelay(int $retries): int
    {
        return 0;
    }

    public function __invoke(RequestInterface $request, array $options): PromiseInterface
    {
        if (!isset($options['retries'])) {
            $options['retries'] = 0;
        }

        $fn = $this->nextHandler;
        return $fn($request, $options)
            ->then(
                $this->onFulfilled($request, $options),
                $this->onRejected($request, $options)
            );
    }

    private function onFulfilled(RequestInterface $request, array $options): callable
    {
        return function ($value) use ($request, $options) {
            if (!($this->decider)(
                $options['retries'],
                $request,
                $value,
                null
            )) {
                return $value;
            }
            return $this->doRetry($request, $options, $value);
        };
    }

    private function onRejected(RequestInterface $req, array $options): callable
    {
        return function ($reason) use ($req, $options) {
            if (!($this->decider)(
                $options['retries'],
                $req,
                null,
                $reason
            )) {
                return P\Create::rejectionFor($reason);
            }
            return $this->doRetry($req, $options);
        };
    }

    private function doRetry(RequestInterface $request, array $options, ResponseInterface $response = null): PromiseInterface
    {
        $options['delay'] = ($this->delay)(++$options['retries'], $response);

        // Callback?
        if( $options['on_retry'] )
        {
            \call_user_func_array( $options['on_retry'], [
                &$request,
                &$options,
                $response
            ] );
        }

        return $this($request, $options);
    }
}

然后像这样发出异步请求：

<?php

$url = "https://example.com/api/endpoint?foo=%s";

$data = [];

$handlerStack = \GuzzleHttp\HandlerStack::create( new \GuzzleHttp\Handler\CurlMultiHandler() );

$handlerStack->push( \App\Http\Client\Middleware::retry( function( $retries, $request, $response = null, $exception = null )
{
    // Limit the number of retries to 10
    if( $retries >= 10 )
    {
        return false;
    }
    
    // Retry connection exceptions
    if( $exception )
    {
        return true;
    }
    
    return false;
} ) );

$client = new \GuzzleHttp\Client( [
    'handler'   => $handlerStack,
    'on_retry'  => function( &$request, &$options )
    {
        $options['proxy'] = $this->proxyManager->get();
    }
] );

$requests = function() use ( $client, $url, $params )
{
    foreach( $params as $param )
    {
        $get = \sprintf( $url, $param );
        
        yield function() use ( $client, $get )
        {
            return $client->getAsync( $get, [
                'proxy' => $this->proxyManager->get()
            ] );
        };
    }
};

$pool = new \GuzzleHttp\Pool( $client, $requests(), [
    'concurrency' => 5, // maximum number of requests to send concurrently
    'fulfilled' => function( \GuzzleHttp\Psr7\Response $response, $index ) use ( &$data )
    {
        $data[] = json_decode( $response->getBody(), true );
    }, // this is delivered each successful response
    'rejected' => function( \Exception $e, $index )
    {
    }, // this is delivered each failed request
] );

$pool->promise()->wait();

如果你想通过重试发出同步请求，那么 Laravel 的 HttpClient 可以：

<?php

$url = "https://example.com/api/endpoint";

$response = \Illuminate\Support\Facades\Http::withOptions( [
        'proxy'     => $this->proxyManager->get(),
        'on_retry'  => function( &$request, &$options )
        {
            $options['proxy'] = $this->proxyManager->get();
        }
    ] )
    ->withMiddleware( \App\Http\Client\Middleware::retry( function( $retries, $request, $response = null, $exception = null )
    {
        // Limit the number of retries to 10
        if( $retries >= 10 )
        {
            return false;
        }
        
        // Retry connection exceptions
        if( $exception )
        {
            return true;
        }
        
        return false;
    } ) )
    ->get( $url );