Совладување на ReactPHP: Градење неблокирачки PHP мрежни услуги

Mastering ReactPHP: Building Non-Blocking PHP Network Services

Совладување на ReactPHP: Градење неблокирачки PHP мрежни услуги

Неблокирачки сервер лесно се демонстрира, но изненадувачки тешко се довршува. Прифаќањето врски е само почеток; на продукциска услуга ѝ се потребни и ограничени влезови, контрола на протокот, автентикација, уредно гасење, корисна телеметрија и однесување при неуспех што нема еден проблематичен клиент да го претвори во прекин на услугата.

Ова упатство изработува целосна TCP услуга за клуч-вредност со PHP 8.3 и екосистемот ReactPHP 1.x. Клиентите разменуваат JSON ограничен со нови редови преку трајни врски. Услугата обработува многу врски во еден процес без создавање нишка или процес за секој клиент.

Предуслови и верзиски граници

Потребен ви е PHP 8.3 со овозможени JSON, sockets и PCNTL, како и Composer 2. Обработката на сигнали бара Unix-сличен хост; самиот мрежен код не зависи од PCNTL.

Примерот намерно останува во рамките на едно семејство на главни верзии на ReactPHP:

react/event-loop:^1.5
react/socket:^1.16

Користете ја lock-датотеката на Composer при распоредување, за секое издание да ги добие верзиите тестирани во развојот.

{
    "name": "example/react-kv",
    "type": "project",
    "require": {
        "php": "^8.3",
        "ext-json": "*",
        "ext-pcntl": "*",
        "react/event-loop": "^1.5",
        "react/socket": "^1.16"
    },
    "config": {
        "sort-packages": true
    }
}

composer install
php -m | grep -E 'json|pcntl|sockets'
composer show react/event-loop react/socket

Архитектура и компромиси

SocketServer го поседува сокетот за слушање. Јамката за настани на ReactPHP го надгледува, ги распределува влезните податоци и закажува тајмери и сигнали. Секоја врска има мал објект за состојба што ја содржи нејзината недовршена рамка, временската ознака на активност и прозорецот за ограничување на стапката. Мапата клуч-вредност е локална за процесот.

Овој дизајн има корисни својства: нема блокирачки читања, нема работници по клиент, предвидливи ограничувања на протоколот и евтини трајни врски. Има и јасни граници. Податоците исчезнуваат при рестартирање, еден процес не може да дели состојба со реплики, а обработувачите на барања што трошат многу CPU би ја закочиле секоја врска. Вистинска трајност или хоризонтално скалирање би барале асинхроно надворешно складиште или намерно партиционирање.

Протоколот барање-одговор помага со повратниот притисок. Дуплексниот поток на ReactPHP го поврзува застојот при запишување со читањето од истата врска, овозможувајќи TCP контролата на протокот да го забави клиентот чии одговори не можат да се испразнат. За разлика од сервер за емитување, оваа услуга не акумулира независни редици за разгорување на ниво на апликацијата.

Структура на проектот

react-kv/
├── composer.json
├── composer.lock
├── server.php
├── client.php
└── vendor/

Имплементирање на серверот

Серверот прифаќа ping, set, get, delete и stats. Секое барање мора да го носи конфигурираниот токен. Рамките, вредностите, клучевите, клиентите, складираните записи и стапките на барања се ограничени.

<?php

declare(strict_types=1);

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

use React\EventLoop\Loop;
use React\Socket\ConnectionInterface;
use React\Socket\SocketServer;

final class ClientState
{
    public string $buffer = '';
    public float $lastActivity;
    public float $windowStarted;
    public int $requestsInWindow = 0;
    public bool $processing = false;

    public function __construct(public readonly ConnectionInterface $connection)
    {
        $this->lastActivity = microtime(true);
        $this->windowStarted = $this->lastActivity;
    }
}

const MAX_CLIENTS = 256;
const MAX_INPUT_BUFFER = 65_536;
const MAX_LINE = 8_192;
const MAX_VALUE = 4_096;
const MAX_KEYS = 10_000;
const REQUESTS_PER_WINDOW = 100;
const RATE_WINDOW_SECONDS = 10.0;
const IDLE_TIMEOUT_SECONDS = 30.0;
const SHUTDOWN_BUDGET_SECONDS = 10.0;
const FRAMES_PER_TICK = 32;

