README

面向 Phalcon 3.4 / PHP 7.3 的 Server-Sent Events（SSE）组件，提供统一的 SSE 消息格式、Phalcon DI 服务注册、用户定向通知、广播通知，以及可切换的 Redis Stream / RabbitMQ 驱动。

Composer 包名：godjarvis/phalcon-sse

GitHub：GodJarvis/phalcon-sse

目标运行环境：PHP >= 7.3 < 7.4、Phalcon ~3.4.0；当前实测 PHP 7.3.22、Phalcon 3.4.5

1. 设计定位

组件复刻了 godjarvis/hyperf-sse 的业务能力，但不是简单修改命名空间，因为两个框架的运行模型不同：

维度	Hyperf	Phalcon 3.4 参考项目
Web 运行模型	Swoole/协程常驻进程	Nginx + PHP-FPM
SSE 连接占用	协程连接	一个 FPM Worker
流式输出	Hyperf EventStream	Phalcon Response 发头 + 原生 `echo/flush`
组件注册	ConfigProvider 自动发现	`ServiceProviderInterface` 手动注册
PHP 语法	PHP 8.1+	PHP 7.3

推荐业务架构：

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

组件负责 SSE 基础设施，不内置：

Controller 和路由；
登录、CSRF、用户权限；
业务事件枚举；
任务数据库和任务查询接口；
Nginx/PHP-FPM 配置。

应用必须从服务端可信的认证上下文解析 subject，不能信任客户端直接传入的 user_id。

2. 安装

Packagist 发布后：

composer require godjarvis/phalcon-sse:^1.0

发布前本地联调：

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

然后执行：

composer update godjarvis/phalcon-sse -W

3. 注册 Phalcon 服务

在参考项目 /home/zw/www/project_demo/settings/servicesShare.php 中注册：

use GodJarvis\PhalconSse\SseServiceProvider;

$di->register(new SseServiceProvider());

注册后可通过 DI 获取：

$sse = $di->getShared('phalconSse');
$emitter = $di->getShared('phalconSse.emitter');
$driver = $di->getShared('phalconSse.driver');
$config = $di->getShared('phalconSse.config');

组件默认读取自身 config/sse.php，然后依次合并：

应用 $config->sse；
new SseServiceProvider($overrides) 传入的覆盖配置。

4. 应用配置

可以在项目的 config/config.php 中加入：

'sse' => [
    'driver' => 'redis',
    'logger_service' => 'sseLogger',
    'heartbeat_interval' => 15,
    'max_idle_time' => 7200,
    'redis' => [
        // 直接复用 project_demo 已注册的 redis 服务。
        'service' => 'redis',
        'broadcast_key' => 'phalcon:sse:stream:broadcast',
        'subject_key_prefix' => 'phalcon:sse:stream:subject:',
        'xread_block_ms' => 3000,
        'xread_count' => 10,
        'broadcast_stream_maxlen' => 1000,
        'subject_stream_maxlen' => 200,
    ],
],

如果应用没有 Redis DI 服务，可把 service 设为空，并配置独立连接：

'redis' => [
    'service' => null,
    'connection' => [
        'host' => '127.0.0.1',
        'port' => 6379,
        'auth' => '',
        'database' => 0,
        'timeout' => 2.5,
    ],
],

日志服务

组件使用 PSR-3 Logger。参考项目的 Monolog 2 可以直接注册：

$di->setShared('sseLogger', function () {
    $logger = new \Monolog\Logger('sse');
    $logger->pushHandler(new \Monolog\Handler\StreamHandler(
        APP_PATH . '/data/log/sse-' . date('Y-m-d') . '.log'
    ));

    return $logger;
});

不配置 logger_service 时使用 Psr\Log\NullLogger，不会强制项目安装 Monolog。

5. Controller 接入

完整示例见 examples/SseController.php。核心代码：

