如何使 Amphp pool/queue 的多个请求？ Curl 处理程序在哪里？

Question

有一个 example/test 使用 GuzzleHttp 制作的代码：

use GuzzleHttp\Client;
use GuzzleHttp\Handler\CurlHandler;
use GuzzleHttp\HandlerStack;
use GuzzleHttp\Middleware;
use GuzzleHttp\Pool;
use Psr\Http\Message\ResponseInterface;

require __DIR__ . '/vendor/autoload.php';

$handler = new CurlHandler();

$stack = new HandlerStack($handler);
$stack->push(Middleware::httpErrors(), 'http_errors');
$stack->push(Middleware::redirect(), 'allow_redirects');
$stack->push(Middleware::cookies(), 'cookies');
$stack->push(Middleware::prepareBody(), 'prepare_body');
$interval = 100;
$concurrency = 50;
$client = new Client(['handler' => $stack]);
echo sprintf("Using Guzzle handler %s\n", get_class($handler));
echo sprintf("Printing memory usage every %d requests\n", $interval);
echo "Fetching package list... ";

$packageNames = json_decode(
    $client->get('https://packagist.org/packages/list.json')
           ->getBody()
           ->getContents()
)->packageNames;

if (empty($packageNames)) {
    echo "Empty result. No reason to continue.";
    return;
}

echo 'done. (' . count($packageNames) . " packages)\n\n";

$requests = function($packageNames) {
    foreach ($packageNames as $packageVendorPair) {
        yield new GuzzleHttp\Psr7\Request('GET', "https://packagist.org/p/{$packageVendorPair}.json");
    }
};

$pool = new Pool($client, $requests($packageNames), [
    'concurrency' => $concurrency,
    'fulfilled' => function (ResponseInterface $response, $index) use (&$counter, $interval) {
        $counter++;
        if ($counter % $interval === 0) {
            echo sprintf(
                "Processed %s requests. Memory used: %s MB\n",
                number_format($counter),
                number_format(memory_get_peak_usage()/1024/1024, 3)
            );
        }
    },
    'rejected' => function($reason, $index) use (&$counter, $interval)
    {
        $counter++;
        if ($counter % $interval === 0) {
            echo sprintf(
                'Processed %s requests. Memory used: %s MB',
                number_format($counter),
                number_format(memory_get_peak_usage()/1024/1024, 3)
            );
        }
    }
]);

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

如何为 Amphp 或 Artax 制作类似的东西？我搜索了 amp 文档和 Whosebug，但找不到任何类似的内容。

顺便说一句，我还发现 Amp 不使用 Curl 作为处理程序。不明白为什么没有这样的选项可用。您可以手动添加它还是有更好的东西，什么可以替代 curl 功能（各种自定义 headers、debug/verbose 可能性等）？

我需要帮助的具体点：

是否有人可以告诉我在哪里可以找到使用 Amp 框架或任何它的库制作的池等效示例 and/or 即使在更简单的示例中也可以显示它？
Amp 中的 Curl 处理程序在哪里？我可以使用它吗？如何使用？

Amphp网站上说：

如果问题足够通用，Stack Overflow 社区可以回答您的问题。使用 amphp 标签，以便合适的人找到您的问题。

由于我提供了足够简单（且有效）的示例，我认为它很容易准确地理解我的需要。

恕我直言。

Answer 1

没有池等效项，但可以使用信号量和异步协程编写。

<?php

use Amp\Artax\DefaultClient;
use Amp\Loop;
use Amp\Sync\LocalSemaphore;

require __DIR__ . "/vendor/autoload.php";

Loop::run(function () {
    $concurrency = 10;
    $client = new DefaultClient;
    $semaphore = new LocalSemaphore(10);

    $packageResponse = yield $client->request("https://packagist.org/packages/list.json");
    $packageNames = json_decode(yield $packageResponse->getBody())->packageNames;

    $requestHandler = Amp\coroutine(function ($package) use ($client) {
        $url = "https://packagist.org/p/{$package}.json";

        $response = yield $client->request($url);
        $body = yield $response->getBody();

        return $body;
    });

    $counter = 0;

    foreach ($packageNames as $package) {
        $lock = yield $semaphore->acquire();

        $promise = $requestHandler($package);
        $promise->onResolve(function ($error, $body) use (&$counter, $lock) {
            $lock->release();

            if (++$counter % 50 === 0) {
                echo sprintf(
                    "Processed %s requests. Memory used: %s MB\n",
                    number_format($counter),
                    number_format(memory_get_peak_usage()/1024/1024, 3)
                );
            }
        });
    }
});

这个例子使用了LocalSemaphore implementation, which is an implementation of Amp\Sync\Semaphore。信号量用于限制并发。

Amp 中没有 Curl 处理程序，因为它不适用于事件循环。 Curl 有自己的事件循环，但只允许多个并发 HTTP 请求，没有其他非阻塞 I/O。这就是为什么 Artax 实现基于原始 PHP 套接字而不依赖于 Curl 的 HTTP。

如何使 Amphp pool/queue 的多个请求？ Curl 处理程序在哪里？

How to make Amphp pool/queue of multiple requests? And where is Curl handler?

php

guzzle

artax

amphp