Async экосистема PHP

Проблема: блокирующий I/O

В традиционном PHP каждая операция ввода-вывода блокирует процесс. Пока ответ от базы данных, HTTP-сервиса или файловой системы не получен, PHP просто ждёт.

Синхронный PHP (типичный):
────── fetch(api1) ██████████░░░░░░░░░░░ fetch(api2) ██████████░░░░░░░░░░░ render
       200ms waiting                      200ms waiting                     = 400ms total

Асинхронный PHP:
────── fetch(api1) ██████████
────── fetch(api2) ██████████
                              ░ render
                              = 200ms total (оба запроса параллельно)

Ключевой принцип: Async PHP не делает код быстрее вычислительно. Он устраняет время ожидания I/O, выполняя несколько операций ввода-вывода одновременно в одном потоке.

Event Loop — сердце async

Event Loop (цикл событий) — бесконечный цикл, который отслеживает готовность I/O-операций (сокеты, таймеры, сигналы) и вызывает соответствующие callback-и или возобновляет Fibers.

┌─────────────────────────────────┐
│          Event Loop             │
│                                 │
│  1. Проверить таймеры           │
│  2. Проверить I/O (poll/epoll)  │
│  3. Выполнить готовые callback  │
│  4. Повторить                   │
│                                 │
│  ┌──────┐ ┌──────┐ ┌──────┐    │
│  │Timer │ │Socket│ │Signal│    │
│  │Queue │ │Watch │ │Watch │    │
│  └──────┘ └──────┘ └──────┘    │
└─────────────────────────────────┘

Revolt — стандартный Event Loop

Revolt — это стандартный event loop для PHP, созданный разработчиками AMPHP. Все библиотеки AMPHP v3 используют Revolt. Это единый event loop, который работает с Fibers.

Установка

composer require revolt/event-loop

Базовые операции

<?php
declare(strict_types=1);

use Revolt\EventLoop;

// defer() — выполнить на следующей итерации цикла
EventLoop::defer(function (): void {
    echo "Deferred callback executed\n";
});

// delay() — выполнить через N секунд
EventLoop::delay(1.5, function (): void {
    echo "Executed after 1.5 seconds\n";
});

// repeat() — выполнять каждые N секунд
$callbackId = EventLoop::repeat(0.5, function (string $callbackId): void {
    static $count = 0;
    echo "Tick " . (++$count) . "\n";

    if ($count >= 5) {
        EventLoop::cancel($callbackId);  // Stop repeating
    }
});

// Run the event loop
EventLoop::run();

I/O наблюдатели (watchers)

<?php
declare(strict_types=1);

use Revolt\EventLoop;

// onReadable() — вызвать callback когда поток готов к чтению
$socket = stream_socket_client('tcp://example.com:80');
stream_set_blocking($socket, false);

// Send HTTP request
fwrite($socket, "GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n");

EventLoop::onReadable($socket, function (string $callbackId, $stream): void {
    $data = fread($stream, 8192);

    if ($data === '' || $data === false || feof($stream)) {
        EventLoop::cancel($callbackId);
        fclose($stream);
        echo "Connection closed. Received data.\n";
        return;
    }

    echo "Received " . strlen($data) . " bytes\n";
});

// onWritable() — вызвать когда поток готов к записи
// Useful for non-blocking writes to sockets
$server = stream_socket_server('tcp://127.0.0.1:8080');
stream_set_blocking($server, false);

EventLoop::onReadable($server, function (string $callbackId, $server): void {
    $client = @stream_socket_accept($server, 0);
    if ($client === false) {
        return;
    }

    stream_set_blocking($client, false);

    EventLoop::onReadable($client, function (string $cbId, $client): void {
        $data = fread($client, 8192);
        if ($data === '' || $data === false) {
            EventLoop::cancel($cbId);
            fclose($client);
            return;
        }
        // Echo server: send back what we received
        fwrite($client, $data);
    });
});

// onSignal() — реакция на POSIX сигналы
EventLoop::onSignal(SIGINT, function (): void {
    echo "\nReceived SIGINT, shutting down...\n";
    EventLoop::getDriver()->stop();
});

