Server-Sent Events (SSE)

OxPHP przesyła strumieniowo dane w czasie rzeczywistym do klientów za pośrednictwem protokołu Server-Sent Events, z wbudowanym przeciwciśnieniem. Ustaw Content-Type: text/event-stream w swoim skrypcie PHP i wywołaj oxphp_stream_flush(). Resztą zajmie się OxPHP.

Jak to działa

Odpowiedź strumieniowa przechodzi przez przewidywalny cykl życia:

Ustaw nagłówki. Twój skrypt PHP ustawia Content-Type: text/event-stream za pomocą header() i zapisuje wiersze w formacie SSE, używając echo.
Pierwsze opróżnienie bufora otwiera strumień. Pierwsze wywołanie oxphp_stream_flush() wysyła nagłówki HTTP do klienta i przechodzi w tryb strumieniowania. Połączenie z klientem pozostaje otwarte.
Każde opróżnienie bufora wysyła porcję. Każde kolejne wywołanie oxphp_stream_flush() opróżnia zbuforowane dane wyjściowe jako nową porcję, dostarczając ją natychmiast do klienta.
Przeciwciśnienie włącza się, gdy bufor się zapełni. OxPHP utrzymuje wewnętrzny bufor mieszczący do 64 porcji między workerem PHP a klientem. Gdy bufor jest pełny — ponieważ wolny klient nie odebrał wcześniejszych porcji — oxphp_stream_flush() blokuje się do momentu zwolnienia miejsca. Zapobiega to nieograniczonemu wzrostowi zużycia pamięci.
Sprzątanie przy zakończeniu lub rozłączeniu. Gdy skrypt PHP kończy działanie, OxPHP łagodnie zamyka połączenie. Jeśli klient rozłączy się w trakcie strumienia, OxPHP wykrywa zamknięty kanał przy następnym opróżnieniu bufora, ustawia flagę connection_aborted() w PHP na true i uzbraja łagodne przerwanie — przenośne pętle, które sprawdzają connection_aborted(), kończą się czysto poprzez swoją normalną ścieżkę zakończenia, natomiast pętle, które tego nie sprawdzają, i tak zostają zakończone przez niejawne przerwanie przy kolejnym opróżnieniu bufora.

Note

Utrzymuj małe ładunki zdarzeń, aby zachować płynną przepustowość. Duże ładunki mogą szybko zapełnić 64-porcjowy bufor, powodując blokowanie PHP przy każdym opróżnieniu bufora.

Przykłady

Podstawowy strumień SSE

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');

for ($i = 0; $i < 100; $i++) {
    $data = json_encode(['counter' => $i, 'time' => microtime(true)]);
    echo "id: {$i}\n";
    echo "event: tick\n";
    echo "data: {$data}\n\n";
    oxphp_stream_flush();

    sleep(1);

    // Send a comment heartbeat every 15 seconds to keep proxies from closing idle connections
    if ($i % 15 === 0) {
        echo ": heartbeat\n\n";
        oxphp_stream_flush();
    }
}

Sprawdzanie stanu strumieniowania

Użyj oxphp_is_streaming(), aby sprawdzić, czy bieżące żądanie jest już w trybie strumieniowania. Przydaje się to w middleware lub we współdzielonych obsługach żądań:

php

<?php
if (!oxphp_is_streaming()) {
    header('Content-Type: text/event-stream');
    header('Cache-Control: no-cache');
}

echo "data: {\"status\": \"connected\"}\n\n";
oxphp_stream_flush();

Wykrywanie rozłączeń klienta

Długo żyjące pętle SSE powinny sprawdzać connection_aborted(), aby czysto się zakończyć, gdy klient zamknie połączenie. Odpowiada to standardowemu idiomowi PHP / php-fpm i pozwala skryptowi wykonać logikę sprzątania (zamknięcie uchwytów bazy danych, zwolnienie blokad, dokończenie bloków finally) przed zakończeniem:

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

$db = new PDO(/* ... */);

try {
    while (!connection_aborted()) {
        echo "data: " . json_encode(['ts' => time()]) . "\n\n";
        oxphp_stream_flush();
        sleep(1);
    }
} finally {
    $db = null; // runs on normal exit AND on connection_aborted exit
}

Note

Jeśli skrypt nigdy nie sprawdza connection_aborted(), OxPHP i tak zakończy go poprzez niejawne przerwanie przy następnym opróżnieniu bufora po rozłączeniu klienta, ale bloki finally znajdujące się na ścieżkach kodu, które omijają wywołanie opróżniające bufor, mogą się nie wykonać. W przypadku kodu przechowującego zewnętrzne zasoby lepiej zastosować jawne sprawdzenie.

Korzystanie z natywnej funkcji flush()

