README

面向 Hyperf 3.1 的 Server-Sent Events（SSE）组件，提供统一的消息协议、流式响应、用户定向通知、广播通知，以及可切换的 Redis Stream / RabbitMQ 驱动。

包名：godjarvis/hyperf-sse

GitHub：GodJarvis/hyperf-sse

PHP：>= 8.1

Hyperf：~3.1.0

1. 设计定位

该组件只负责可复用的 SSE 基础设施：

业务任务先提交最终状态到数据库
              │
              ▼
SseService::notify(subject, event, payload)
              │
     ┌────────┴────────┐
     ▼                 ▼
Redis Stream       RabbitMQ Topic
可通过 ID 回放       仅实时投递
     └────────┬────────┘
              ▼
SseDriverInterface::subscribe()
              ▼
SseEmitter（text/event-stream）
              ▼
浏览器 EventSource

组件不内置 Controller、路由或用户认证。应用必须从服务端可信的登录上下文解析 subject（用户 ID、租户用户 ID、页面会话 ID 等），不能信任客户端直接提交的用户 ID。

对于异步任务通知，推荐继续保留数据库作为事实来源：

Worker 先提交任务最终状态；
再通过本组件发布轻量通知；
页面初始化及 SSE 重连后调用任务查询接口校准状态；
Redis Stream 可按 Last-Event-ID 补发保留范围内的事件，但仍不替代业务数据库。

2. 安装

2.1 Packagist 发布后

composer require godjarvis/hyperf-sse:^1.0

Hyperf 会通过 composer.json > extra.hyperf.config 自动发现 ConfigProvider，并注册：

GodJarvis\HyperfSse\Contract\SseDriverInterface
    => GodJarvis\HyperfSse\Driver\SseDriverFactory

发布配置：

php bin/hyperf.php vendor:publish godjarvis/hyperf-sse

生成：

config/autoload/sse.php

2.2 Packagist 发布前本地联调

项目与组件目录相邻时：

{
  "repositories": [
    {
      "type": "path",
      "url": "../hyperf-sse",
      "options": {
        "symlink": false
      }
    }
  ],
  "require": {
    "godjarvis/hyperf-sse": "dev-main"
  }
}

然后执行：

composer update godjarvis/hyperf-sse -W

3. Redis Stream 驱动（默认）

默认配置：

SSE_DRIVER=redis
SSE_REDIS_POOL=default
SSE_HEARTBEAT_INTERVAL=15
SSE_MAX_IDLE_TIME=7200

特性：

每个 subject 一个私有 Stream，同时订阅一个广播 Stream；
使用近似 MAXLEN 控制消息数量；
SSE id 同时记录广播与私有 Stream 的消费位置；
浏览器重连时可把 Last-Event-ID 传给 subscribe()；
新连接从建连时刻的最新位置开始，不回放历史消息；
断线重连仅能回放 Redis 中尚未被裁剪的消息。

默认 Key：

hyperf:sse:stream:broadcast
hyperf:sse:stream:subject:{urlencoded-subject}

如果应用使用独立 Redis 连接池，先在 config/autoload/redis.php 中配置对应池，再设置 SSE_REDIS_POOL。

4. RabbitMQ 驱动（可选）

先安装 Hyperf AMQP：

composer require hyperf/amqp:~3.1.0

配置：

SSE_DRIVER=rabbitmq
SSE_RABBITMQ_POOL=default
SSE_RABBITMQ_EXCHANGE=hyperf_sse_notify
SSE_RABBITMQ_SUBJECT_PREFIX=hyperf.sse.subject.
SSE_RABBITMQ_BROADCAST_KEY=hyperf.sse.broadcast

RabbitMQ 驱动会为每条 SSE 连接创建一个独占、自动删除的临时队列，并绑定：

当前 subject 的 routing key；
广播 routing key。

边界：

该驱动是实时通知，连接不存在时消息不会保留；
Last-Event-ID 对 RabbitMQ 驱动无回放作用；
每条 SSE 连接会占用 RabbitMQ Channel，高并发连接前必须评估连接/Channel 上限；
大规模场景优先使用 Redis Stream，或自行实现每 Worker 共享订阅器的驱动。

5. Controller 接入

完整示例见 examples/SseController.php。核心写法：

use GodJarvis\HyperfSse\SseEmitter;
use GodJarvis\HyperfSse\SseService;
use Hyperf\Context\RequestContext;

