> ## Documentation Index > Fetch the complete documentation index at: https://kawax.biz/llms.txt > Use this file to discover all available pages before exploring further. # WebSocket (Jetstream / Firehose) > Laravel Bluesky 的 WebSocket 功能。介绍如何通过 Bluesky Jetstream 与 AT Protocol Firehose 处理实时事件，以及 JetstreamServeCommand 和 FirehoseServeCommand 的用法。 ## 概览 `laravel-bluesky` 提供了两条 WebSocket 命令，用来连接 Bluesky 的实时事件流。 * **Jetstream** —— Bluesky 提供的、已经过滤的 WebSocket 端点。JSON 格式，比较轻量。 * **Firehose** —— AT Protocol 的原始事件流。以 DAG-CBOR 二进制格式接收全部数据。 ```mermaid theme={null} graph LR subgraph "Bluesky 网络" J["Jetstream
jetstream1.us-west.bsky.network
JSON / 可过滤"] F["Firehose
bsky.network
DAG-CBOR / 全量数据"] end subgraph "Laravel 应用" WS["JetstreamServeCommand
FirehoseServeCommand
(Workerman)"] E["Laravel Events
JetstreamCommitMessage
FirehoseCommitMessage 等"] L["Listener / Job"] end J -->|WebSocket| WS F -->|WebSocket| WS WS --> E --> L ``` 基于 WebSocket 的长时运行进程需要 **VPS 或 EC2 等常驻服务器**，或者 **Laravel Cloud 的自定义 worker**。它不能在 Laravel Vapor、Vercel 等无服务器环境中运行。 ## 安装 WebSocket 功能依赖 [Workerman](https://github.com/walkor/workerman)。 ```bash theme={null} composer require workerman/workerman ``` ## Jetstream ### 概览 Jetstream 是 Bluesky 提供的、已经预先过滤的 WebSocket 服务。可以按 collection 类型或用户 DID 过滤，从而只接收需要的事件，效率更高。 | 特性 | 说明 | | ---- | -------------------- | | 数据格式 | JSON | | 过滤 | 可按 collection、DID 过滤 | | 数据量 | 取决于过滤条件，可以较为轻量 | | 用途 | 监听帖子、点赞、关注等 | ### 启动方式 ```bash theme={null} # 接收全部消息（无过滤） php artisan bluesky:ws start # 调试：显示所有接收到的消息 php artisan bluesky:ws start -v ``` ### collection 过滤 通过 `-C` 选项过滤要接收的 collection。可以多次指定。 ```bash theme={null} # 只接收帖子和点赞 php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like # 只接收关注事件 php artisan bluesky:ws start -C app.bsky.graph.follow ``` 常用 collection: | collection | 内容 | | ----------------------- | ---------- | | `app.bsky.feed.post` | 帖子的创建 / 删除 | | `app.bsky.feed.like` | 点赞 | | `app.bsky.feed.repost` | 转发 | | `app.bsky.graph.follow` | 关注 | | `app.bsky.graph.block` | 屏蔽 | ### DID 过滤 通过 `-D` 选项只接收特定用户的事件。 ```bash theme={null} # 只接收特定用户的帖子 php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy ``` ### 事件处理 Jetstream 命令会根据接收到的消息类型派发 Laravel 事件。 | 事件类 | 触发时机 | | -------------------------- | ----------------------- | | `JetstreamMessageReceived` | 收到任何消息时 | | `JetstreamCommitMessage` | record 的创建 / 更新 / 删除时 | | `JetstreamIdentityMessage` | handle 变更等 identity 事件时 | | `JetstreamAccountMessage` | 账号启用 / 停用时 | 创建事件监听器来处理事件。 ```bash theme={null} php artisan make:listener JetstreamPostListener ``` ```php theme={null} namespace App\Listeners; use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage; class JetstreamPostListener { public function handle(JetstreamCommitMessage $event): void { // 判断 collection 类型 $collection = data_get($event->message, 'commit.collection'); if ($collection !== 'app.bsky.feed.post') { return; } // 操作类型：create / update / delete $operation = $event->operation; // 发帖者 DID $did = $event->message['did']; // record 内容 $record = data_get($event->message, 'commit.record'); $text = data_get($record, 'text', ''); info("[$operation] $did: $text"); } } ``` ## Firehose ### 概览 Firehose 是 AT Protocol 的原始事件流。可以以二进制（DAG-CBOR）格式接收 Bluesky 网络上所有的 record 操作。 | 特性 | 说明 | | ---- | -------------------- | | 数据格式 | DAG-CBOR 二进制（包会自动解码） | | 过滤 | 无（全部接收） | | 数据量 | 非常庞大 | | 用途 | 全量数据的收集与归档 | DAG-CBOR 的解码由包自动完成。在事件监听器中，你可以像使用普通 PHP 数组一样处理数据。 ### 启动方式 ```bash theme={null} php artisan bluesky:firehose start # 调试：显示接收到的消息 php artisan bluesky:firehose start -v ``` ### 事件处理 Firehose 命令也是通过 Laravel 事件来处理消息。 | 事件类 | 触发时机 | | ------------------------- | --------------------- | | `FirehoseMessageReceived` | 收到任何消息时（包含原始数据） | | `FirehoseCommitMessage` | record 的创建 / 更新 / 删除时 | | `FirehoseIdentityMessage` | identity 事件时 | | `FirehoseAccountMessage` | 账号事件时 | | `FirehoseSyncMessage` | 仓库同步事件时 | ```bash theme={null} php artisan make:listener FirehosePostListener ``` ```php theme={null} namespace App\Listeners; use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage; class FirehosePostListener { public function handle(FirehoseCommitMessage $event): void { // 判断 collection 类型 if ($event->collection !== 'app.bsky.feed.post') { return; } // 操作类型：create / update / delete $action = $event->action; // 发帖者 DID $did = $event->did; // record 内容（已解码为数组） $record = $event->record; $text = data_get($record, 'value.text', ''); info("[$action] $did: $text"); } } ``` ## 配置 可以在 `config/bluesky.php` 中修改连接目标主机和日志配置。 ```php theme={null} // Jetstream 'jetstream' => [ 'host' => env('BLUESKY_JETSTREAM_HOST', 'jetstream1.us-west.bsky.network'), 'max' => env('BLUESKY_JETSTREAM_MAX', 0), // maxMessageSizeBytes (0 = 无限制) 'logging' => [ 'driver' => env('BLUESKY_JETSTREAM_LOG_DRIVER', 'daily'), 'days' => 7, 'path' => env('BLUESKY_JETSTREAM_LOG_PATH', storage_path('logs/jetstream.log')), ], ], // Firehose 'firehose' => [ 'host' => env('BLUESKY_FIREHOSE_HOST', 'bsky.network'), 'logging' => [ 'driver' => env('BLUESKY_FIREHOSE_LOG_DRIVER', 'daily'), 'days' => 7, 'path' => env('BLUESKY_FIREHOSE_LOG_PATH', storage_path('logs/firehose.log')), ], ], ``` `.env` 配置示例： ```ini theme={null} BLUESKY_JETSTREAM_HOST=jetstream2.us-east.bsky.network BLUESKY_JETSTREAM_MAX=1000000 ``` ## 与 Labeler 结合 可以同时启动 Labeler 服务器和 Jetstream / Firehose。Labeler 处理打 label 请求时可以利用 Jetstream 或 Firehose 的数据。 ```bash theme={null} # Labeler + Jetstream（监听关注事件） php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow # Labeler + Firehose（接收全部数据） php artisan bluesky:labeler:server start --firehose ``` Labeler 的详情请参考 [Labeler 页面](/zh/packages/laravel-bluesky/labeler)。 ## 长时进程的运维 WebSocket 命令是长期运行的进程。在生产环境中，请使用 Supervisor 等进程管理工具。 ### Supervisor 配置示例 `/etc/supervisor/conf.d/bluesky-jetstream.conf`： ```ini theme={null} [program:bluesky-jetstream] process_name=%(program_name)s_%(process_num)02d command=php /var/www/html/artisan bluesky:ws start -C app.bsky.feed.post autostart=true autorestart=true stopasgroup=true killasgroup=true user=www-data numprocs=1 redirect_stderr=true stdout_logfile=/var/www/html/storage/logs/jetstream-worker.log stopwaitsecs=3600 ``` ```bash theme={null} # 重新加载 Supervisor 并启动 sudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start bluesky-jetstream:* ``` ### 在 Laravel Forge 中配置守护进程 如果你使用 Laravel Forge，可以在 **Daemons** 部分添加守护进程。 * **Command**: `php artisan bluesky:ws start -C app.bsky.feed.post` * **Directory**: `/var/www/html` * **User**: `forge` ### 在 Laravel Cloud 中配置后台进程 WebSocket 命令作为 **WebSocket 客户端** 连接到 Bluesky 的事件流，因此在 Laravel Cloud 上也能运行。将其作为 Laravel Cloud 的后台进程（自定义 worker）配置即可。 在 Laravel Cloud 的后台进程配置中添加 **Custom Worker**。 **Jetstream：** ```bash theme={null} php artisan bluesky:ws start ``` **Firehose：** ```bash theme={null} php artisan bluesky:firehose start ``` 部署时的进程停止与重启等操作，都会由 Laravel Cloud 自动处理。除了后台进程本身的配置外，不需要其他额外设置。 ### 注意事项 * 进程意外退出时，`autorestart=true` 会自动将其重启。 * 为了避免内存泄漏，可以考虑定期重启。 * 对于消息量巨大的 Firehose，建议把 listener 中的处理异步化（通过 Queue Job）。 ```php theme={null} // 在 listener 中将处理派发给 Queue Job 的示例 namespace App\Listeners; use App\Jobs\ProcessFirehosePost; use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage; class FirehosePostListener { public function handle(FirehoseCommitMessage $event): void { if ($event->collection !== 'app.bsky.feed.post') { return; } // 将耗时处理委托给 Queue Job ProcessFirehosePost::dispatch($event->did, $event->record, $event->action); } } ``` Source: [src/Console/WebSocket](https://github.com/invokable/laravel-bluesky/tree/main/src/Console/WebSocket) ## Related topics - [Laravel Cloud — Laravel 专用 PaaS 全貌](/zh/blog/laravel-cloud.md) - [流式输出](/zh/packages/laravel-copilot-sdk/streaming.md) - [Laravel Nostr](/zh/packages/laravel-nostr.md) - [Laravel Reverb](/zh/reverb.md) - [Laravel Notification for Discord(Webhook)](/zh/packages/laravel-notification-discord-webhook.md)