连接到数据流并使用数据

使用以下步骤连接到数据流并使用数据：

步骤 1：构建 WebSocket URL

从生成的 AsyncAPI YAML 中读取以下值：

servers.<name>.protocol：ws 或 wss。
servers.<name>.host：主机名和端口（如果存在）。
channels.<name>.address：完成数据流的 WebSocket 路径。

wss://<server-host><channel-address>?begin=<percent-encoded-begin-value>

请勿对路径进行硬编码： OCI GoldenGate、内部部署和反向代理可以公开不同的路由前缀。从生成的 YAML 或 REST API 返回的规范链接复制主机和通道地址。

步骤 2：选择开始值

根据下表中提供的准则指定 BEGIN 值：

值	使用	准则
`now`	从新活动开始	转换为当前时间戳并根据可用线索进行解析。不交付解析点之前的更改。
`earliest`	来自可用线索的初始回填	已转换为时间戳零。交付从已配置线索中仍可用的最早记录开始，不一定是源历史记录的开始。
ISO 8601 时间戳	在业务时间边界附近开始	服务执行基于时间戳的查找。使用显式时区，例如 `2026-07-13T04:00:00Z`。
已保存 `pos`	先前的成功运行后恢复	服务执行基于职位的查找。职位不可用时的结果取决于服务质量。

起始值的百分比编码

pos 标记可以包含 +、/ 和 = 等在 URL 中具有特殊含义的字符。在将完整标记添加到查询字符串之前对其进行百分比编码。请勿在应用程序逻辑中修改或解码标记。

步骤 3：验证和 TLS

选择验证方法。使用下表确定最合适的方法：

模式	何时使用	客户端响应
基本身份验证	与直接公开的服务或反向代理共用	发送 `Authorization: Basic ...` 标头。将凭证保存在密钥库或环境注入中，而不是在 URL 或源代码中。
摘要身份验证	反向代理需要时	使用在 WebSocket 握手期间实施 HTTP 摘要质询的客户机或生成的模板。
客户端证书	相互 TLS 或基于证书的服务验证	将客户端证书和私钥加载到 TLS 上下文中，并根据配置的 CA 验证服务器。
安全反向代理	网关终止或转发 TLS	遵循代理验证模式。安全代理可能需要基本/摘要凭证和客户端证书。

步骤 4：从 AsyncAPI 生成客户机

Oracle 为 AsyncAPI Generator 提供了 WebSocket 客户端模板。它可以生成 JavaScript、Python、Java、Go 或所有受支持的客户端。发电机需要 Node.js 18.12 或更高版本以及 npm 8.19 或更高版本。

npm install -g @asyncapi/generator

git clone https://github.com/oracle-samples/websocket-client-template.git
cd websocket-client-template

ag /path/to/sales-orders.yaml .
  -o output
  -p server=production
  -p language=python
  -p authorization=basic

cd output
python -m pip install -r requirements.txt
export ASYNCAPI_WS_CLIENT_USERNAME="$OGG_USER"
export ASYNCAPI_WS_CLIENT_PASSWORD="$OGG_PASSWORD"
python client.py

该模板还支持证书变量 ASYNCAPI_WS_CLIENT_CERT、ASYNCAPI_WS_CLIENT_KEY 和 ASYNCAPI_WS_CA_CERT。在组织通常的安全软件开发流程中查看生成的代码和依赖项版本。

步骤 5：了解 WebSocket 消息

通常的消息有效负载是 JSON 数组。一个帧可以包含一个或多个方案、元数据、DML 或 DDL 记录。客户机必须迭代数组并对每个对象进行独立分类。

背压：服务在没有按记录客户机确认的情况下推送记录。使接收循环保持轻量级。将缓慢的业务处理放在有边界的内部队列或持久暂存层后面，监视队列深度，并在内存耗尽之前停止或缩放使用者。

步骤 6：使用者处理循环

收到完整的 WebSocket 消息。
解析 JSON 数组。启用 CloudEvents 时，解包 data 成员。
对方案、流元数据、对象元数据、DML 和 DDL 记录进行分类。
需要验证时，根据最新的缓存方案验证记录。
应用业务运营或将其发布到持久的下游系统。
操作成功后，使用选定的原子或幂等检查点设计保存记录 pos。
在连接丢失时，重新连接最后一个持久 pos 和指数回退。