public function connect(SseService $sse, SseEmitter $emitter): void
{
    // 必须从服务端认证上下文获取，不要直接使用 query 中的 user_id。
    $subject = (string) $this->request->getAttribute('authenticated_user_id');
    $lastEventId = RequestContext::get()?->getHeaderLine('Last-Event-ID') ?: null;

    $emitter->emit($sse->subscribe($subject, $lastEventId));
}

SseEmitter::emit() 会直接持有并结束底层 HTTP 流。调用后 Controller 必须结束执行，推荐声明返回类型为 void，不要再返回普通响应内容。

默认响应头：

Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache, no-transform
Connection: keep-alive
X-Accel-Buffering: no
Transfer-Encoding: chunked

6. 发布通知

定向通知：

$sse->notify(
    subject: $userId,
    event: 'task_completed',
    data: [
        'task_id' => $taskId,
        'status' => 'completed',
        'result_url' => $resultUrl,
    ],
);

广播：

$sse->broadcast('maintenance', [
    'message' => '系统将在 10 分钟后维护',
]);

建议只推送任务 ID、状态和结果地址等小消息，不通过 SSE 发送大文件或大型任务结果。

7. 前端 EventSource

const source = new EventSource('/sse/connect', { withCredentials: true });

source.addEventListener('task_completed', (event) => {
  const payload = JSON.parse(event.data);
  console.log(payload.task_id, payload.status);
});

source.addEventListener('server_close', () => {
  source.close();
});

source.onerror = () => {
  // EventSource 会按浏览器策略自动重连；重连后仍应查询任务接口校准状态。
};

原生 EventSource 不能自由设置 Authorization Header：

Cookie 登录：使用 withCredentials；
Bearer Token：建议使用短期一次性连接票据，或改用 Fetch Streaming；
不要把长期访问令牌直接放进 URL。

8. Nginx / Ingress

Nginx 示例：

location /sse/ {
    proxy_pass http://hyperf_upstream;
    proxy_http_version 1.1;

    proxy_set_header Connection '';
    proxy_buffering off;
    proxy_cache off;
    gzip off;

    proxy_read_timeout 2h;
    proxy_send_timeout 2h;
}

还需要检查外层 Ingress、API 网关、CDN 或负载均衡器的响应缓冲、压缩和空闲超时。应用层心跳间隔必须小于链路中最短的空闲超时。

9. 自定义驱动

实现接口：

use GodJarvis\HyperfSse\Contract\SseDriverInterface;

final class CustomSseDriver implements SseDriverInterface
{
    // subscribe / notify / broadcast
}

在 config/autoload/sse.php 注册：

'driver' => 'custom',
'drivers' => [
    'custom' => CustomSseDriver::class,
],

subscribe() 必须返回 Generator<SseMessage>；心跳使用：

yield SseMessage::heartbeat();

10. 从当前项目迁移

原项目概念与组件 API 对照：

原实现	组件实现
`App\Service\SseService`	`GodJarvis\HyperfSse\SseService`
`App\Service\Sse\SseDriverInterface`	`GodJarvis\HyperfSse\Contract\SseDriverInterface`
`App\Service\Sse\SseRedisDriver`	`GodJarvis\HyperfSse\Driver\RedisStreamDriver`
`App\Service\Sse\SseRabbitMqDriver`	`GodJarvis\HyperfSse\Driver\RabbitMqDriver`
`SseResponseFormatTrait::sseReturn()`	`SseEmitter::emit()`
`SseEvent` 枚举	普通事件字符串，由业务项目自行定义枚举
`uid`	`string

迁移时保留业务项目自己的：

Controller 路由；
登录态、CSRF 与权限校验；
业务事件枚举；
任务状态查询与重连校准接口；
Nginx / Ingress 配置。

11. 开发与验证

composer install
composer test
composer analyse
composer cs-check
composer validate --strict

当前测试覆盖消息格式、心跳、服务委托、驱动工厂、ConfigProvider 与 AMQP 拓扑元数据。Redis/RabbitMQ 真连接、浏览器断线、多 Worker/多实例及代理链路需要在使用方环境执行集成测试和压测。

12. 发布到 Composer

仅推送 GitHub 仓库还不足以让所有项目直接执行标准 composer require。首次发布还需要：

将代码推送到 GodJarvis/hyperf-sse；
在 Packagist 提交仓库；
配置 GitHub Webhook 或 Packagist GitHub Service；
创建语义化版本标签，例如：

git tag -a v1.0.0 -m "Release v1.0.0"
git push origin main --tags

发布后用一个全新目录验证：

composer require godjarvis/hyperf-sse:^1.0

godjarvis / hyperf-sse

Maintainers

Package info

Statistics

Security