$listen = getenv('REACT_KV_LISTEN') ?: '127.0.0.1:9010';
$token = getenv('REACT_KV_TOKEN');

if ($token === false || strlen($token) < 32) {
    fwrite(STDERR, "REACT_KV_TOKEN must contain at least 32 bytes\n");
    exit(1);
}

$log = static function (string $event, array $context = []): void {
    $record = ['time' => gmdate('c'), 'event' => $event] + $context;
    fwrite(STDOUT, json_encode($record, JSON_THROW_ON_ERROR) . "\n");
};

$server = new SocketServer($listen);
$clients = new SplObjectStorage();
$store = [];
$stopping = false;

$encode = static fn(array $message): string =>
    json_encode($message, JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES) . "\n";

$closeWith = static function (
    ClientState $state,
    array $message
) use ($encode): void {
    $state->connection->end($encode($message));
};

$handle = static function (
    ClientState $state,
    string $line
) use (&$store, $token, $encode, $closeWith): void {
    $now = microtime(true);

    if ($now - $state->windowStarted >= RATE_WINDOW_SECONDS) {
        $state->windowStarted = $now;
        $state->requestsInWindow = 0;
    }

    if (++$state->requestsInWindow > REQUESTS_PER_WINDOW) {
        $closeWith($state, ['ok' => false, 'error' => 'rate_limit']);
        return;
    }

    try {
        $request = json_decode($line, true, 16, JSON_THROW_ON_ERROR);
    } catch (JsonException) {
        $state->connection->write(
            $encode(['ok' => false, 'error' => 'invalid_json'])
        );
        return;
    }

    if (!is_array($request) ||
        !isset($request['token']) ||
        !is_string($request['token']) ||
        !hash_equals($token, $request['token'])) {
        $closeWith($state, ['ok' => false, 'error' => 'unauthorized']);
        return;
    }

    $operation = $request['op'] ?? null;

    if ($operation === 'ping') {
        $state->connection->write($encode(['ok' => true, 'pong' => true]));
        return;
    }

    if ($operation === 'stats') {
        $state->connection->write($encode([
            'ok' => true,
            'keys' => count($store),
            'memory_bytes' => memory_get_usage(true),
        ]));
        return;
    }

    $key = $request['key'] ?? null;
    if (!is_string($key) ||
        preg_match('/\A[a-zA-Z0-9:._-]{1,128}\z/', $key) !== 1) {
        $state->connection->write(
            $encode(['ok' => false, 'error' => 'invalid_key'])
        );
        return;
    }

    if ($operation === 'get') {
        $found = array_key_exists($key, $store);
        $state->connection->write($encode([
            'ok' => true,
            'found' => $found,
            'value' => $found ? $store[$key] : null,
        ]));
        return;
    }

    if ($operation === 'delete') {
        $deleted = array_key_exists($key, $store);
        unset($store[$key]);
        $state->connection->write(
            $encode(['ok' => true, 'deleted' => $deleted])
        );
        return;
    }

    if ($operation === 'set') {
        $value = $request['value'] ?? null;

        if (!is_string($value) || strlen($value) > MAX_VALUE) {
            $state->connection->write(
                $encode(['ok' => false, 'error' => 'invalid_value'])
            );
            return;
        }

        if (!array_key_exists($key, $store) && count($store) >= MAX_KEYS) {
            $state->connection->write(
                $encode(['ok' => false, 'error' => 'capacity_reached'])
            );
            return;
        }

        $store[$key] = $value;
        $state->connection->write($encode(['ok' => true]));
        return;
    }

    $state->connection->write(
        $encode(['ok' => false, 'error' => 'unknown_operation'])
    );
};