EventLoop::run();

Suspension API

<?php
declare(strict_types=1);

use Revolt\EventLoop;
use Revolt\EventLoop\Suspension;

// Suspension — мост между event loop и Fibers
// Позволяет приостановить Fiber до наступления события

function asyncDelay(float $seconds): void
{
    $suspension = EventLoop::getSuspension();

    EventLoop::delay($seconds, function () use ($suspension): void {
        $suspension->resume();  // Wake up the fiber
    });

    $suspension->suspend();  // Pause the fiber
}

function asyncReadLine(string $prompt): string
{
    echo $prompt;

    $suspension = EventLoop::getSuspension();

    EventLoop::onReadable(STDIN, function (string $id) use ($suspension): void {
        EventLoop::cancel($id);
        $line = trim((string) fgets(STDIN));
        $suspension->resume($line);
    });

    return $suspension->suspend();
}

// Usage inside a fiber (Revolt runs code in fibers automatically)
EventLoop::queue(function (): void {
    echo "Starting...\n";
    asyncDelay(1.0);
    echo "1 second passed\n";
    asyncDelay(0.5);
    echo "0.5 more seconds\n";
});

EventLoop::run();

Отмена (Cancellation)

<?php
declare(strict_types=1);

use Revolt\EventLoop;

// Cancel a callback by its ID
$id = EventLoop::delay(5.0, function (): void {
    echo "This will not execute\n";
});

EventLoop::delay(1.0, function () use ($id): void {
    EventLoop::cancel($id);
    echo "Cancelled the 5-second timer\n";
});

// Enable/Disable callbacks (pause/unpause)
$repeatId = EventLoop::repeat(0.5, function (): void {
    echo "Tick\n";
});

EventLoop::delay(2.0, function () use ($repeatId): void {
    EventLoop::disable($repeatId);
    echo "Paused ticking\n";
});

EventLoop::delay(4.0, function () use ($repeatId): void {
    EventLoop::enable($repeatId);
    echo "Resumed ticking\n";
});

EventLoop::delay(6.0, function () use ($repeatId): void {
    EventLoop::cancel($repeatId);
    echo "Stopped\n";
});

EventLoop::run();

AMPHP v3

AMPHP v3 — набор библиотек для асинхронного PHP, построенных на Revolt event loop и Fibers. Ключевое отличие от v2: вместо Promise используется Future, а благодаря Fibers async код выглядит как синхронный.

Ключевые концепции

<?php
declare(strict_types=1);

use Amp\Future;
use function Amp\async;
use function Amp\delay;

// async() — запустить функцию конкурентно (в отдельном Fiber)
$future1 = async(function (): string {
    delay(1);  // Non-blocking sleep (1 second)
    return 'Result from task 1';
});

$future2 = async(function (): string {
    delay(0.5);
    return 'Result from task 2';
});

// await() — дождаться результата (блокирует текущий Fiber, не поток!)
$result1 = $future1->await();  // Waits for future to complete
$result2 = $future2->await();

echo "{$result1}\n{$result2}\n";
// Total time: ~1 second (not 1.5), because tasks run concurrently

Amp\Future

<?php
declare(strict_types=1);

use Amp\Future;
use function Amp\async;
use function Amp\delay;

// Future::await() — получить результат
$future = async(fn () => 42);
$value = $future->await();  // 42

// Future::await() with timeout
use Amp\TimeoutCancellation;

$future = async(function (): string {
    delay(10);  // 10 seconds
    return 'done';
});

try {
    // Wait max 2 seconds
    $result = $future->await(new TimeoutCancellation(2));
} catch (\Amp\CancelledException $e) {
    echo "Timed out!\n";
}

// Future::map() — transform the result
$future = async(fn () => 42);
$doubled = $future->map(fn (int $v) => $v * 2);
echo $doubled->await();  // 84

// Future::catch() — handle errors
$future = async(function (): never {
    throw new RuntimeException('Oops');
});
$safe = $future->catch(fn (\Throwable $e) => "Error: {$e->getMessage()}");
echo $safe->await();  // "Error: Oops"

