一、认识 Pub/Sub API
本单元从基础开始：Pub/Sub API 是什么、它与传统 API 有何不同，以及它的关键架构特性。
学完本单元，你将能够：解释 Pub/Sub API 的用途、列出 Pub/Sub API 的特性与功能。
开始之前
开始前应熟悉各类实时事件，建议先完成 Platform Events Basics、Change Data Capture Basics、Real-Time Event Monitoring Events 这几个模块。Pub/Sub API 基于 gRPC——一个高性能、开源的 RPC 框架。与 REST API 的"请求-响应"模式不同，gRPC 通过 HTTP/2 使用双向流进行实时通信：客户端与服务器可随时互相发送消息，实现真正的实时事件流。理解这一根本架构差异，是掌握 Pub/Sub API 的前提。
Pub/Sub API 用来做什么
Pub/Sub API 是 Salesforce 基于 gRPC 的 API，用于实时事件流与发布。两个主要方向：订阅（Subscribe）——实时监听 Salesforce 事件（Change Data Capture 记录变更、Platform Events 自定义事件消息），对数据变化即时响应、无需轮询；发布（Publish）——把事件发送进 Salesforce（外部系统发布平台事件、实时集成外部数据源）。
用 Pub/Sub API 可以做：发布与订阅自定义平台事件，在另一个系统发送/接收实时通知做业务流程集成（如维护服务或产品订单的实时通知）；订阅 change data capture 事件，在外部库存系统同步订单数据；订阅实时事件监控事件，实时监控安全相关的用户操作（登录、查询敏感数据、运行报告）；订阅标准平台事件（如 AppointmentSchedulingEvent）与 Google Calendar 集成更新用户日历。相比旧的 Streaming API（CometD/Bayeux），Pub/Sub 用 gRPC + HTTP/2（现代高效）、Protocol Buffers 二进制格式（而非 JSON）、双向流（而非仅服务器到客户端）。
Pub/Sub API 的特性与功能
Pub/Sub API 有五个核心特性：精简的数据传输（HTTP/2 上的二进制 Protocol Buffer 消息，比 JSON 小得多）、单一 API 完成所有操作（订阅与发布通过一个统一接口）、双向流（客户端与服务器同时发送消息）、最终的发布确认（确认事件已成功发布）、基于拉取的订阅（客户端控制事件消费速率，即流量控制）。下面逐一展开。
用二进制事件消息与 HTTP/2 精简数据传输
Pub/Sub API 基于 gRPC（一个健壮的开源 RPC 框架，用于构建可扩展、快速的 API）与 HTTP/2，受益于 HTTP/2 的压缩，传输的数据更小。它发布与投递二进制编码的事件，payload 用 Apache Avro schema 编码——Avro 是一个数据序列化系统，提供二进制或 JSON 数据编码与 schema，二进制编码比 JSON 更高效（序列化更快、序列化后的数据更小）。
Protocol Buffers 与 HTTP/2 相比 JSON over HTTP/1.1 是显著的性能升级：Protobuf 消息是二进制、基于 schema 的，既更小又在编译期（而非运行时）校验；HTTP/2 在单个 TCP 连接上多路复用多个流，消除了困扰 REST API 的连接开销。对每秒处理数千事件的事件流场景，这些效率不是锦上添花，而是实时集成可行的必要条件。
单一 API 完成多种操作
一个 API、两个方向、多种操作。传统方式：REST API 用于发布（每个事件类型一个端点）、Streaming API 用于订阅（单独的 CometD 连接），协议与客户端库都不同。Pub/Sub API 的统一方式：一个 gRPC 服务含多个 RPC 方法——Subscribe RPC（从 Salesforce 接收事件）、Publish RPC（发送单个事件到 Salesforce）、PublishStream RPC（发送批量事件），全部通过同一种连接类型、同一个客户端。这简化了架构、减少了集成点，一套客户端库搞定一切。
通过双向流提升性能
REST 是"请求 → 响应 → 连接关闭"，Pub/Sub 的 gRPC 是"持久连接，双方都可流式发送"。双向流的优势：客户端与服务器同时交换消息、无轮询开销（事件发生时即时到达）、连接保持打开（无需反复认证/TLS 握手）、服务器可随时推送事件、客户端可随时发布事件。类比：REST 像每次提问都要打一次电话，Pub/Sub 像保持一条开着的对讲机频道。
使用双向流的 RPC 方法：PublishStream、Subscribe、ManagedSubscribe。以 PublishStream 为例：它可以边发送一串发布请求、边接收一串发布响应，无需等一个响应再发下一个请求，多个发布批次可排队，从而获得比 Publish 一元 RPC 更高的发布速率。
最终的发布确认
用 Publish 或 PublishStream 发布事件时，Pub/Sub API 会等系统把事件存入事件总线后才返回发布结果——PublishResponse 承载的是最终的发布结果，返回后事件总线上发布已完成、没有遗留的异步过程。每个事件返回一个 PublishResult，指示成功或失败：成功意味着事件已发布并存储在事件总线，失败意味着事件未发布、不在事件总线。
相比之下，用 REST API 或 Apex 等方法发布事件不提供最终结果——返回的只是系统异步发布排队事件之前的中间入队结果。用 Pub/Sub API，你可以确信发布结果是最终的。这解决了事件驱动架构的关键可靠性缺口：不会因网络问题静默丢事件。
基于拉取的订阅与流量控制
Pub/Sub 使用基于拉取的订阅——客户端控制自己接收事件的速率。流程：客户端请求 N 个事件（如"给我 100 个事件"）→ 服务器最多发送 N 个 → 客户端按自己的节奏处理 → 准备好时请求下一批。用 Subscribe 和 ManagedSubscribe RPC，客户端发送请求指定想接收的事件数（最多 100），收到完整或部分的批量事件。
流量控制的优势：慢消费者 → 一次少请求一些事件；快消费者 → 请求更大批次；处理高峰 → 暂时暂停请求；不会因消费者过载而丢事件。对比旧的 push 式（CometD）——服务器尽快推送，慢消费者被压垮、缓冲区溢出时事件会丢失。基于拉取 = 消费者保护：你的客户端自己定节奏，永不被压垮。
二、用受管订阅（Beta）订阅事件
本单元探索受管订阅（Managed Subscriptions）——一种声明式订阅事件的方式，无需直接管理 gRPC 连接。
学完本单元，你将能够：描述受管订阅是什么、描述其好处、解释何时及如何发送 commit 请求、解释配置受管订阅所用的元数据。
受管订阅（Managed Subscriptions，Beta）
用受管订阅跟踪订阅客户端已消费的事件，并在客户端断开后从上次位置恢复订阅。用 ManagedSubscribe RPC，你可以在服务器上提交（commit）最后处理事件的 Replay ID；订阅断开并重启后，从事件总线中存储的未处理事件处无缝恢复。没有受管订阅时，你可以用 Subscribe RPC 从特定 Replay ID 恢复，但这要求你在客户端管理 Replay ID 的存储；用受管订阅，你不用操心在客户端存储 Replay ID。
与 Subscribe 一样，ManagedSubscribe 用双向流、基于拉取：客户端用 ManagedFetchRequest 请求事件数，收到含一批事件的 ManagedFetchResponse，处理完事件后 commit 最后处理事件的 Replay ID。注意：受管订阅是 Beta 服务，受 Beta Services Terms 约束；一个受管订阅对每个客户端是唯一的，同一 Salesforce org 下不能与其他客户端共享。
配置受管订阅
用 Metadata API 和 Tooling API 中的 ManagedEventSubscription 配置受管订阅的行为与状态。关键字段：topicName（指定平台事件、变更事件或实时事件监控事件的订阅路径，也可指定对多个平台/变更事件分组的 channel）；defaultReplay（指定订阅首次启动、停止后、或无存储 replay ID 时从哪里开始，如从最新收到的事件开始）；state（订阅是 running 还是 stopped，管理员可改为 STOP 停止订阅）；errorRecoveryReplay（若已提交的 Replay ID 无效——如对应事件在事件保留窗口之外——订阅从该位置重启，如从最新事件开始）。
用受管订阅提交 Replay ID
收到并处理一批事件后，commit 最后处理事件的 Replay ID（协议规范中的 replay_id 字段）。要 commit 一个 Replay ID，把它设在 CommitReplayRequest 里、再放进 ManagedFetchRequest 发送。为什么 commit：告诉 Salesforce"我已处理到这个事件"，重启/断开时从最后提交的 ID 恢复，防止重复处理已处理事件、防止错过停机期间到达的事件。
最佳实践：处理后再 commit，而非处理前。若处理失败就不 commit → 事件会重投。还要在收到空批次事件后 commit——ManagedFetchResponse 的 latest_replay_id 字段含最新 Replay ID，指向流中最后消费事件之后的更前位置（即使没收到事件也可能有更前位置，因为事件总线合并了多个 org 的流，但不会跨 org 共享任何 org 特定信息或 PII）。存储较新的 Replay ID 确保订阅恢复时服务器不用花太多时间在事件总线中查找事件。
保持订阅活跃
如果服务器有基于客户端请求事件数的待投递事件，ManagedSubscribe 调用流会保持打开；如果有待投递事件但事件总线暂无新事件，服务器会发送 keepalive 消息；否则客户端应发送 fetch 请求和 commit 请求来保持流活跃。事件流可能长时间安静（无记录变更、无平台事件），没有 keepalive 信号时 Salesforce 可能认为消费者已断开并停用订阅，当事件最终到达时却因订阅不活跃而无法投递。因此始终在消费者里实现 keepalive 逻辑（定期发送一个简单请求即可），并监控订阅健康状态以便在订阅意外不活跃时收到告警。
三、订阅 Opportunity 变更事件
本单元动手实践：通过 Tooling API 配置受管订阅、创建启用变更事件的 Opportunity 记录，并走完订阅的启动/关闭/恢复生命周期。
学完本单元，你将能够：在 Tooling API 中配置受管订阅、用 ManagedSubscribe RPC 订阅变更事件（运行 Java 示例客户端）、在断开并重启客户端后从上次位置恢复订阅。
用 Tooling API 配置受管订阅
背景：Get Cloudy Consulting 的软件开发者 Robert Bullard 正在为一个销售太阳能板的客户开发订单管理的事件驱动应用。机会关闭时，订单应用向外部订单履行系统发送请求；订单履行后，外部系统向 Salesforce 发送一个带运单号的 Order Fulfillment 平台事件。Robert 决定用 Pub/Sub API 订阅机会变更事件、发布 Order Fulfillment 事件回去，并选择受管订阅（便于跟踪最后收到/处理的事件、客户端意外断开时能恢复，让客户端更健壮）。
连接 Postman：完成 Quick Start: Connect Postman to Salesforce 后，打开 Postman，在 Authorization 标签点 Get New Access Token → Allow，复制 instance URL → Use Token，在 Variables 标签把 _endpoint 的 CURRENT VALUE 粘贴成 instance URL → Save；在 Collections 选 Salesforce Platform APIs collection → REST → GET Limits → Send，Status 应显示 200 OK。
创建 ManagedEventSubscription：向 REST 端点 /tooling/sobjects/ManagedEventSubscription 发送 POST（在 collection 展开 Event Platform → Managed Event Subscriptions → Create managed event subscription），使用请求体，Send 后响应返回创建的 ManagedEventSubscription 记录 ID。若找不到 Managed Event Subscriptions 文件夹，刷新 collection（点 collection 旁的 ... → Pull changes）。
为 Opportunity 记录启用变更事件
变更事件必须按对象显式启用。Setup → Change Data Capture → 在 Available Entities 选 Opportunity (Opportunity) → 点 > 箭头 → Save。启用后，Opportunity 记录的每次修改都会生成变更事件。变更事件包含：哪些字段变化（旧值 → 新值）、谁做的更改、何时发生、用于关联的记录 ID。没有启用变更事件，订阅就没有事件可投递。
创建 Opportunity 记录：在 Trailhead Playground 打开 Accounts → All Accounts → Grand Hotels & Resorts Ltd → Opportunities related list → New，创建两个机会：Grand Hotels - 100 solar panels（Amount 30,000，Close Date 未来某天，Stage Value Proposition）与 Grand Hotels - solar panel installation（Amount 20,000，Close Date 未来某天，Stage Value Proposition）。
启动受管订阅
用 Java 示例客户端启动受管订阅（调用 ManagedSubscribe RPC，Beta）。前提：Java 11、Apache Maven，以及安全令牌（若 org 未设置可信 IP 范围，需把安全令牌追加到密码后）。
步骤：fork 并克隆 pub-sub-api GitHub 仓库 → 终端进入 java 文件夹 → 运行 mvn clean install 构建并生成源码 → 在 arguments.yaml 里配置参数：PUBSUB_HOST/PUBSUB_PORT 保持默认，LOGIN_URL 填 Salesforce 登录 URL（https://login.salesforce.com 或 My Domain URL），USERNAME 填 playground 用户名，PASSWORD 填密码+安全令牌，MANAGED_SUB_DEVELOPER_NAME: Managed_Sub_OpportunityChangeEvent，PROCESS_CHANGE_EVENT_HEADER_FIELDS: true → 在 java 文件夹运行 ./run.sh genericpubsub.ManagedSubscribe。由于还没发布事件，客户端暂时收不到任何事件。
关闭一个 Opportunity 并接收变更事件
测试订阅：创建并关闭一个机会。App Launcher → Opportunities → All Opportunities → 点 Grand Hotels - 100 solar panels → Edit → Description 填 Closing Grand Hotels - 100 solar panels opportunity. → Stage 改为 Closed Won → Save。机会记录的变化触发变更事件生成，订阅客户端收到它。事件 payload 包含旧的 Stage 值与新的 Stage 值（Closed Won）、更改时间戳、做更改的用户、记录 ID。收到事件就确认了整条链路：记录变更 → 事件 → 投递。
从最后提交的 Replay ID 恢复订阅
验证受管订阅跟踪已处理事件、并从上次位置恢复：Java 客户端已代你 commit 了最后处理的事件。断开客户端（终端按 Ctrl+C），在 Salesforce 中再打开另一个机会 Grand Hotels - solar panel installation → Edit → Description 填 Closing Grand Hotels - solar panel installation opportunity. → Stage 改为 Closed Won → Save，然后重启客户端（同一终端运行 ./run.sh genericpubsub.ManagedSubscribe）。订阅返回客户端关闭期间错过的变更事件——已处理的事件被跳过（无重复）、停机期间到达的事件被投递（零丢失）。
这就是 Replay ID 作为事件流游标的机制：它让订阅可重启、可靠，是生产级事件处理的基础。
检查示例代码
示例 Java 客户端里，ManagedSubscribe RPC 在 ManagedSubscribe.java 中调用：startManagedSubscription() 方法启动受管订阅，构建一个含 ManagedEventSubscription 的 Tooling API 记录 ID 或 developer name 的 ManagedFetchRequest，用 serverStream.onNext() 在流上发送（serverStream 类型为 StreamObserver<ManagedFetchRequest>）；doCommitReplay() 方法提交最后处理事件的 Replay ID，构建一个含 Replay ID 与 commit 请求 ID（随机 UUID）的 CommitReplayRequest，加入 ManagedFetchRequest 后在同一流上发送。
示例代码展示了完整的生产级 Pub/Sub 客户端模式——认证、订阅管理、事件处理、Replay ID 提交、错误处理与重连逻辑，是构建自己事件消费者的模板。研究它、改造它、构建你自己的消费者。
创建 Opportunity 记录并启用变更事件——概览
变更事件必须为 Opportunity 对象启用，订阅才能收到。Setup → Change Data Capture → 选 Opportunity → 为 Create、Update、Delete 启用变更事件。然后创建测试 Opportunity 记录，作为验证订阅工作的事件源。变更事件包含丰富的变更信息：旧值与新值、做更改的用户、时间戳、记录 ID——这些丰富的 payload 让复杂的集成逻辑成为可能。
启动、关闭、恢复——订阅生命周期
订阅生命周期演示了 Replay ID 概念的实际应用。启动订阅，它开始接收变更事件——更新一个 Opportunity，事件到达。关闭一个 Opportunity，变更事件携带新旧 Stage 值到达。若订阅断开，记下最后提交的 Replay ID，创建一个从该 ID 开始的新订阅，处理就从停止的地方无缝继续——停机期间到达的事件被投递、已处理的事件不重复。Replay ID 就是你的恢复点，是可靠、可重启事件处理的关键。
四、发布订单履行事件
本单元从订阅转向发布——定义 Order Fulfillment 平台事件，并用 Pub/Sub API 的 Publish 与 PublishStream 方法发布到 Salesforce。
学完本单元，你将能够：解释 Publish 与 PublishStream RPC 方法的区别、用 PublishStream RPC 发布订单履行事件、订阅订单履行事件。
Publish RPC 方法
自定义订单应用收到每个已关闭机会的变更事件后，会在外部订单系统下单（这部分超出本模块范围）。应用收到订单已履行的通知后，向 Salesforce 发布一个平台事件；平台事件收到后，应用用发货信息（如运单号）更新机会记录。
Publish RPC：想等一个发布请求的结果再发下一批、且不需要大批量高发布速率时使用。Publish 是一元 RPC——只发一个请求、收一个响应，同步发布 PublishRequest 中的一批事件，服务器发布完事件消息后向客户端发送响应。PublishResponse 中的 PublishResult 在发布失败时含错误，无错误即发布成功、事件在事件总线中，可重试失败事件。最适合：每个事件代表独立业务交易（如单个订单履行）的场景。
PublishStream RPC 方法
PublishStream RPC：想获得比 Publish 更高的发布速率、处理大批量事件时使用。它用双向流，可边发一串发布请求、边收一串发布响应，服务器对每个 PublishRequest 返回一个 PublishResponse（批次发布完成时），客户端无需等一个 PublishResponse 再发新 PublishRequest，多个批次可排队——因为客户端可在服务器端仍在处理发布操作时异步发布更多事件，从而获得更高发布速率。
对 PublishRequest 中每批事件，PublishResponse 中结果的顺序对应发布请求中事件的顺序，你可以把每个成功/失败结果与对应事件关联。PublishResult 在发布失败时含错误，可重试失败事件。最适合：高吞吐场景——IoT 数据摄取、批量订单处理、实时数据管道。本单元用 PublishStream 发布两个 Order Fulfillment 事件（每个已履行订单一个）。
在 Setup 中定义 Order Fulfillment 事件
定义平台事件的 schema。Setup → Platform Events → New Platform Event，Label 填 Order Fulfillment Event、Plural Label 填 Order Fulfillment Events、勾选 Starts with vowel sound（如可用）、Description 填 Order events contain order data and are received by order processing apps. → Save。然后在 Custom Fields & Relationships 点 New，添加两个字段：Order Id（Text(18)）与 Tracking Number（Text(18)）。
平台事件就像用于消息的自定义对象——持久化、可查询、能触发自动化。schema 定义了事件携带的数据，自定义字段就是 payload 结构。定义后，外部系统即可发布匹配此 schema 的事件。
发布平台事件并在订阅者中接收
完整发布-订阅循环：外部系统 → gRPC 客户端 → Publish/PublishStream → 事件到达 Salesforce → 持久化为平台事件 → 触发任何配置的自动化（Flow、Apex）→ 受管订阅把事件投递给订阅端点 → 订阅者处理并 commit Replay ID。
发布前，先为 Order Fulfillment 事件配置受管订阅：向 /tooling/sobjects/ManagedEventSubscription 发送 POST 创建 ManagedEventSubscription（用 Order Fulfillment 事件的请求体）。然后配置 arguments.yaml（MANAGED_SUB_DEVELOPER_NAME: Managed_Sub_Order_Fulfillment_Event、PROCESS_CHANGE_EVENT_HEADER_FIELDS: false），在 java 文件夹运行 ./run.sh genericpubsub.ManagedSubscribe 启动订阅。
启动受管订阅
激活 Order Fulfillment 事件的订阅：订阅通过 Tooling API 创建后，激活（status 变 Active），订阅开始监听发布的匹配事件。前提：Postman/gRPC 客户端已连接 playground、平台事件 Order_Fulfillment__e 已定义、ManagedEventSubscription 记录已创建。注意创建 ManagedEventSubscription 后新配置在 Pub/Sub API 中生效可能有延迟——若订阅返回"找不到受管订阅配置"的错误，等几分钟再订阅。准备就绪后即可发布测试事件。
发布事件
发布前，先修改示例代码指定要发布的事件字段与其他配置：打开 java/src/main/java/utility/CommonContext.java，获取 Salesforce 用户 ID（点个人头像 → 名字，从个人页 URL 复制以 005 前缀开头的 ID），修改 createEventMessages() 方法、把 CreatedById 的占位符替换为你的用户 ID；在 arguments.yaml 设置 TOPIC: /event/Order_Fulfillment_Event__e、NUMBER_OF_EVENTS_TO_PUBLISH: 2、SINGLE_PUBLISH_REQUEST: true；因改了代码，从 java 文件夹运行 mvn clean install 重建；然后运行 ./run.sh genericpubsub.PublishStream 发布事件。切回订阅者终端窗口，可看到收到的事件。
完整循环完成：发布 → 持久化 → 投递 → 确认。恭喜，你已经端到端学会了 Pub/Sub API、受管订阅，以及如何发布与订阅事件！
关键要点（发布工作流）
完整的发布工作流：1. 在 Setup 定义事件 schema（Order_Fulfillment__e 及自定义字段）；2. 创建受管订阅接收事件；3. 外部系统通过 gRPC 发布事件（Publish 单个或 PublishStream 批量）；4. Salesforce 接收并持久化事件、触发 Flow/Apex 等自动化；5. 订阅者处理并 commit Replay ID。端到端：外部事件 → Salesforce → 自动化动作。
关键要点（整体架构）
Pub/Sub API 是 Salesforce 事件流的现代方式。架构：gRPC + HTTP/2（而非 REST）、Protocol Buffers（而非 JSON）、双向流（而非轮询）。订阅：实时的 Change Data Capture 事件与平台事件、受管订阅（Beta）简化消费、Replay ID 确保可靠可重启的处理。发布：Publish RPC（单事件、逐个确认）与 PublishStream RPC（高吞吐流式发布）、最终确认确认持久化。流量控制：基于拉取，客户端控制消费速率，永不被压垮、永不丢事件。Pub/Sub API 是 Salesforce 平台上事件驱动架构的基础。