@aws- blocks/ bb-实时

由API Gateway WebSocket + DynamoDB支持的实时酒吧/子消息.

何时使用 : 实时将数据从服务器推向连接客户端——聊天信息,直播通知,仪表板更新,协作状态同步.

不使用时 : 如果您需要请求响应 API, 请使用 ApiNamespace。如果需要长长的信息排出保证发送的队列,则使用 AsyncJob。 。 。 .

缩放信封 : 最适合拥有上万至下千个同时用户的频道。 以线性方式公布延迟比例, 并列出用户数( 1 000 个用户的 ~ 100 ms) 。 对于有10K+订阅者的频道,使用明显的扇出模式(如:sharded) AsyncJob),对于100K+广播受众,考虑建立一个专门的WebSocket车队.

设计与模拟等价细节 : 设计

快速启动

import { Realtime } from '@aws-blocks/bb-realtime';
import { z } from 'zod';

const rt = new Realtime(scope, 'collab', {
  namespaces: {
    cursors: Realtime.namespace(z.object({ userId: z.string(), x: z.number(), y: z.number() })),
    chat: Realtime.namespace(z.object({ sender: z.string(), text: z.string() })),
  },
});

API 密码

命名空间

使用 Realtime.namespace(schema)。该计划既提供TypeScript类型(受孕),也提供发布时的运行时间验证.

这个 Realtime 实例显示三种方法, 都按命名空间名键( 类型为完全自动完成) :