// Await multiple futures
$futures = [];
for ($i = 0; $i < 5; $i++) {
    $futures[] = async(function () use ($i): int {
        delay(0.1 * $i);
        return $i * 10;
    });
}

// Wait for ALL futures
$results = Future\await($futures);
// [0, 10, 20, 30, 40]

// Wait for FIRST completed future
$firstResult = Future\awaitFirst($futures);

// Wait for ANY N futures
$anyTwo = Future\awaitAny($futures, 2);

Async HTTP клиент (amphp/http-client)

composer require amphp/http-client

<?php
declare(strict_types=1);

use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Request;
use function Amp\async;
use function Amp\Future;

// Create HTTP client
$client = HttpClientBuilder::buildDefault();

// Concurrent requests
$urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/1',
    'https://httpbin.org/get',
    'https://httpbin.org/ip',
];

$startTime = microtime(true);

$futures = array_map(
    fn (string $url) => async(function () use ($client, $url): string {
        $request = new Request($url);
        $response = $client->request($request);
        $body = $response->getBody()->buffer();
        return "URL: {$url} — Status: {$response->getStatus()}, Size: " . strlen($body);
    }),
    $urls,
);

// Wait for all responses
$results = Future\await($futures);

$elapsed = microtime(true) - $startTime;
echo "Completed in " . number_format($elapsed, 2) . " seconds\n";
// ~2 seconds (not 5), because requests run concurrently

foreach ($results as $result) {
    echo "{$result}\n";
}

POST запрос с телом

<?php
declare(strict_types=1);

use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Request;
use Amp\Http\Client\Body\FormBody;
use Amp\Http\Client\Body\JsonBody;

$client = HttpClientBuilder::buildDefault();

// JSON POST
$request = new Request('https://httpbin.org/post', 'POST');
$request->setHeader('Content-Type', 'application/json');
$request->setBody(json_encode(['name' => 'John', 'age' => 30]));

$response = $client->request($request);
echo $response->getBody()->buffer();

// Form POST
$body = new FormBody();
$body->addField('username', 'admin');
$body->addField('password', 'secret');

$request = new Request('https://httpbin.org/post', 'POST');
$request->setBody($body);

$response = $client->request($request);
echo $response->getBody()->buffer();

Параллельные вычисления (amphp/parallel)

composer require amphp/parallel

<?php
declare(strict_types=1);

use Amp\Parallel\Worker;
use function Amp\async;
use function Amp\Future;

// Run CPU-intensive tasks in separate processes/threads
$futures = [];
for ($i = 0; $i < 4; $i++) {
    $futures[] = async(function () use ($i): string {
        // This runs in a SEPARATE worker process
        $result = Worker\submit(new class($i) implements Worker\Task {
            public function __construct(
                private readonly int $id,
            ) {}

            public function run(Worker\Sync $sync): string
            {
                // Heavy computation in a worker
                $hash = hash('sha256', str_repeat((string) $this->id, 1_000_000));
                return "Worker {$this->id}: {$hash}";
            }
        });

        return $result->await();
    });
}

$results = Future\await($futures);
foreach ($results as $result) {
    echo "{$result}\n";
}

Async TCP-сокет (amphp/socket)

composer require amphp/socket

<?php
declare(strict_types=1);

use Amp\Socket;
use function Amp\async;

// Simple TCP echo server
$server = Socket\listen('127.0.0.1:1337');
echo "Server listening on 127.0.0.1:1337\n";

while ($client = $server->accept()) {
    // Each client handled in its own fiber
    async(function () use ($client): void {
        $address = $client->getRemoteAddress()->toString();
        echo "Client connected: {$address}\n";

        while (($chunk = $client->read()) !== null) {
            echo "Received from {$address}: {$chunk}";
            $client->write("Echo: {$chunk}");
        }

        echo "Client disconnected: {$address}\n";
        $client->close();
    });
}

ReactPHP

ReactPHP — первая и самая зрелая async-библиотека для PHP. Появилась задолго до Fibers и основана на callback/promise подходе. С PHP 8.1 также поддерживает Fibers через адаптеры.

Event Loop

<?php
declare(strict_types=1);

