Vodiči

Mastering ReactPHP: Building Non-Blocking PHP Network Services

Ovladavanje ReactPHP-om: Izgradnja neblokirajućih PHP mrežnih usluga

Poslužitelj koji ne blokira lako je demonstrirati, a iznenađujuće teško dovršiti. Prihvaćanje veza tek je početak; produkcijskoj usluzi također su potrebni ograničeni ulazi, kontrola protoka, autentikacija, uredno gašenje, korisna telemetrija i ponašanje pri pogreškama koje jednog problematičnog klijenta ne pretvara u prekid rada.

Ovaj vodič izrađuje potpunu TCP uslugu ključ-vrijednost uz PHP 8.3 i ekosustav ReactPHP 1.x. Klijenti razmjenjuju JSON razdvojen novim redovima preko trajnih veza. Usluga obrađuje mnoge veze u jednom procesu bez stvaranja dretve ili procesa za svakog klijenta.

Preduvjeti i granice verzija

Potreban vam je PHP 8.3 s omogućenim JSON-om, sockets i PCNTL-om, uz Composer 2. Obrada signala zahtijeva sustav nalik Unixu; sam mrežni kôd ne ovisi o PCNTL-u.

Primjer se namjerno zadržava unutar jedne obitelji glavne verzije ReactPHP-a:

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

U implementaciji koristite Composerovu datoteku zaključavanja kako bi svako izdanje dobilo verzije testirane tijekom razvoja.