$process = null;
$process = static function (ClientState $state) use (
    &$process,
    $clients,
    $handle,
    $closeWith
): void {
    if (!$clients->contains($state->connection)) {
        return;
    }

    $processed = 0;

    while ($processed < FRAMES_PER_TICK) {
        $newline = strpos($state->buffer, "\n");

        if ($newline === false) {
            if (strlen($state->buffer) > MAX_LINE) {
                $closeWith($state, ['ok' => false, 'error' => 'frame_too_large']);
            }
            $state->processing = false;
            return;
        }

        if ($newline > MAX_LINE) {
            $closeWith($state, ['ok' => false, 'error' => 'frame_too_large']);
            return;
        }

        $line = rtrim(substr($state->buffer, 0, $newline), "\r");
        $state->buffer = substr($state->buffer, $newline + 1);
        ++$processed;

        if ($line !== '') {
            $handle($state, $line);
        }
    }

    Loop::futureTick(static fn() => $process($state));
};

$server->on('connection', static function (
    ConnectionInterface $connection
) use (
    $clients,
    &$stopping,
    $process,
    $closeWith,
    $log
): void {
    if ($stopping || count($clients) >= MAX_CLIENTS) {
        $connection->end("{\"ok\":false,\"error\":\"unavailable\"}\n");
        return;
    }

    $state = new ClientState($connection);
    $clients->attach($connection, $state);
    $log('client_connected', ['remote' => $connection->getRemoteAddress()]);

    $connection->on('data', static function (string $chunk) use (
        $state,
        $process,
        $closeWith
    ): void {
        $state->lastActivity = microtime(true);
        $state->buffer .= $chunk;

        if (strlen($state->buffer) > MAX_INPUT_BUFFER) {
            $closeWith($state, ['ok' => false, 'error' => 'buffer_limit']);
            return;
        }

        if (!$state->processing) {
            $state->processing = true;
            Loop::futureTick(static fn() => $process($state));
        }
    });

    $connection->on('error', static function (Throwable $error) use ($log): void {
        $log('connection_error', ['message' => $error->getMessage()]);
    });

    $connection->on('close', static function () use (
        $connection,
        $clients,
        $log
    ): void {
        if ($clients->contains($connection)) {
            $clients->detach($connection);
        }
        $log('client_closed', ['remote' => $connection->getRemoteAddress()]);
    });
});

Loop::addPeriodicTimer(5.0, static function () use ($clients, $closeWith): void {
    $cutoff = microtime(true) - IDLE_TIMEOUT_SECONDS;

    foreach ($clients as $connection) {
        $state = $clients[$connection];
        if ($state->lastActivity < $cutoff) {
            $closeWith($state, ['ok' => false, 'error' => 'idle_timeout']);
        }
    }
});

$shutdown = static function (int $signal) use (
    &$stopping,
    $server,
    $clients,
    $encode,
    $log
): void {
    if ($stopping) {
        return;
    }

    $stopping = true;
    $server->close();
    $log('shutdown_started', ['signal' => $signal]);

    foreach ($clients as $connection) {
        $connection->end($encode(['ok' => false, 'error' => 'shutdown']));
    }

    Loop::addTimer(SHUTDOWN_BUDGET_SECONDS, static function () use (
        $clients,
        $log
    ): void {
        foreach ($clients as $connection) {
            $connection->close();
        }
        $log('shutdown_deadline_reached');
    });
};

Loop::addSignal(SIGTERM, $shutdown);
Loop::addSignal(SIGINT, $shutdown);

$log('server_started', ['listen' => $listen]);
Loop::run();

Обработувањето најмногу 32 рамки по вртење на јамката за настани е важно. Без тоа ограничување за правичност, клиент што испраќа голема серија би можел да го монополизира процесот додека други подготвени сокети и тајмери чекаат.

Изградба на детерминистички тест-клиент

Тест-клиентот намерно користи блокирачки PHP: претставува обичен надворешен потрошувач и поставува изрични двосекундни ограничувања за воспоставување врска и читања. Неговиот помошник за запишување обработува делумни запишувања наместо да претпостави дека едно fwrite() пренесува цела рамка.

<?php

declare(strict_types=1);

$token = getenv('REACT_KV_TOKEN');
if ($token === false) {
    throw new RuntimeException('REACT_KV_TOKEN is required');
}