// ReactPHP v3 uses event loop implicitly through the Loop class
use React\EventLoop\Loop;

// Timers
Loop::addTimer(1.0, function (): void {
    echo "Executed once after 1 second\n";
});

$counter = 0;
Loop::addPeriodicTimer(0.5, function ($timer) use (&$counter): void {
    echo "Tick " . (++$counter) . "\n";
    if ($counter >= 5) {
        Loop::cancelTimer($timer);
    }
});

// Deferred execution
Loop::futureTick(function (): void {
    echo "Runs on next tick\n";
});

// The loop runs automatically when the script has pending events

Promises

<?php
declare(strict_types=1);

use React\Promise\Deferred;
use React\Promise\PromiseInterface;
use function React\Promise\resolve;
use function React\Promise\reject;
use function React\Promise\all;
use function React\Promise\race;
use function React\Promise\any;

// Creating a promise
function asyncOperation(): PromiseInterface
{
    $deferred = new Deferred();

    // Simulate async work
    Loop::addTimer(1.0, function () use ($deferred): void {
        $deferred->resolve('Operation completed');
    });

    return $deferred->promise();
}

// Using promises
asyncOperation()->then(
    function (string $result): void {
        echo "Success: {$result}\n";
    },
    function (\Throwable $error): void {
        echo "Error: {$error->getMessage()}\n";
    }
);

// Chaining promises
asyncOperation()
    ->then(fn (string $result) => strtoupper($result))
    ->then(fn (string $upper) => "Processed: {$upper}")
    ->then(fn (string $final) => echo "{$final}\n");

// Immediate values
$resolved = resolve(42);
$rejected = reject(new RuntimeException('Failure'));

// Combining promises
$promise1 = resolve('A');
$promise2 = resolve('B');
$promise3 = resolve('C');

// all() — wait for ALL (fail on first error)
all([$promise1, $promise2, $promise3])
    ->then(fn (array $results) => print_r($results));
// ['A', 'B', 'C']

// race() — first to settle (resolve or reject)
race([$promise1, $promise2])
    ->then(fn ($first) => echo "First: {$first}\n");

// any() — first to RESOLVE (ignores rejections)
any([$rejected, $promise1, $promise2])
    ->then(fn ($first) => echo "First success: {$first}\n");

Async HTTP-сервер (reactphp/http)

composer require react/http

<?php
declare(strict_types=1);

use React\Http\HttpServer;
use React\Http\Message\Response;
use React\Socket\SocketServer;
use Psr\Http\Message\ServerRequestInterface;

// Create HTTP server
$http = new HttpServer(function (ServerRequestInterface $request): Response {
    $method = $request->getMethod();
    $path = $request->getUri()->getPath();

    return match (true) {
        $path === '/' => new Response(
            status: 200,
            headers: ['Content-Type' => 'text/plain'],
            body: "Hello from ReactPHP!\n",
        ),
        $path === '/json' => new Response(
            status: 200,
            headers: ['Content-Type' => 'application/json'],
            body: json_encode(['time' => time(), 'method' => $method]),
        ),
        $path === '/stream' => new Response(
            status: 200,
            headers: ['Content-Type' => 'text/plain'],
            body: generateStream(),  // Returns a ReadableStreamInterface
        ),
        default => new Response(
            status: 404,
            body: "Not Found\n",
        ),
    };
});

$socket = new SocketServer('0.0.0.0:8080');
$http->listen($socket);

echo "Server running at http://127.0.0.1:8080\n";

Async HTTP-клиент (reactphp/http)

<?php
declare(strict_types=1);

use React\Http\Browser;
use React\EventLoop\Loop;
use function React\Promise\all;

$browser = new Browser();

// GET request
$browser->get('https://httpbin.org/get')
    ->then(function (Psr\Http\Message\ResponseInterface $response): void {
        echo "Status: {$response->getStatusCode()}\n";
        echo "Body: {$response->getBody()}\n";
    });

// POST request with JSON
$browser->post(
    'https://httpbin.org/post',
    ['Content-Type' => 'application/json'],
    json_encode(['key' => 'value']),
)->then(function ($response): void {
    echo $response->getBody() . "\n";
});

