服务器发送事件（SSE）

OxPHP 通过服务器发送事件（Server-Sent Events）协议向客户端流式推送实时数据，并内置背压机制。只需在 PHP 脚本中设置 Content-Type: text/event-stream 并调用 oxphp_stream_flush()，其余的交给 OxPHP 处理。

工作原理

一个流式响应会经历一个可预测的生命周期：

设置响应头。 你的 PHP 脚本通过 header() 设置 Content-Type: text/event-stream，并使用 echo 写入符合 SSE 格式的文本行。
首次 flush 打开流。 第一次调用 oxphp_stream_flush() 会把 HTTP 响应头发送给客户端并进入流式模式。客户端连接保持打开。
每次 flush 发送一个数据块。 之后每次调用 oxphp_stream_flush() 都会把缓冲的输出作为一个新的数据块刷出，立即投递给客户端。
缓冲区填满时触发背压。 OxPHP 在 PHP 工作进程与客户端之间维护一个最多 64 个数据块的内部缓冲区。当缓冲区已满时——因为慢速客户端尚未消费掉先前的数据块——oxphp_stream_flush() 会阻塞，直到腾出空间为止。这可以防止内存无限增长。
退出或断开时清理。 当 PHP 脚本执行完毕时，OxPHP 会优雅地关闭连接。如果客户端在流中途断开，脚本会在 OxPHP 察觉到连接关闭之后的第一次 flush 时得知这一点。OxPHP 会设置 PHP 的 connection_aborted() 标志，并且除非脚本调用过 ignore_user_abort(true)，否则会在这次 flush 内部停止脚本：该调用不会返回，这正是 PHP 在 PHP-FPM 下停止一个输出已无法写出的脚本的方式。调用过 ignore_user_abort(true) 的脚本则会从这次 flush 正常返回并继续运行，而检查 connection_aborted() 才是结束其循环的方式（参见检测客户端断开）。

Note

保持事件负载（请求体）较小，以维持平滑的吞吐量。较大的负载会很快填满这个 64 块的缓冲区，导致 PHP 在每次 flush 时阻塞。

示例

基础 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();
    }
}

检查流式状态

使用 oxphp_is_streaming() 来检查当前请求是否已经处于流式模式。这在中间件或共享的请求处理器中很有用：

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

检测客户端断开

已经发出响应头的流，只有在 flush 时才会得知其客户端已经离开。默认情况下，脚本就停在这次 flush 处：oxphp_stream_flush()（或 flush()）不会返回，因此 while (!connection_aborted()) 条件永远不会再被求值，而这次 flush 所处的每个 finally 块都会被跳过。PHP-FPM 也以同样的方式停止输出已无法写出的脚本。

仍会运行的是请求的收尾部分：register_shutdown_function() 回调（在其中 connection_aborted() 返回 1），以及随请求一起销毁的对象的析构函数。PHP-FPM 会丢弃它们写出的任何内容，而 OxPHP 不同：它会在回调发送第一次输出时停止该回调，并跳过在它之后注册的回调。析构函数仍会运行，其中第一个发送输出的析构函数也会以同样的方式被停止，排在它之后的析构函数则会被跳过。保存在输出缓冲区中的输出（用 ob_start() 打开的缓冲区，或由 php.ini 中的 output_buffering 打开的缓冲区）只有在该缓冲区被刷出时才会发送。请让这些清理逻辑不产生任何输出，或者像下面这样调用 ignore_user_abort(true)。

若要在脚本自身中完成清理（在 finally 块中或循环之后），请调用 ignore_user_abort(true)，并让 connection_aborted() 来结束循环。这样，发现客户端已离开的那次 flush 会正常返回，从那一刻起 connection_aborted() 返回 1，循环经由其条件退出：

php

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

// A stream outlives max_execution_time, which is 30 s unless the PHP configuration sets another value
set_time_limit(0);

// Keep running when the client leaves, so that the loop below sees it and ends
ignore_user_abort(true);

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

try {
    while (!connection_aborted()) {
        echo "data: " . json_encode(['ts' => time()]) . "\n\n";
        oxphp_stream_flush();
        sleep(1);
    }
} finally {
    $db = null; // also runs when the loop ends on a disconnect
}

`ignore_user_abort(true)` 会让流继续运行