public function connectAction()
{
    $this->view->disable();

    // 必须来自服务端登录上下文。
    $subject = (string) $this->user->id;
    $lastEventId = $this->request->getHeader('Last-Event-ID') ?: null;

    $sse = $this->getDI()->getShared('phalconSse');
    $emitter = $this->getDI()->getShared('phalconSse.emitter');

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

SseEmitter 会：

设置 text/event-stream 响应头；
关闭 PHP 输出压缩；
解除正在使用的 PHP Session 锁；
设置无限执行时间；
逐帧 echo 并执行 ob_flush()/flush()；
客户端断开后退出循环；
返回内容为空的 Phalcon Response，避免入口文件二次输出正文。

6. 发布通知

定向通知：

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

广播：

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

建议只推送任务 ID、状态、错误摘要和结果地址，不通过 SSE 发送大型结果文件。

7. Redis Stream 驱动（默认）

环境变量示例：

SSE_DRIVER=redis
SSE_HEARTBEAT_INTERVAL=15
SSE_MAX_IDLE_TIME=7200
SSE_REDIS_HOST=127.0.0.1
SSE_REDIS_PORT=6379

能力：

每个 subject 一个私有 Stream；
所有连接共同读取一个广播 Stream；
使用近似 MAXLEN 裁剪；
SSE id 同时保存广播和私有 Stream 游标；
浏览器重连后可通过 Last-Event-ID 回放尚未被裁剪的消息；
新连接从建连时刻开始，不主动回放旧历史。

默认 Key：

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

Redis Stream 依赖 phpredis 5.x，组件要求：

ext-redis >= 5.0

参考项目实测版本为 phpredis 5.1.1。

8. RabbitMQ 驱动（可选）

RabbitMQ 驱动使用 PHP 扩展 ext-amqp，不是 php-amqplib。

启用：

'sse' => [
    'driver' => 'rabbitmq',
    'rabbitmq' => [
        'connection' => [
            'host' => '127.0.0.1',
            'port' => 5672,
            'login' => 'guest',
            'password' => 'guest',
            'vhost' => '/',
        ],
        'exchange' => 'phalcon_sse_notify',
        'subject_routing_key_prefix' => 'phalcon.sse.subject.',
        'broadcast_routing_key' => 'phalcon.sse.broadcast',
        'poll_interval_us' => 200000,
    ],
],

行为：

Topic Exchange；
每条 SSE 连接创建一个独占、自动删除的临时队列；
同时绑定用户私有 routing key 和广播 routing key；
使用 basic.get 非阻塞轮询，空轮询期间发送心跳；
连接结束后关闭 RabbitMQ 连接。

边界：

RabbitMQ 驱动只负责实时投递，不支持 Last-Event-ID 回放；
每条 SSE 连接会额外占用一个 RabbitMQ Connection；
大量连接场景优先使用 Redis Stream，或自行实现共享订阅/进程内路由驱动。

9. 前端 EventSource

const source = new EventSource('/admin/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 不能自由设置 Authorization Header：

Cookie 登录：使用 withCredentials；
Bearer Token：推荐短期一次性连接票据或 Fetch Streaming；
不要把长期令牌直接写入 URL。

10. Nginx 配置

参考项目是 Nginx + PHP-FPM，应关闭 FastCGI 缓冲：

location /admin/sse/ {
    try_files $uri $uri/ /index.php?_url=$uri&$args;

    include fastcgi_params;
    fastcgi_param SCRIPT_FILENAME $document_root/index.php;
    fastcgi_pass php73_fpm;

    fastcgi_buffering off;
    fastcgi_request_buffering off;
    gzip off;

    fastcgi_read_timeout 2h;
    fastcgi_send_timeout 2h;
}

响应头默认包含：

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

如果前面还有 Ingress、API 网关、CDN 或负载均衡器，也要关闭响应缓冲、压缩，并把空闲超时设为大于 SSE 心跳间隔。

11. PHP-FPM 容量边界

每条 Phalcon SSE 连接会长期占用一个 PHP-FPM Worker。

上线前至少确认：

; SSE 专用 FPM Pool 示例
pm = dynamic
pm.max_children = 100
request_terminate_timeout = 0
php_admin_value[max_execution_time] = 0
php_admin_flag[zlib.output_compression] = off

生产环境更推荐：

SSE 使用独立域名或路由；
SSE 使用独立 PHP-FPM Pool；
普通 API 与 SSE Worker 容量隔离；
监控活跃连接、FPM busy/idle process、Redis 连接数和内存；
根据峰值连接数压测后设置 pm.max_children。

如果预期同时存在数千至数万条长连接，PHP-FPM 不是理想承载模型，应考虑将 SSE 网关迁移到 Hyperf/Swoole、Node.js、Go 或专用推送服务。

12. 自定义驱动

实现：

use GodJarvis\PhalconSse\Contract\SseDriverInterface;

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

$di->setShared('customSseDriver', function () {
    return new CustomSseDriver();
});

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

subscribe() 必须返回生成器，并逐条产生 SseMessage：

yield SseMessage::event('notification', ['message' => 'hello']);
yield SseMessage::heartbeat();

13. 开发验证

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

当前测试覆盖：

SSE 帧格式和多行数据；
心跳和关闭事件；
Phalcon Response Header；
输出失败后停止；
Redis Stream 定向通知、广播和重连游标；
Phalcon DI 服务提供者；
自定义驱动注册。

真实 RabbitMQ Broker、Nginx/PHP-FPM 流式链路、浏览器断线重连、多实例以及容量压测，需要在使用方环境执行。

14. 发布到 Packagist

组件校验完成后：

git add .
git commit -m "feat: release Phalcon SSE component"
git push origin main

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

然后在 Packagist 提交：

https://github.com/GodJarvis/phalcon-sse

发布后用全新目录验证：

composer require godjarvis/phalcon-sse:^1.0

godjarvis / phalcon-sse

Maintainers

Package info

Statistics

Security