// Concurrent requests
$urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/1',
];

$promises = array_map(
    fn (string $url) => $browser->get($url),
    $urls,
);

$start = microtime(true);

all($promises)->then(function (array $responses) use ($start): void {
    $elapsed = microtime(true) - $start;
    echo "All " . count($responses) . " requests completed in "
        . number_format($elapsed, 2) . " seconds\n";
    // ~2 seconds (not 4)
});

TCP-сервер (reactphp/socket)

<?php
declare(strict_types=1);

use React\Socket\SocketServer;
use React\Socket\ConnectionInterface;

$server = new SocketServer('127.0.0.1:4000');

$server->on('connection', function (ConnectionInterface $conn): void {
    $addr = $conn->getRemoteAddress();
    echo "New connection: {$addr}\n";

    $conn->write("Welcome! Type something:\n");

    $conn->on('data', function (string $data) use ($conn, $addr): void {
        $data = trim($data);
        echo "[{$addr}] {$data}\n";
        $conn->write("Echo: {$data}\n");

        if ($data === 'quit') {
            $conn->end("Bye!\n");
        }
    });

    $conn->on('close', function () use ($addr): void {
        echo "Connection closed: {$addr}\n";
    });
});

echo "TCP server listening on 127.0.0.1:4000\n";

Дочерние процессы (reactphp/child-process)

composer require react/child-process

<?php
declare(strict_types=1);

use React\ChildProcess\Process;
use React\EventLoop\Loop;

// Run external command
$process = new Process('ls -la /tmp');

$process->start();

$process->stdout->on('data', function (string $chunk): void {
    echo "STDOUT: {$chunk}";
});

$process->stderr->on('data', function (string $chunk): void {
    echo "STDERR: {$chunk}";
});

$process->on('exit', function (int $exitCode, ?int $termSignal): void {
    echo "Process exited with code {$exitCode}\n";
});

// Run PHP script in background
$worker = new Process('php worker.php');
$worker->start();

// Send data to worker via stdin
$worker->stdin->write("process this data\n");
$worker->stdin->end();

Сравнение: AMPHP v3 vs ReactPHP

Стиль кода — ключевое отличие

<?php
declare(strict_types=1);

// AMPHP v3 — выглядит как синхронный код
use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Request;

function amphpStyle(): void
{
    $client = HttpClientBuilder::buildDefault();

    // Looks synchronous, but is non-blocking!
    $response = $client->request(new Request('https://example.com'));
    $body = $response->getBody()->buffer();

    echo "Got: " . strlen($body) . " bytes\n";
}

// ReactPHP — callback/promise chain
use React\Http\Browser;

function reactStyle(): void
{
    $browser = new Browser();

    $browser->get('https://example.com')
        ->then(function ($response) {
            $body = (string) $response->getBody();
            echo "Got: " . strlen($body) . " bytes\n";
        })
        ->catch(function (\Throwable $e) {
            echo "Error: {$e->getMessage()}\n";
        });
}

Рекомендация: Для новых проектов предпочитайте AMPHP v3 — код выглядит как обычный PHP, легче читается и отлаживается. ReactPHP отлично подходит, если вы уже знакомы с его экосистемой или используете библиотеки, которые есть только для ReactPHP.

Когда использовать Async PHP

Подходит отлично

• WebSocket-сервер — постоянные соединения, двусторонний обмен
• Real-time чат — множество одновременных подключений
• API Gateway — агрегация данных из нескольких сервисов
• Long Polling — держать соединение открытым
• Streaming — передача данных по частям (Server-Sent Events)
• Очереди задач — обработка сообщений из RabbitMQ/Redis
• Микросервисы — легковесные HTTP-серверы
• Веб-скрапинг — параллельный парсинг множества URL
• CLI-инструменты — параллельное выполнение задач

НЕ подходит

• Типичный CRUD API — один запрос = один ответ, PHP-FPM справляется отлично
• Рендеринг HTML-страниц — нет выигрыша от async
• CPU-интенсивные задачи — async не добавляет параллелизм (один поток)
• Простые скрипты — оверхед не оправдан