Natywna funkcja flush() z PHP również nadaje się do strumieniowania, ale najpierw wymaga wyczyszczenia wszystkich warstw bufora wyjściowego. Lepiej używać oxphp_stream_flush() — automatycznie zarządza buforami wyjściowymi i integruje się z systemem przeciwciśnienia OxPHP.

php

<?php
header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');

while (ob_get_level()) {
    ob_end_clean();
}

for ($i = 0; $i < 100; $i++) {
    echo "data: " . json_encode(['counter' => $i]) . "\n\n";
    flush();
    sleep(1);
}

SSE w trybie worker

SSE działa zarówno w trybie standardowym, jak i w trybie worker. W trybie worker połączenie strumieniowe zajmuje workera przez cały czas trwania strumienia. Worker obsłuży kolejne żądanie dopiero po zakończeniu skryptu.

php

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

$redis = new Redis();
$redis->pconnect('redis', 6379);

oxphp_worker(function () use ($redis) {
    if (($_SERVER['HTTP_ACCEPT'] ?? '') !== 'text/event-stream') {
        http_response_code(400);
        echo json_encode(['error' => 'SSE only']);
        return;
    }

    header('Content-Type: text/event-stream');
    header('Cache-Control: no-cache');

    while (true) {
        $message = $redis->brPop('events', 25);
        if ($message) {
            echo "data: {$message[1]}\n\n";
        } else {
            // No message within timeout — send heartbeat to keep the connection alive
            echo ": heartbeat\n\n";
        }
        oxphp_stream_flush();
    }
});

Rozwiązywanie problemów

Klient nie otrzymuje żadnych danych, dopóki skrypt się nie zakończy

Bufor wyjściowy PHP przechwytuje dane wyjściowe zamiast je strumieniować. Dzieje się tak, gdy warstwy OB są aktywne, a oxphp_stream_flush() nie jest wywoływane.

Rozwiązanie: Wywołuj oxphp_stream_flush() po każdym zdarzeniu. Ta funkcja opróżnia wszystkie warstwy bufora wyjściowego PHP i wysyła nagromadzone dane wyjściowe jako porcję.

Połączenia SSE są zamykane po kilku minutach

Zadziałał max_execution_time z PHP i zakończył skrypt. Strumienie SSE muszą działać dłużej niż skonfigurowany limit.

Rozwiązanie: Wyłącz licznik czasu wykonania dla danego żądania na początku skryptu strumieniującego:

php

set_time_limit(0);

Jest to preferowane względem globalnego ustawienia max_execution_time = 0 — pozostawia limit w mocy dla endpointów innych niż SSE. Alternatywnie, jeśli cała instancja jest przeznaczona do długo żyjących strumieni:

php.ini

; php.ini
max_execution_time = 0

Pośrednie proxy zamykają bezczynne połączenia SSE

Load balancery i proxy często zamykają połączenia, przez które nie przepływają żadne dane przez 30–60 sekund.

Rozwiązanie: Wysyłaj heartbeat w formie komentarza w regularnych odstępach czasu, aby utrzymać aktywne połączenie:

php

echo ": heartbeat\n\n";
oxphp_stream_flush();

oxphp_stream_flush() zwraca false

oxphp_finish_request() zostało wywołane wcześniej w tym samym żądaniu. Po zakończeniu odpowiedzi strumieniowanie nie jest możliwe. Sprawdź w swoim kodzie, czy nie ma przypadkowych wywołań oxphp_finish_request() przed rozpoczęciem strumieniowania.

Przykład z Dockerem

Endpointy SSE wymagają wyłączenia licznika czasu wykonania PHP lub ustawienia go na wysoką wartość. Każde aktywne połączenie SSE zajmuje jednego workera PHP przez cały czas trwania strumienia, więc dobierz rozmiar puli workerów tak, aby pomieściła oczekiwaną liczbę równoczesnych strumieni.

compose.yaml

services:
  app:
    image: ghcr.io/oxphp/oxphp:0.10.0
    ports:
      - "8080:8080"
    volumes:
      - ./src:/var/www/html
    environment:
      DOCUMENT_ROOT: "/var/www/html/public"
      ENTRY_FILE: "index.php"
      PHP_WORKERS: "32"

Każdy skrypt strumieniujący powinien wywołać set_time_limit(0) na początku, aby licznik czasu dla danego żądania nie zadziałał w trakcie strumienia. Dzięki temu globalny max_execution_time pozostaje w mocy dla żądań innych niż SSE.

Dobre praktyki

