PHP使用Guzzle发起的异步请求示例详解

目录

• Guzzle中的异步请求
• 使用Guzzle发起异步请求
• PHP发起HTTP请求的几种方式
• curl
• stream
• 什么是URI？URI的组成
• 请求的组装
• 请求的发起
• 同步调用与异步调用
• wait的实现
• waitFn是什么
• queue() 是的入队时机

$client = new Client();
$request = new Request('GET', 'http://www.baidu.com');
$promise = $client->sendAsync($request)->then(function ($response) {
    echo $response->getBody();
});
// todo something
echo 1;
$promise->wait();

function choose_handler()
{
    $handler = null;
    if (function_exists('curl_multi_exec') && function_exists('curl_exec')) {
        $handler = Proxy::wrapSync(new CurlMultiHandler(), new CurlHandler());
    } elseif (function_exists('curl_exec')) {
        $handler = new CurlHandler();
    } elseif (function_exists('curl_multi_exec')) {
        $handler = new CurlMultiHandler();
    }
    if (ini_get('allow_url_fopen')) {
        $handler = $handler
            ? Proxy::wrapStreaming($handler, new StreamHandler())
            : new StreamHandler();
    } elseif (!$handler) {
        throw new \RuntimeException('GuzzleHttp requires cURL, the '
            . 'allow_url_fopen ini setting, or a custom HTTP handler.');
    }
    return $handler;
}

public static function wrapStreaming(
        callable $default,
        callable $streaming
    ) {
        return function (RequestInterface $request, array $options) use ($default, $streaming) {
            return empty($options['stream'])
                ? $default($request, $options)
                : $streaming($request, $options);
        };
    }

// 根据系统当前状态，选择一个发起Http请求的协议的方法句柄
function choose_handler()
{
    $handler = null;
    if (function_exists('curl_multi_exec') && function_exists('curl_exec')) {
        $handler = Proxy::wrapSync(new CurlMultiHandler(), new CurlHandler());
    } elseif (function_exists('curl_exec')) {
        $handler = new CurlHandler();
    } elseif (function_exists('curl_multi_exec')) {
        $handler = new CurlMultiHandler();
    }
    if (ini_get('allow_url_fopen')) {
        $handler = $handler
            ? Proxy::wrapStreaming($handler, new StreamHandler())
            : new StreamHandler();
    } elseif (!$handler) {
        throw new \RuntimeException('GuzzleHttp requires cURL, the '
            . 'allow_url_fopen ini setting, or a custom HTTP handler.');
    }
    return $handler;
}

public function send(RequestInterface $request, array $options = [])
    {
        $options[RequestOptions::SYNCHRONOUS] = true;
        return $this->sendAsync($request, $options)->wait();
    }

public function wait($unwrap = true)
    {
        $this->waitIfPending();
        $inner = $this->result instanceof PromiseInterface
            ? $this->result->wait($unwrap)
            : $this->result;
        if ($unwrap) {
            if ($this->result instanceof PromiseInterface
                || $this->state === self::FULFILLED
            ) {
                return $inner;
            } else {
                // It's rejected so "unwrap" and throw an exception.
                throw exception_for($inner);
            }
        }
    }

private function waitIfPending()
    {
        if ($this->state !== self::PENDING) {
            return;
        } elseif ($this->waitFn) {
            $this->invokeWaitFn();
        } elseif ($this->waitList) {
            $this->invokeWaitList();
        } else {
            // If there's not wait function, then reject the promise.
            $this->reject('Cannot wait on a promise that has '
                . 'no internal wait function. You must provide a wait '
                . 'function when constructing the promise to be able to '
                . 'wait on a promise.');
        }
        queue()->run();
        if ($this->state === self::PENDING) {
            $this->reject('Invoking the wait callback did not resolve the promise');
        }
    }
    public function run()
    {
        /** @var callable $task */
        while ($task = array_shift($this->queue)) {
            $task();
        }
    }

$handler = $options['handler'];
// 返回一个promise类，这个类有一个属性是waitFn
return Promise\promise_for($handler($request, $options));

// curl的实现
                $promise = new Promise(
            [$this, 'execute'],
            function () use ($id) {
                return $this->cancel($id);
            }
        );

$queue->add(static function () use ($p, $value, $onFulfilled) {
            if ($p->getState() === self::PENDING) {
                try {
                    $p->resolve($onFulfilled($value));
                } catch (\Throwable $e) {
                    $p->reject($e);
                } catch (\Exception $e) {
                    $p->reject($e);
                }
            }
        });