使用 Amp 并发处理循环中的 Promise-php教程-PHP中文网

使用 Amp 并发处理循环中的 Promise

心靈之曲

发布： 2025-09-04 18:34:30

原创

310人浏览过

使用 amp 并发处理循环中的 promise

正如摘要所说，本文旨在解决在使用 Amp 进行异步编程时，在循环中处理 Promise 时遇到的阻塞问题。在异步编程中，我们经常需要在循环中发起多个异步操作，并等待它们全部完成。然而，如果直接在循环中使用 yield 等待 Promise，会导致循环阻塞，无法充分利用异步的优势。

下面我们通过一个示例来说明这个问题以及如何解决它。假设我们需要从多个 URL 下载数据，并将下载过程分解为多个 chunk，然后并发地下载这些 chunk。

<?php

use Amp\Loop;
use Amp\Promise;
use Amp\Deferred;

/**
 * 模拟下载类
 */
class Downloader
{
    private string $path = '/tmp';
    private int $chunkCount = 10;
    private int $chunkSize = 1024;
    private int $connections = 3;
    private int $size = 10240; // 总大小

    public function __construct(private $app)
    {
    }

    /**
     * 开始下载
     *
     * @return void
     */
    public function download(): void
    {
        $chunks = [];
        for ($i = 0; $i < $this->chunkCount + 20; $i++) {
            $start = $i * $this->chunkSize;
            $end = ($i + 1) * $this->chunkSize;

            if ($i == $this->chunkCount - 1) {
                $end = $this->size;
            }

            $chunks[] = (object) ['id' => ($i + 1), 'start' => $start, 'end' => $end, 'path' => $this->path . "/" . $i];
        }

        $chunkedChunks = array_chunk($chunks, $this->connections);

        foreach ($chunkedChunks as $key => $chunkedChunk) {
            // 将整个 foreach 块封装在 Amp\call 中
            \Amp\Loop::run(function () use ($chunkedChunk) {
                $urls = [
                    'https://secure.php.net',
                    'https://amphp.org',
                    'https://github.com',
                ];

                $promises = [];
                foreach ($urls as $url) {
                    $promises[$url] = \Amp\call(function () use ($url) {
                        $deferred = new \Amp\Deferred();

                        \Amp\Loop::delay(3 * 1000, function () use ($url, $deferred) {
                            $deferred->resolve($url);
                        });

                        return $deferred->promise();
                    });
                }

                $responses = yield \Amp\Promise\all($promises);

                foreach ($responses as $url => $response) {
                    \printf("Read %d bytes from %s\n", \strlen($response), $url);
                }
            });
        }
    }
}

// 示例使用
Loop::run(function () {
    $downloader = new Downloader(null);
    $downloader->download();
});

登录后复制

代码解释：

Downloader 类: 模拟一个下载器，将下载任务分解为多个 chunk。
download() 方法:
- 将下载任务分解为多个 chunk，并存储在 $chunks 数组中。
- 使用 array_chunk 将 $chunks 数组分割成多个小数组，每个小数组包含 $this-youjiankuohaophpcnconnections 个 chunk。
- 关键部分: 使用 foreach 循环遍历 $chunkedChunks 数组。 为了实现并发，将整个 foreach 循环体封装在 Amp\call 中。 Amp\call 创建一个协程，允许循环中的异步操作并发执行。
- 在循环内部，为每个 URL 创建一个 Promise，并将其存储在 $promises 数组中。
- 使用 Amp\Promise\all($promises) 等待所有 Promise 完成。
- 遍历 $responses 数组，处理每个 URL 的响应。
Loop::run(): 启动 Amp 事件循环，并执行下载任务。

关键点：将 foreach 循环体封装在 Amp\call 中。

如果没有将 foreach 循环体封装在 Amp\call 中，yield Amp\Promise\all($promises) 将会阻塞整个循环，导致每次只能处理一批 Promise，无法实现真正的并发。通过使用 Amp\call，我们创建了一个协程，允许循环中的异步操作并发执行，从而提高了程序的效率。