调用了该函数后，connection_aborted() 检查就是结束循环的唯一途径：客户端断开后脚本会继续运行，它写出的内容会被丢弃，而 connection_aborted() 会报告断开。这与在其他任何 SAPI 下的约定相同。这样的循环需要自己的边界：一个要求比客户端活得更久的流，如果既不检查 connection_aborted()，也不会自行结束，就会一直占用其工作进程，直到服务器关闭。

该调用只作用于发起它的那个请求。不过，请求起始时的取值同样算数：php.ini 中的 ignore_user_abort = On，或者工作进程脚本引导阶段（其请求循环之前的代码，一些工作进程示例就是以它开头的）里的 ignore_user_abort(true)，都是每个请求的起始基线，因此在那里提供的每个流都会表现得像已经调用过它一样，需要自己的 connection_aborted() 检查。

使用原生 flush()

PHP 的原生 flush() 也可用于流式传输，但需要先清空所有输出缓冲层。请优先使用 oxphp_stream_flush()——它会自动管理输出缓冲，并与 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

SSE 在标准模式和工作进程模式下都能工作。在工作进程模式下，只有当流的脚本正在运行，或在不挂起其纤程的情况下等待时（例如 64 块缓冲区已满时的 flush），流才会占用其工作线程。当流在 oxphp_sleep() 中等待，或在经由 streams 运行时钩子的读取上等待时，该线程会去处理其他请求；一个工作线程最多可同时运行 256 个请求（参见纤程多路复用）。

在广播场景中，每个打开的流都必须收到每一个事件，因此该示例从 Redis Stream 中读取事件，而不是从列表中弹出。读取 stream 会把条目留在原处，且每个读取者各自维护自己的读取位置，因此每个客户端都能读到 stream 中仍保留的每一个事件。列表上的 BRPOP 则恰好相反：它会移除所返回的元素，而该元素只会交给在该列表上等待的连接中的恰好一个。这适合工作队列，却不适合广播：如果有两个流处于打开状态，每个事件只会到达其中一个。

php

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

oxphp_worker(function () {
    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');

    // A stream outlives max_execution_time, which is 30 s unless the PHP configuration sets another value
    set_time_limit(0);

    // One connection per stream: the blocking read below holds it for as long as the
    // stream lives. The last argument is the read timeout, which has to exceed the
    // 25 s the read blocks for.
    $redis = new Redis();
    $redis->connect('redis', 6379, 2.0, null, 0, 30.0);

    // The newest entry in the stream. It also finds out whether the key holds a stream at all,
    // while the answer can still be an error status: once the first frame is out the response is
    // a 200, and a browser reconnects after every close of one.
    $tail = $redis->xRevRange('events', '+', '-', 1);
    if ($tail === false) {
        error_log('SSE: ' . $redis->getLastError());
        http_response_code(500);
        return;
    }

    // Resume after the last event the browser saw (it sends the id back in Last-Event-ID).
    // A first connection, or a header that is not a stream id, starts after the newest entry.
    $last = $_SERVER['HTTP_LAST_EVENT_ID'] ?? '';
    if (!preg_match('/^\d{1,19}-\d{1,19}$/', $last)) {
        $last = $tail ? array_key_first($tail) : '0-0';
    }

    // Send the headers now; a quiet stream would otherwise send none until its first read returns.
    // This named event carries the starting id, which the browser takes as its Last-Event-ID before
    // the first event arrives; onmessage does not fire for it.
    echo "event: ready\nid: {$last}\ndata:\n\n";
    oxphp_stream_flush();

    while (true) {
        // Reading leaves the entries in the stream, so every open stream gets every one
        $read = $redis->xRead(['events' => $last], 100, 25000);
        if ($read === false) {
            // Redis refused the read, the key having been replaced by a non-stream, say. End the stream;
            // the browser reconnects and the check above answers it with an error status
            return;
        }
        foreach ($read['events'] ?? [] as $id => $fields) {
            echo "id: {$id}\ndata: {$fields['data']}\n\n";
            $last = $id;
        }
        if (!$read) {
            // No entry within 25 s — send heartbeat to keep the connection alive
            echo ": heartbeat\n\n";
        }
        oxphp_stream_flush();
    }
});

从任意 PHP 进程（请求处理器、队列工作进程、CLI 脚本）发布事件：

php

$redis->xAdd('events', '*', ['data' => json_encode($payload)], 1000, true);

最后两个参数把 stream 的长度上限设为大约 1000 个条目（MAXLEN ~ 1000）。请以 data 发送 JSON 或其他任何不含换行符的内容：示例会把它写在单独一行 data: 上。

该示例依赖于以下几点：