{
    "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

Arhitektura i kompromisi

SocketServer upravlja utičnicom za osluškivanje. Petlja događaja ReactPHP-a nadzire je, obrađuje dolazne podatke te raspoređuje mjerače vremena i signale. Svaka veza ima mali objekt stanja koji sadrži njezin nedovršeni okvir, vremensku oznaku aktivnosti i prozor ograničenja brzine. Mapa ključ-vrijednost lokalna je za proces.

Ovaj dizajn ima korisna svojstva: nema blokirajućih čitanja, nema radnika po klijentu, ograničenja protokola su predvidljiva, a trajne veze su jeftine. Također ima jasne granice. Podaci nestaju pri ponovnom pokretanju, jedan proces ne može dijeliti stanje s replikama, a procesorički zahtjevni obrađivači zahtjeva zaustavili bi svaku vezu. Stvarna perzistencija ili horizontalno skaliranje zahtijevali bi asinkrono vanjsko spremište ili namjerno dijeljenje na particije.

Protokol zahtjev-odgovor pomaže pri povratnom pritisku. ReactPHP-ov dvosmjerni tok povezuje zagušenje pri pisanju s čitanjem iz iste veze, omogućujući TCP kontroli protoka da uspori klijenta čiji se odgovori ne mogu isprazniti. Za razliku od poslužitelja za emitiranje, ova usluga ne akumulira neovisne redove za razgranavanje na razini aplikacije.

Struktura projekta

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

Implementacija poslužitelja

Poslužitelj prihvaća ping, set, get, delete i stats. Svaki zahtjev mora sadržavati konfigurirani token. Okviri, vrijednosti, ključevi, klijenti, pohranjeni unosi i stope zahtjeva ograničeni su.

<?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();

Obrada najviše 32 okvira po prolazu petlje događaja važna je. Bez tog ograničenja pravednosti, klijent koji isporuči veliku skupinu mogao bi monopolizirati proces dok druge spremne utičnice i mjerači vremena čekaju.

Izrada determinističkog testnog klijenta

Testni klijent namjerno koristi blokirajući PHP: predstavlja običnog vanjskog potrošača i postavlja izričita ograničenja od dvije sekunde za uspostavu veze i čitanja. Njegov pomoćnik za pisanje obrađuje djelomična pisanja umjesto da pretpostavlja kako jedan fwrite() prenosi cijeli okvir.

<?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

# In a second terminal, with the same environment value:
php client.php

# Exercise concurrent connections without modifying the service:
seq 1 50 | xargs -P 10 -I '{}' php client.php >/dev/null

Uspješno pokretanje vraća pong, potvrđuje zapisivanje, čita ready i prijavljuje barem jedan pohranjeni ključ. Također testirajte neispravan JSON, prevelik okvir, pogrešan token, neaktivne veze i više od 100 zahtjeva unutar deset sekundi. Ti putovi trebali bi proizvesti izričite pogreške ili uredno zatvaranje, nikada neograničen rast memorije.

Sigurnost i mrežna izloženost

Zadano vezanje na povratnu petlju namjerno je. Token štiti pristup protokolu, ali ne šifrira promet. Nemojte ovu uslugu u običnom tekstu izravno izlagati nepouzdanoj mreži. Za udaljeni pristup postavite je iza uzajamno autentificiranog TLS tunela, VPN-a ili namjenskog TLS proxyja, a zatim ograničite vatrozid glavnog računala na tu pouzdanu ulaznu točku.

Generirajte snažan token s pomoću openssl rand -hex 32, pohranite ga izvan direktorija aplikacije i nikada ga ne stavljajte u zapisnike ni upravljanje izvornim kodom. Poslužitelj koristi hash_equals(), zatvara neautorizirane veze, ograničava sintaksu ključeva te ograničava vrijednosti i ukupan broj unosa. Autentikacija je i dalje rad na razini veze za svaki zahtjev; ako se protokol razvija, autentificirano rukovanje s izričitim stanjem sesije može biti učinkovitije.

Promatranje i performanse

Zapisnici su JSON razdvojen novim redovima, što ih čini prikladnima za sistemski dnevnik ili prikupljač zapisnika. Kao minimum, upozoravajte na ponovljene pogreške veze, odbijanja zbog kapaciteta, neuspjele autentikacije dodane kao zbirni brojači, neočekivana ponovna pokretanja i memoriju koja se približava ograničenju usluge.

Operacija stats korisna je za provjeru, ali ne treba je zamijeniti s potpunim sustavom metrika. Produkcijsko proširenje moglo bi izložiti brojače putem zasebne HTTP krajnje točke dostupne samo preko povratne petlje. Izbjegavajte oznake velike kardinalnosti, kao što su neobrađene udaljene adrese ili ključevi.

Obrađivače održavajte kratkima i neblokirajućima. Operacije s datotekama, sinkroni klijenti baza podataka, naredbe ljuske, raspršivanje lozinki, kompresija i velike JSON transformacije mogu zaustaviti petlju događaja. Prebacite procesorski zahtjevan rad ograničenim radnicima i za mrežne ovisnosti koristite asinkrone klijente kompatibilne s ReactPHP-om. Svaka ovisnost treba vlastito vremensko ograničenje veze i operacije; samo vremensko ograničenje veze ne ograničava kasnija čitanja ni upite.

Implementacija pod systemd-om

Postavite pregledano izdanje u /opt/react-kv i instalirajte ovisnosti s composer install --no-dev --classmap-authoritative. Composer pokrenite tijekom izrade izdanja, a ne iz dugotrajne usluge.

[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

Izradite /etc/react-kv.env kao datoteku u vlasništvu roota s načinom rada 0600, koja sadržava REACT_KV_TOKEN=... i REACT_KV_LISTEN=127.0.0.1:9010. Instalirajte jedinicu kao /etc/systemd/system/react-kv.service. Te operacije na razini glavnog računala zahtijevaju root ovlasti; najprije pregledajte te točne putanje i ne prepisujte nenamjerno postojeću uslugu ili konfiguraciju.

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

Proračun za gašenje usluge iznosi deset sekundi, a systemd dopušta dvanaest. Pri završavanju se najprije zatvara osluškivač, postojeći klijenti primaju okvir za gašenje, a preostale se utičnice prisilno zatvaraju po isteku roka. Budući da je osluškivač dostupan samo preko povratne petlje, dolazno pravilo vatrozida nije potrebno ni poželjno.

Uobičajeni kvarovi

  • Poslužitelj se pokrene i odmah prekine: provjerite da token ima najmanje 32 bajta i da priključak 9010 nije već zauzet.
  • Klijenti čekaju zauvijek: provjerite završava li svaki okvir novim redom i konfigurira li svaki klijent neovisna vremenska ograničenja veze i čitanja.
  • Jedan klijent uzrokuje skokove latencije: potražite blokirajući rad ili skupe transformacije unutar obrađivača zahtjeva.
  • Memorija raste: potvrdite da se ograničenja ključeva i vrijednosti i dalje primjenjuju, zatim pregledajte broj veza i PHP proširenja umjesto da samo povećate MemoryMax.
  • Signali ne rade: potvrdite da je PCNTL omogućen u CLI PHP binarnoj datoteci koju koristi systemd, a ne samo u nekoj drugoj PHP instalaciji.
  • Udaljeni klijenti ne mogu se povezati: vezanje na povratnu petlju radi prema namjeri. Dodajte sigurni tunel umjesto da olako promijenite osluškivač na sva sučelja.

Završni kontrolni popis za provjeru

  • Composer instalira zaključane ReactPHP 1.x ovisnosti pod PHP-om 8.3.
  • Osluškivač je vidljiv samo na 127.0.0.1:9010.
  • Valjani zahtjevi set, get, delete, ping i stats uspijevaju.
  • Neispravna autentikacija, neispravan JSON, prekomjerne stope, dugi okviri i neaktivni klijenti odbijaju se predvidljivo.
  • Istodobni testni klijenti završavaju bez zastoja petlje događaja.
  • Zapisnici stižu u dnevnik kao valjan JSON bez tokena ili pohranjenih vrijednosti.
  • SIGTERM zaustavlja nova prihvaćanja i završava unutar systemd-ovog proračuna za gašenje.
  • Ograničenja memorije i opisnika datoteka odgovaraju konfiguriranom kapacitetu klijenata.

Najvrjednija lekcija nije da PHP može držati otvorenima stotine utičnica. Nego da neblokirajući softver mora svaku granicu učiniti izričitom: koliko čita, koliko čeka, koliko pravedno raspoređuje, što izlaže i kako se zaustavlja. Kada su te granice oblikovane umjesto pretpostavljene, ReactPHP postaje discipliniran temelj za mrežne usluge, a ne samo pametna demonstracija petlje događaja.

Portret autora bloga

Mihajlo

Ja sam Mihajlo — programer vođen znatiželjom, disciplinom i stalnom željom da stvorim nešto smisleno. Dijelim uvide, tutorijale i besplatne usluge kako bih pomogao drugima da pojednostave svoj rad i rastu u svijetu softvera i umjetne inteligencije koji se neprestano razvija.

Pretraži

Kategorije

Nedavni postovi

Arhitektov nacrt za izgradnju umjetne inteligencije koja doista uči
Usavršite svoje API-je kako biste AI-ju pružili stvarnu poslovnu logiku
Vlasništvo nad sustavom: oblikovanje poslovne intuicije umjetne inteligencije kroz kod
Ugradite poslovnu logiku: osmislite softver koji umjetna inteligencija zaista može naučiti
Refactor for Reliability: Why Your Backend Needs a Database Overhaul