Правило большого пальца:
├── Много I/O ожидания?          → Async поможет
├── Много одновременных клиентов? → Async поможет
├── CPU-bound вычисления?         → Async НЕ поможет (используй amphp/parallel)
└── Простой request/response?     → PHP-FPM достаточно

Практический пример: Async HTTP-клиент

10 параллельных запросов с AMPHP

<?php
declare(strict_types=1);

use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Request;
use function Amp\async;
use function Amp\Future;

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

$client = HttpClientBuilder::buildDefault();

$urls = [
    'https://jsonplaceholder.typicode.com/posts/1',
    'https://jsonplaceholder.typicode.com/posts/2',
    'https://jsonplaceholder.typicode.com/posts/3',
    'https://jsonplaceholder.typicode.com/users/1',
    'https://jsonplaceholder.typicode.com/users/2',
    'https://jsonplaceholder.typicode.com/comments/1',
    'https://jsonplaceholder.typicode.com/comments/2',
    'https://jsonplaceholder.typicode.com/todos/1',
    'https://jsonplaceholder.typicode.com/albums/1',
    'https://jsonplaceholder.typicode.com/photos/1',
];

$start = hrtime(true);

$futures = array_map(
    fn (string $url) => async(function () use ($client, $url): array {
        $request = new Request($url);
        $response = $client->request($request);
        $body = $response->getBody()->buffer();

        return [
            'url' => $url,
            'status' => $response->getStatus(),
            'size' => strlen($body),
            'data' => json_decode($body, true),
        ];
    }),
    $urls,
);

// Await all — returns when ALL are done
$results = Future\await($futures);

$elapsed = (hrtime(true) - $start) / 1_000_000;  // Convert to ms

echo "Fetched " . count($results) . " URLs in " . number_format($elapsed, 1) . " ms\n\n";

foreach ($results as $result) {
    echo sprintf(
        "  %s — %d (%d bytes)\n",
        $result['url'],
        $result['status'],
        $result['size'],
    );
}

// Sequential comparison:
// 10 requests × ~100ms each = ~1000ms
// Concurrent: ~100-200ms (all in parallel)

WebSocket Echo Server с AMPHP

composer require amphp/websocket-server amphp/http-server

<?php
declare(strict_types=1);

use Amp\Http\HttpStatus;
use Amp\Http\Server\DefaultErrorHandler;
use Amp\Http\Server\Request;
use Amp\Http\Server\Response;
use Amp\Http\Server\Router;
use Amp\Http\Server\SocketHttpServer;
use Amp\Socket;
use Amp\Websocket\Server\Websocket;
use Amp\Websocket\Server\WebsocketClientHandler;
use Amp\Websocket\Server\WebsocketGateway;
use Amp\Websocket\Server\WebsocketAcceptHandler;
use Amp\Websocket\WebsocketClient;
use Psr\Log\NullLogger;

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

// WebSocket handler
$handler = new class implements WebsocketClientHandler {
    public function handleClient(
        WebsocketClient $client,
        Request $request,
        Response $response,
    ): void {
        $addr = $request->getClient()->getRemoteAddress()->toString();
        echo "Client connected: {$addr}\n";

        foreach ($client as $message) {
            $payload = $message->buffer();
            echo "[{$addr}] {$payload}\n";

            // Echo back with timestamp
            $client->sendText(sprintf(
                '[%s] Echo: %s',
                date('H:i:s'),
                $payload,
            ));
        }

        echo "Client disconnected: {$addr}\n";
    }
};

$server = SocketHttpServer::createForDirectAccess(new NullLogger());

$server->expose(new Socket\InternetAddress('127.0.0.1', 9001));

$websocket = new Websocket($server, new NullLogger(), new WebsocketAcceptHandler(), $handler);

$router = new Router($server, new NullLogger(), new DefaultErrorHandler());
$router->addRoute('GET', '/ws', $websocket);

$server->start($router, new DefaultErrorHandler());

echo "WebSocket server running at ws://127.0.0.1:9001/ws\n";

// Keep running until signal
Amp\trapSignal([SIGINT, SIGTERM]);

$server->stop();

Итоговая таблица API