条目 id 就是事件 id。 像 1790976033892-0 这样的 id 会作为事件的 id: 发出，浏览器在重连时会通过 Last-Event-ID 回传它看到的最后一个 id，流则从它之后继续（参见关闭时的行为）。
第一帧是一个携带起始 id 的事件。 Safari 和 Firefox 只会从带有 data: 行的帧中把 id: 作为浏览器的 Last-Event-ID，因此单独一个 id: 会让在第一个事件到来之前就被断开的浏览器没有 id 可发送：它会重连，从最新条目开始，并错过这期间发布的所有内容。带有空 data: 行的 ready 事件为它提供了一个 id。onmessage 不会为具名事件触发，而 addEventListener('ready', …) 会；把所有事件交给同一个回调的客户端库需要跳过它。
无法启动的流会在第一帧之前得到错误状态码。 Redis 拒绝读取存放着非 stream 内容的键（比如之前的队列遗留下来的列表），而任何 id 都无法让这次读取成功，因此示例用它的第一条命令发现这一点，记录 Redis 给出的消息，并以 500 应答。这里必须返回状态码，而不是让流直接结束：一旦第一帧发出，响应就是 200，而 200 流每次关闭后浏览器都会重连，只要页面保持打开，每隔几秒就重连一次，每次都会建立新的 Redis 连接，日志里却什么也没有。EventSource 在收到错误状态码后不会重试；页面会收到一个 error 事件，此时 readyState 为 EventSource.CLOSED，如果还需要，页面必须自己打开一个新的 EventSource。无法连接到 Redis 时也是如此：connect() 抛出异常，响应为 500，EventSource 就此永久关闭。流开始之后被 Redis 拒绝的读取（比如键被替换成了非 stream）会通过 return 结束该流，而重连会经过上面的检查。如果没有这个 return，循环会把 false 当作超时，一个接一个地只发送心跳。
会检查 Last-Event-ID 的格式。 该请求头来自客户端，可能包含任何内容，而 Redis 会拒绝不是 stream id 的值，例如 abc。示例把这样的值当作缺失的请求头处理，像首次连接那样从最新条目之后开始。该模式允许短横线两侧各最多 19 位数字，这使它接受的每个值都处于 Redis 能读取的范围之内。
新的流只读取一次最新 id。 它不会在每次调用时都传入 $：Redis 会在每次调用的那一刻把 $ 解析为最新的 id，因此两次调用之间添加的条目会被跳过。
历史是有界的。 如果读取者的最后一个事件已被裁剪掉（浏览器曾断开连接，或者流被慢速客户端拖住，而这期间发布的条目超过了上限），它会从仍然保留的最旧条目继续。中间被裁剪掉的条目不会被投递，也不会有任何提示。请把上限设得足以覆盖你希望流能够从中恢复的最长中断。
每个流都有自己的连接，其读取超时大于阻塞时间。 如果不传超时参数，phpredis 会回退到 default_socket_timeout，阻塞时间超过它的读取会以 RedisException 失败。不要为每个工作进程只打开一个连接再交给所有流共用：启用 streams 运行时钩子时，当某个流在读取上等待，工作进程会去运行其他请求，而使用过某个连接的请求会一直持有它，直到该请求结束。到达同一连接的第二个流会等待第一个流。如果等待到时的时候第一个流仍然打开（该上限来自 max_execution_time 和 default_socket_timeout；参见钩子一节），phpredis 会抛出 RedisException，第二个流随之失败。
脚本关闭了 PHP 的执行计时器。 除非 PHP 配置设置了其他值，否则 max_execution_time 为 30 秒，而发布镜像不附带 php.ini；流本来就要比它活得更久。如果计时器保持开启，脚本会以致命错误结束：计时器在 30 秒后触发，并在正在进行的读取返回时生效，因此在这里 25 秒一次的读取下，一个安静的流会在第 50 秒结束。在工作进程模式下，该调用只关闭这个流自身请求的限制，不影响其他请求；参见工作进程模式 → 请求截止时间。
已关闭连接的客户端会在下一次 flush 时被发现。 在此之前，它的请求以及随之的 Redis 连接都保持打开：这里最多 25 秒，即阻塞时间，这同时也是心跳间隔。

Redis Pub/Sub 同样会把每条消息投递给每个已连接的订阅者，但它不保留任何内容：客户端重连期间发布的消息会丢失，也没有可供恢复的 id。只在错过事件无关紧要的场景中使用它。

负载均衡器后的多个实例