只有运行时间. 这些方法(publish, (中文(简体) ), subscribe, (中文(简体) ), getChannel(按请求时间运行) ApiNamespace 方法, RawRoute 处理器、工作处理器或运行时脚本, 没有 在您的顶级 aws-blocks/index.ts。顶级代码运行于 CDK 合成期间,其中块解决其基础设施构造,其数据方法为 守护:顶级调用 Error 读取 Realtime.<method>() cannot be called during CDK synth. Data methods run at request time — …。要发布种子数据,请从处理器内部或单独的运行时脚本中进行。 在模块范围构造块是好的;只有方法调用才能移动到处理器中.

频道处理

getChannel() 返回 a Promise<RealtimeChannel<T>> —— 说 await 以获得一个仅订阅的手柄,通过 toJSON() 用于客户转账 :

频道控点有 没有 有个 publish() 方法。 出版总是通过 rt.publish() (server-side) 这样授权逻辑就留在您的代码中.

实时订阅

归来 subscribe() 在频道控件上 :

永远 await sub.established 在依赖订阅之前——这确保了WebSocket握手和服务器侧授权完成.

用法图案

通过 API 发布服务器系统

export const api = new ApiNamespace(scope, 'api', (context) => ({
  async sendMessage(roomId: string, text: string) {
    const user = await auth.requireAuth(context);
    await rt.publish('chat', roomId, { sender: user.id, text });
    return { sent: true };
  },
}));

返回频道处理( 认证门)

建议客户订阅模式。 授权在您的 API 方法中发生—— 频道控件只有在允许用户使用的情况下才会返回 :

export const api = new ApiNamespace(scope, 'api', (context) => ({
  async joinRoom(roomId: string) {
    const user = await auth.requireAuth(context);
    if (!canAccessRoom(user, roomId)) throw new Error('Forbidden');
    return rt.getChannel('chat', roomId);
  },
}));

客户端 :

const channel = await api.joinRoom('room-1');
const sub = channel.subscribe((msg) => {
  console.log(msg.sender, msg.text); // fully typed
});
await sub.established; // wait for server confirmation

服务器订阅

服务器也可以订阅频道——用于聚合、记录或触发副作用:

export const api = new ApiNamespace(scope, 'api', (context) => ({
  async monitorRoom(roomId: string) {
    const ch = await rt.getChannel('chat', roomId);
    const sub = ch.subscribe((msg) => {
      console.log(`[${roomId}] ${msg.sender}: ${msg.text}`);
    });
    await sub.established;
    return { monitoring: true };
  },
}));

在AWS上,服务器端的订阅使用真实的WebSocket连接,因此无论Lambda引用哪个内容发布,它都会收到消息. 在本地,它使用一个进程内的事件Emmitter.

多个订阅共享连接

当一个客户端订阅多个频道时,他们共享一个单一的WebSocket连接:

const ch1 = await api.joinRoom('room-1');
const ch2 = await api.joinRoom('room-2');

const sub1 = ch1.subscribe(handler1);
const sub2 = ch2.subscribe(handler2);

await sub1.established;
await sub2.established;

// sub1.connection === sub2.connection (same WebSocket)

信件被路由到正确的处理器 – 1室消息只到 handler1,2号房只到 handler2。 。 。 .

处理 Auth 失败

如果频道的信号无效或过期, established 拒绝 :

const sub = channel.subscribe(handler);
try {
  await sub.established;
} catch (err) {
  if (err.name === 'ConnectionFailedException') {
    // token was rejected — re-fetch the channel from the API
  }
}

失败的订阅 没有 在同一连接上杀死其他订阅 .

处理断开连接( R)

API 网关封装一个 WebSocket, 最多2小时的连接时间和10分钟的闲置时间 超时,所以一个长寿的订阅 将 放弃并重建 一辈子 客户端运输为您办理: 意外地放下 透明地重新连接到指数备份并重新订阅每个活动频道 (重放其存储的令牌),所以不需要手动重新订阅. ~9分钟 保持活平可以避免闲置的超时.

使用选项形式 subscribe() 观察生命周期:

const sub = channel.subscribe({
  onMessage: (msg) => { console.log(msg); },
  onDisconnect: (reason) => {
    // reason: 'client' | 'timeout' | 'error' | 'unknown'
    // Fires on EVERY drop (not just the first). Filter out 'client' (your own unsubscribe()).
    if (reason === 'client') return;
  },
  onReconnect: () => {
    // Fires once after the transport has reconnected AND the server has re-confirmed
    // the resubscribe (after the corresponding onDisconnect). Backfill any messages
    // missed during the gap here — WebSocket pub/sub is not durable, so re-read your
    // authoritative store (e.g. the conversation/record) rather than trusting the channel.
  },
});

onDisconnect 包括用户启动的断开点火 unsubscribe() (理由 'client'),并在之后的每滴同一逻辑订阅.

如果重新订阅最终被拒绝(存储的已通过 ~1h/~2h TTL 的令牌,或 或复试已用完,运输工具显示为 onDisconnect('error')。 。 。 . 没有 refresh 调回( 下方) , 即重排频道和 手动重新订阅.

与符号 TTL 共活 refresh

要让订阅能活过TTL 无 手卷重订、通过 (单位:千美元) refresh 回调 :

const sub = channel.subscribe({
  onMessage: (msg) => { console.log(msg); },
  // Called before EACH reconnect to re-mint a fresh descriptor (new connect + channel
  // token), so the subscription outlives the channel (~1h) / connect (~2h) token TTLs.
  // Return the RAW descriptor (the wire object with `__blocks`/token fields), e.g. your
  // server's `channel.toJSON()` with `__blocks` stripped, then re-add the discriminant.
  refresh: async () => ({ ...(await fetchFreshDescriptor()), __blocks: 'realtime/channel' }),
});

refresh: () => Promise<RealtimeChannelDescriptor> 在套接字重新打开前被引用 每个重连接; 新建连接符 新建套接字 URL, 而新建频道符 是 在重新订阅时重播, 因此通过 ~1h/~2h 边界的一滴会透明恢复 放弃。 连接上的每个信道都同时重新闪烁自己的符号; a refresh 拒绝(或超过内部的超时)而降入已贮存的印记 全通道失败结果表面 onDisconnect('error')。后面的服务器方法 refresh 重新发布频道符号,所以它必须应用与方法相同的授权 发布原始频道.

没有连接 refresh 横跨连接符的~2h生命. 没有 refresh, (中文(简体) ), 重新连接 存储 连接符 (~2h TTL, 匹配 API Gateway 最大 2h) 连接); 超过 2 h $connect 握手被拒绝 重新连接放弃 终端 onDisconnect('error')。提供 refresh (以上) 在每个上面重新点燃新鲜的符号 重新连接并关闭窗口; 否则处理终端 onDisconnect('error') 作为 提示重勾新频道控件并手动重新订阅 .

Schema 验证

每个 publish() 运行时对照计划进行验证 :

import { isBlocksError } from '@aws-blocks/core';
import { RealtimeErrors } from '@aws-blocks/bb-realtime';

try {
  await rt.publish('chat', 'room-1', { sender: 123 }); // wrong type
} catch (e: unknown) {
  if (isBlocksError(e, RealtimeErrors.ValidationFailed)) {
    // data failed schema validation
  }
}

错误常数

import { RealtimeErrors } from '@aws-blocks/bb-realtime';

RealtimeErrors.ValidationFailed  // data failed schema validation on publish
RealtimeErrors.PublishFailed     // Publish fan-out failed (AWS only)
RealtimeErrors.ConnectionFailed  // WebSocket connection or subscribe rejected

最佳做法

• 等着 established 在发布或依赖订阅之前——不要与WebSocket握手.

• 发布前订阅. 实时是火与遗忘, 没有消息缓冲( 本地 dev 和 AWS 相同 ): 用户只接收已发布的消息 之后 其订阅登记册(等待) sub.established) (中文(简体) ). 如果服务器端的发布器(如AsyncJob或Agent stream)可能在客户端订阅之前发火,客户端必须先订阅,再触发发布器,并回填来自持久来源(DB/History)的任何更早消息.