Używaj oxphp_stream_flush() zamiast natywnej funkcji flush(), aby uzyskać automatyczne zarządzanie buforem wyjściowym i integrację z przeciwciśnieniem.
Wysyłaj okresowe heartbeaty w formie komentarza (: heartbeat\n\n) co 20–30 sekund, aby pośrednie proxy nie zamykały bezczynnych połączeń i aby wcześnie wykrywać rozłączenia klientów.
Utrzymuj małe ładunki zdarzeń. Duże ładunki szybciej zapełniają 64-porcjowy bufor, powodując zacinanie się PHP przy każdym opróżnieniu bufora. W przypadku dużych danych wyślij identyfikator zdarzenia i pozwól klientowi pobrać pełny ładunek za pomocą osobnego żądania.
Wyłącz licznik czasu wykonania PHP dla poszczególnych skryptów za pomocą set_time_limit(0) w przypadku długo żyjących endpointów SSE lub ustaw max_execution_time na tyle wysoko, aby pokryć najdłuższy oczekiwany czas trwania strumienia.
Dobierz rozmiar puli workerów do szczytowej liczby równoczesnych strumieni. Każde aktywne połączenie SSE zajmuje jednego workera PHP przez cały czas swojego trwania. Zaplanuj co najmniej jednego workera na każdego oczekiwanego równoczesnego klienta oraz dodatkowych workerów na zwykłe żądania inne niż SSE.

Uwagi

Kompresja Brotli jest automatycznie pomijana dla odpowiedzi strumieniowych. Kompresja dotyczy wyłącznie odpowiedzi w pełni zbuforowanych.
oxphp_stream_flush() zwraca false, jeśli oxphp_finish_request() zostało już wywołane w tym samym żądaniu.
W trybie worker worker pozostaje zajęty przez cały czas trwania strumienia i obsługuje kolejne żądanie dopiero po zakończeniu skryptu PHP.

Zachowanie przy zamknięciu

Gdy serwer otrzyma SIGTERM (kroczące wdrożenie, docker stop, eksmisja poda w Kubernetes), łagodnie kończy pracę, wygaszając połączenia:

Każdy otwarty strumień SSE jest czysto kończony przy swoim następnym flush — jego handler przerywa działanie tak, jakby połączenie zostało zamknięte, więc funkcje zarejestrowane przez register_shutdown_function() nadal się uruchamiają, a error_get_last()['message'] zawiera Request cancelled (shutdown).
Klienci HTTP/2 otrzymują ramkę GOAWAY; połączenia keep-alive HTTP/1.1 są zamykane. EventSource w przeglądarce łączy się ponownie automatycznie — z działającą instancją, gdy przed flotą stoi load balancer.
Strumień nie otrzymuje odpowiedzi 503: jego nagłówki 200 zostały już wysłane, więc statusu nie da się nadpisać. Projektuj klientów tak, aby wznawiali od ostatniego identyfikatora zdarzenia (wysyłaj id: z każdym zdarzeniem; przy ponownym połączeniu przeglądarka odsyła go z powrotem w nagłówku żądania Last-Event-ID, dostępnym w PHP jako $_SERVER['HTTP_LAST_EVENT_ID']).

Zwykłe (niestrumieniowe) żądania będące w toku w momencie SIGTERM nie są przerywane: mają całe okno wygaszania na normalne zakończenie, a anulowane zostają tylko te żądania, które nadal działają po upływie DRAIN_TIMEOUT_SECONDS (domyślnie 25), przy czym dostają jeszcze ~2 sekundy na dokończenie. Ustaw okres karencji przy zakończeniu (termination grace period) orkiestratora powyżej DRAIN_TIMEOUT_SECONDS + 2.

„Strumieniowanie" oznacza tu każdą odpowiedź, która już opróżniła fragmentowane dane wyjściowe — serwer nie potrafi odróżnić skończonego pobierania strumieniowego od nieskończonego strumienia zdarzeń, więc duże opróżnione pobieranie będące w toku w momencie SIGTERM również zostaje zakończone wcześniej, nie tylko SSE. Żądanie, które wywołało oxphp_finish_request() przed SIGTERM, liczy się jako zwykłe: jego odpowiedź jest kompletna, a pozostała praca w tle otrzymuje okno wygaszania.

Handler zablokowany wewnątrz jednego długo trwającego wywołania natywnego (natywny sleep(), kosztowny preg_match, blokujące zapytanie do bazy danych) nie może zostać przerwany, dopóki to wywołanie nie zwróci sterowania. W trybie worker w pętlach strumieniujących preferuj kooperacyjny oxphp_sleep() — oddaje sterowanie planiście Fiberów i zostaje wybudzony natychmiast przy zamknięciu. Poza trybem worker oxphp_sleep() przechodzi na zwykły blokujący sen, więc strumień zareaguje na zamknięcie przy swoim następnym flush po powrocie z uśpienia.

Zobacz także

Tryb worker — trwałe procesy PHP zmniejszające narzut na inicjalizację
Limity czasu — konfigurowanie lub wyłączanie limitu czasu żądania dla długo żyjących połączeń
Funkcje PHP — pełna dokumentacja oxphp_stream_flush() i oxphp_is_streaming()
Kompresja — działanie kompresji Brotli i to, które odpowiedzi są kompresowane