一个流就是到某个实例的一条长连接，该实例由负载均衡器一次性选定。当有多个实例时：

事件必须跨实例传递。 OxPHP\Shared\* 对象只存在于单个进程内部（参见共享状态），因此一个实例上的流无法通过它们看到在另一个实例上发布的事件。两者必须读取同一个外部存储，上面的 Redis Stream 正是这样的存储。
容量按实例计算。 每个实例能服务多少个流，取决于它自己的工作进程池留出的余量（参见最佳实践）。优先选择会统计打开连接数的均衡方式（例如 nginx 中的 least_conn），而不是轮询（round-robin）：轮询按到达顺序分配连接，而不考虑连接会保持打开多久，而一个流保持打开的时间远比普通请求长。
不需要粘性会话。 重连的浏览器可能会落到另一个实例上，只要第一帧和每个事件都带有 id:，并且每个实例都读取同一个 stream，这就没有问题。
代理不得缓冲流。 进行缓冲的代理会把事件扣住，直到其缓冲区填满。在 nginx 中，可以用 proxy_buffering off 为该流的 location 关闭缓冲，或者从脚本中发送 X-Accel-Buffering: no，除非 proxy_ignore_headers 另有设置，否则 nginx 会遵从它。
心跳必须比负载均衡器的空闲超时更频繁。 参见故障排查中的“中间代理关闭空闲的 SSE 连接”。
在浏览器与负载均衡器之间使用 HTTP/2。 在 HTTP/1.1 下，浏览器对每个主机在所有标签页中总共只允许大约六个打开的连接，而每个流都要占用一个；HTTP/2 则可以在单个连接上承载多个流。

故障排查

脚本结束前客户端收不到任何数据

PHP 的输出缓冲正在捕获输出，而不是将其流式发出。这发生在 OB 层处于激活状态且未调用 oxphp_stream_flush() 时。

修复方法： 在每个事件之后调用 oxphp_stream_flush()。该函数会刷出所有 PHP 输出缓冲层，并把累积的输出作为一个数据块发送。

SSE 连接在几分钟后被关闭

PHP 的 max_execution_time 触发并终止了脚本。SSE 流的运行时间必须超过所配置的限制。

修复方法： 在流式脚本的顶部禁用按请求计算的执行计时器：

php

set_time_limit(0);

这优于全局设置 max_execution_time = 0——它会为非 SSE 端点保留该限制。或者，如果整个实例都专门用于长时间运行的流：

php.ini

; php.ini
max_execution_time = 0

中间代理关闭空闲的 SSE 连接

负载均衡器和代理通常会关闭 30–60 秒内没有数据传输的连接。

修复方法： 定期发送一个注释心跳，以保持连接活跃：

php

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

oxphp_stream_flush() 返回 false

在同一个请求中较早地调用了 oxphp_finish_request()。一旦响应已经结束，就无法再进行流式传输。请检查你的代码，看是否在流式传输开始之前无意中调用了 oxphp_finish_request()。

Docker 示例

SSE 端点要求禁用 PHP 的执行计时器，或将其设置得较高。在工作进程模式之外（如本示例），每个活跃的 SSE 连接都会在整个流的持续时间内占用一个 PHP 工作进程，因此要根据你预期的并发流数量来调整工作进程池的大小。

compose.yaml

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

每个流式脚本都应在顶部调用 set_time_limit(0)，这样按请求计算的计时器就不会在流中途触发。这能让全局的 max_execution_time 对非 SSE 请求继续生效，在工作进程模式下也是如此：在那里每个请求都有自己的限制，因此一个流关闭了自己的限制后，与它一同被处理的其他请求仍然受到约束（参见工作进程模式 → 请求截止时间）。

最佳实践

使用 oxphp_stream_flush() 而非原生 flush()，以获得自动的输出缓冲管理和背压集成。
每 20–30 秒发送一次周期性的注释心跳（: heartbeat\n\n），以防止中间代理关闭空闲连接，并尽早检测客户端断开。
保持事件负载（请求体）较小。 较大的负载会更快填满这个 64 块的缓冲区，导致 PHP 在每次 flush 时停顿。对于大数据，应发送一个事件 ID，让客户端通过单独的请求去获取完整负载。
为长时间运行的 SSE 端点按脚本禁用 PHP 的执行计时器，使用 set_time_limit(0)，或者把 max_execution_time 设置得足够高，以覆盖你预期的最长流持续时间。
根据并发流的峰值来调整工作进程池的大小。 在工作进程模式之外，以及在工作进程模式下对于等待时不挂起其纤程的流而言，每个活跃的 SSE 连接在其整个持续时间内都会占用一个 PHP 工作进程（参见工作进程模式下的 SSE）。在这些情况下，至少要为每个预期的并发客户端预留一个工作进程，另外再为常规的非 SSE 请求预留额外的工作进程。