• 通过 API 发布,而不是频道控件——将授权逻辑保存在一个地方.

• 使用频道进行动态范围界定 —— 说 room-123, (中文(简体) ), user-456, (中文(简体) ), game-abc一个名字空间,很多频道.

• 保持有效载荷较小 ——大型信息会增加延迟性和成本. Max 128 KiB (131,072字节) 每条已发布消息(包括有线信封).

• 一个 Realtime 每个域实例 ——对不同消息类型(cursors,聊天,存在)使用其内部的多个命名空间.

• 完成时取消订阅 - 特别是在起/起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起起 被泄露的订阅会让WebSocket打开.

• 送货最好 —— 说 publish() 以并行方式向所有连接的订阅者发送。 如果发送到单个连接失败,则失败会被记录下来,其余继续. 连接自动清理。 这与UDP类似:每条接通的火和被遗忘.

• 大粉丝出局 —— 说 publish() 等待所有订户收到,然后返回。 对于有几十个订户的频道,这增加了微不足道的延迟。 对于较大的观众,卸载到 AsyncJob 因此API响应没有被屏蔽:

  const broadcast = new AsyncJob(scope, 'broadcast', {
    schema: z.object({ channel: z.string(), data: cursorSchema }),
    handler: async ({ channel, data }) => { await rt.publish('cursors', channel, data); },
  });
  
  // In your API method:
  await broadcast.submit({ channel: roomId, data: cursor }); // returns immediately
  

• 广播规模(10K+订户) - 单一的Lambda引用击打 postToConnection 每个用户都会点击 API 网关 TPS 限制( ~ 10K 默认) 。 对于这个尺度,将粉丝的出动 跨多个 AsyncJob 引用, 每一个负责一个片的订阅者。 这弥补了差距,同时你评估一个专用的WebSocket机队真实的广播工作量.

• 测试沙盒中的认证流 ——使用 npm run test:e2e:sandbox 以验证已全部部署的认证流端到端.

缩放成本( 美国东部)

以发送信息为线性成本尺度。 API Gateway消息费占主导地位.

每页成本(对N用户)

出版的呼号本身是一个Lambda引用,而不是WebSocket消息——对入境方没有API Gateway充电. DynamoDB查询成本为 <0.01%.

连接成本

1 000个联机24/7美元 每月18美元 10 000个联机每天8小时 48美元.

出版间隔

publish() 并行发送消息,受SDK的 HTTP 连接池(50个默认的并发套接字)的限制. 每个 postToConnection 在同一区域内,使用TCP保持活性,需要~5-10ms.

AWS SDK 重试自动被节奏的请求( 有指数回放的重试) .

限制

地方发展

当地 dev(npm run dev), Realtime 使用一个带有WebSocket桥段的处理中事件Emitter来浏览器客户端. 本地 WebSocket 服务器发送 subscribe_success 有效订阅和 error 在无效的符号上。 Schema验证运行与AWS相同.

信息是瞬息万变的——在重启过程中没有持久性.

频道名称

频道名称被实时实例和命名空间所覆盖 : {fullId}/{namespace}/{channel}编辑 channel 向您传递的参数 publish(), (中文(简体) ), subscribe(),以及 getChannel() 是用户界面部分.

建议的限额:

• 保留频道名称 256 个字符。完整通道路径(包括实例ID和命名空间)作为 DynamotB 排序键存储,并包含在每条 WebSocket 消息中 .
• 使用简短的描述性标识符: room-123, (中文(简体) ), user-456, (中文(简体) ), game-abc。 。 。 .
• 避免将大型有效载荷或UUID嵌入超过必要的频道名称.

硬性限制 :

• 全频道路径(包括 {fullId}/{namespace}/ 前缀)必须符合DynamotB排序键(1024字节). 本地的dev 和 AWS 在运行时执行, 超过它扔 ValidationFailedException。 。 。 .
• 已发布的消息(包括带频道路径的有线信封)必须符合API Gateway的128 KiB逻辑WebSocket消息限制. API Gateway 单独的 32 KB 配额是一个 边框 限制——服务重新组装被打碎的帧,所以它不是每个消息封顶. 这在两种环境中都是在运行时间执行的.
• 对用户-造型通道名称没有强制字符限制,但过长的名称会增加DynamomDB的项目大小(影响读取/写作费用)和WebSocket消息大小(影响交付空档和计费,因为消息以32 KB增量计).

之前的AppSync Eventions执行有5个分块×50个字符频道名限制. 这一限制不再适用.

另见

• (单位:千美元) ——常用案例,缩放模式,行业实例