$errno = 0;
$error = '';
$socket = stream_socket_client(
    'tcp://127.0.0.1:9010',
    $errno,
    $error,
    2.0,
    STREAM_CLIENT_CONNECT
);

if ($socket === false) {
    throw new RuntimeException("Connection failed: {$error} ({$errno})");
}

stream_set_timeout($socket, 2);

$requests = [
    ['token' => $token, 'op' => 'ping'],
    ['token' => $token, 'op' => 'set', 'key' => 'release', 'value' => 'ready'],
    ['token' => $token, 'op' => 'get', 'key' => 'release'],
    ['token' => $token, 'op' => 'stats'],
];

foreach ($requests as $request) {
    $payload = json_encode($request, JSON_THROW_ON_ERROR) . "\n";
    $remaining = $payload;

    while ($remaining !== '') {
        $written = fwrite($socket, $remaining);
        if ($written === false || $written === 0) {
            throw new RuntimeException('Socket write failed');
        }
        $remaining = substr($remaining, $written);
    }

    $line = fgets($socket, 16_384);
    $metadata = stream_get_meta_data($socket);

    if ($line === false) {
        $reason = $metadata['timed_out'] ? 'read timeout' : 'connection closed';
        throw new RuntimeException($reason);
    }

    $response = json_decode($line, true, 16, JSON_THROW_ON_ERROR);
    echo json_encode($response, JSON_PRETTY_PRINT | JSON_THROW_ON_ERROR), "\n";
}

fclose($socket);

export REACT_KV_TOKEN='replace-with-at-least-32-random-bytes'
php server.php

# Во втор терминал, со истата вредност на околината:
php client.php

# Тестирајте истовремени врски без менување на услугата:
seq 1 50 | xargs -P 10 -I '{}' php client.php >/dev/null

Успешно извршување враќа pong, го потврдува запишувањето, ја чита вредноста ready и пријавува најмалку еден складиран клуч. Тестирајте и невалиден JSON, преголема рамка, погрешен токен, неактивни врски и повеќе од 100 барања во рок од десет секунди. Тие патеки треба да создадат изрични грешки или уредно затворање, но никогаш неограничен раст на меморијата.

Безбедност и мрежна изложеност

Стандардното поврзување само на loopback е намерно. Токенот го заштитува пристапот до протоколот, но не го шифрира сообраќајот. Не ја изложувајте оваа услуга со обичен текст директно на недоверлива мрежа. За далечински пристап, поставете ја зад TLS-тунел со заемна автентикација, VPN или TLS-прокси наменет за таа цел, а потоа ограничете го заштитниот ѕид на хостот на таа доверлива влезна точка.

Генерирајте силен токен со openssl rand -hex 32, чувајте го надвор од директориумот на апликацијата и никогаш не го ставајте во логови или контрола на изворниот код. Серверот користи hash_equals(), ги затвора неовластените врски, ја ограничува синтаксата на клучевите и ги ограничува вредностите и вкупниот број записи. Автентикацијата сепак е работа на ниво на врска за секое барање; ако протоколот се развие, автентицирано ракување со изрична состојба на сесија може да биде поефикасно.

Набљудливост и перформанси

Логовите се JSON ограничен со нови редови, што ги прави погодни за системскиот журнал или собирач на логови. Како минимум, алармирајте за повторени грешки на врската, одбивања поради капацитет, неуспеси на автентикацијата додадени како збирни бројачи, неочекувани рестартирања и меморија што се приближува до ограничувањето на услугата.

Операцијата stats е корисна за потврда, но не треба да се смета за целосен систем за метрики. Продукциско проширување би можело да изложи бројачи преку посебна HTTP крајна точка достапна само преку loopback. Избегнувајте ознаки со висока кардиналност, како необработени далечински адреси или клучеви.

Одржувајте ги обработувачите кратки и неблокирачки. Операции со датотеки, синхрони клиенти за бази на податоци, shell-команди, хеширање лозинки, компресија и големи JSON трансформации можат да ја закочат јамката за настани. Префрлете ја работата што троши многу CPU на ограничени работници и користете асинхрони клиенти компатибилни со ReactPHP за мрежни зависности. На секоја зависност ѝ треба сопствено истекување на времето за врска и операција; само истекувањето за врска не ги ограничува подоцнежните читања или пребарувања.