注意事项

事件流永远不会被压缩，这由两条相互独立的规则保证：text/event-stream 不是可压缩的媒体类型；而在响应头发出时长度未知的响应，无论客户端接受什么编码，都会不经压缩直接透传。这一点对 OxPHP 提供的每种编码（Brotli、Zstandard 和 gzip）都成立，因此任何配置都无法为流开启压缩。
如果在同一个请求上已经调用过 oxphp_finish_request()，则 oxphp_stream_flush() 返回 false。
在工作进程模式下，只有当流正在运行，或在不挂起其纤程的情况下等待时，它才会占用其工作线程；当它在 oxphp_sleep() 中或在被挂钩的读取上等待时，该线程会处理其他请求。在其他模式下，工作进程会一直被占用，直到 PHP 脚本退出。

关闭时的行为

当服务器收到 SIGTERM 时（滚动部署、docker stop、Kubernetes pod 驱逐），它会优雅地进行排空：

每个打开的 SSE 流都会在其下一次 flush 时被干净地结束——其处理器会像连接已关闭那样退出，因此 register_shutdown_function() 回调仍会运行，且 error_get_last()['message'] 读取到 Request cancelled (shutdown)。
HTTP/2 客户端会收到一个 GOAWAY 帧；HTTP/1.1 keep-alive 连接会被关闭。浏览器的 EventSource 会自动重连——当负载均衡器位于集群前面时，会重连到一个健康的实例。
该流不会收到 503：它的 200 响应头已经发送，因此状态码无法被改写。应将客户端设计为从最后一个事件 id 处恢复（随每个事件发送 id:；重连时浏览器会将其作为 Last-Event-ID 请求头回传，在 PHP 中可通过 $_SERVER['HTTP_LAST_EVENT_ID'] 获取）。工作进程模式示例正是这样做的。

在 SIGTERM 时正在处理中的普通（非流式）请求不会被中断：它们会获得整个排空窗口来正常完成，只有那些在 DRAIN_TIMEOUT_SECONDS（默认 25）到期时仍在运行的请求才会被取消，另有约 2 秒的时间来收尾。请将编排器的终止宽限期设置为高于 DRAIN_TIMEOUT_SECONDS + 2，再加上请求在截止时刻可能正处于的最长阻塞调用时间（参见优雅关闭）。

这里所说的“流式”指的是任何已经刷出过分块输出的响应——服务器无法区分一个有限的流式下载和一个无限的事件流，因此在 SIGTERM 时正在处理中的大型已刷出下载也会被提前结束，而不仅仅是 SSE。一个在 SIGTERM 之前就调用过 oxphp_finish_request() 的请求会被视为普通请求：它的响应已经完成，其剩余的后台工作会获得排空窗口。

一个阻塞在单个长时间运行的原生调用内部的处理器（原生 sleep()、繁重的 preg_match、阻塞式数据库查询）在该调用返回之前无法被中断。在工作进程模式下，流式循环中应优先使用协作式的 oxphp_sleep()——它会让出给纤程调度器，并在关闭时立即被唤醒。在工作进程模式之外，oxphp_sleep() 会退化为常规的阻塞式 sleep，因此流会在 sleep 返回后的下一次 flush 时对关闭做出反应。

给你的智能体的提示词

这些提示词假定你的智能体已连接 OxPHP 文档 MCP 服务器——需要哪些页面它会自己去读。

prompt

在这个项目里用 OxPHP 建一个服务器发送事件（SSE）端点。通过 OxPHP 文档 MCP 服务器读取 `features/sse` 页面。
围绕这个仓库本来就必须推送的数据来写这个处理函数——心跳、客户端断开、从最后一个事件 id 处恢复——并告诉我按我预期的并发量需要多少个工作进程。

参见

工作进程模式 —— 持久化的 PHP 工作进程，用于降低启动开销
超时 —— 为长时间运行的连接配置或禁用请求超时
PHP 函数 —— oxphp_stream_flush() 和 oxphp_is_streaming() 的完整参考
压缩 —— 哪些响应会被压缩，以及使用哪种编码