Распоредување под systemd

Поставете прегледано издание во /opt/react-kv и инсталирајте зависности со composer install --no-dev --classmap-authoritative. Извршувајте го Composer при изработката на изданието, а не од долготрајната услуга.

[Unit]
Description=ReactPHP non-blocking key-value service
After=network.target

[Service]
Type=simple
DynamicUser=yes
WorkingDirectory=/opt/react-kv
EnvironmentFile=/etc/react-kv.env
ExecStart=/usr/bin/php /opt/react-kv/server.php
Restart=on-failure
RestartSec=2
TimeoutStopSec=12
NoNewPrivileges=yes
PrivateTmp=yes
ProtectSystem=strict
ProtectHome=yes
ProtectProc=invisible
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryMax=256M
LimitNOFILE=1024
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

Создадете /etc/react-kv.env како датотека во сопственост на root со режим 0600, што содржи REACT_KV_TOKEN=... и REACT_KV_LISTEN=127.0.0.1:9010. Инсталирајте ја единицата како /etc/systemd/system/react-kv.service. Овие операции на ниво на хост бараат root-привилегии; прво проверете ги тие точни патеки и не презапишувајте постоечка услуга или конфигурација ненамерно.

sudo systemctl daemon-reload
sudo systemctl enable --now react-kv.service
sudo systemctl status react-kv.service
sudo journalctl -u react-kv.service -n 50 --no-pager
sudo ss -ltnp 'sport = :9010'
sudo systemctl kill --signal=TERM react-kv.service

Буџетот за гасење на услугата е десет секунди, а systemd дозволува дванаесет. При прекин, слушачот прво се затвора, постоечките клиенти добиваат рамка за гасење, а преостанатите сокети насилно се затвораат на крајниот рок. Бидејќи слушачот е само loopback, не е потребно ниту пожелно правило за влезен заштитен ѕид.

Вообичаени неуспеси

Серверот се стартува и веднаш се гаси: проверете дали токенот има најмалку 32 бајти и дали портата 9010 не е веќе зафатена.
Клиентите чекаат засекогаш: осигурете се дека секоја рамка завршува со нов ред и дека секој клиент конфигурира независни истекувања на времето за врска и читање.
Еден клиент предизвикува скокови на латентноста: побарајте блокирачка работа или скапи трансформации во обработувачот на барања.
Меморијата расте: потврдете дека ограничувањата за клучевите и вредностите остануваат на сила, па проверете го бројот на врски и PHP-екстензиите наместо само да го зголемите MemoryMax.
Сигналите не работат: потврдете дека PCNTL е овозможен во CLI PHP-бинарната датотека што ја користи systemd, а не само во друга PHP-инсталација.
Далечинските клиенти не можат да се поврзат: поврзувањето само на loopback работи како што е наменето. Додадете безбеден тунел наместо лежерно да го промените слушачот на сите интерфејси.

Конечна контролна листа за потврда

Composer ги инсталира заклучените ReactPHP 1.x зависности под PHP 8.3.
Слушачот е видлив само на 127.0.0.1:9010.
Валидните барања set, get, delete, ping и stats успеваат.
Невалидна автентикација, невалиден JSON, прекумерни стапки, долги рамки и неактивни клиенти се одбиваат предвидливо.
Истовремените тест-клиенти завршуваат без застои во јамката за настани.
Логовите стигнуваат до журналот како валиден JSON без токени или складирани вредности.
SIGTERM престанува да прифаќа нови врски и завршува во рамките на буџетот за гасење на systemd.
Ограничувањата за меморијата и дескрипторите на датотеки одговараат на конфигурираниот капацитет на клиенти.

Највредната лекција не е дека PHP може да држи отворени стотици сокети. Таа е дека неблокирачкиот софтвер мора да ја направи секоја граница изрична: колку чита, колку долго чека, колку правично закажува, што изложува и како запира. Штом тие граници се дизајнирани наместо претпоставени, ReactPHP станува дисциплинирана основа за мрежни услуги, наместо само паметна демонстрација на јамка за настани.