Salesforce Streaming API：数据同步到实时决策的探索之旅

二月 06, 2026

几年前，我们团队负责维护一个相对复杂的集成场景：需要把 Salesforce 内部的一些核心业务数据，几乎实时地同步到一个外部的数据仓库（DWH），用于后续的报表和分析。在此之前，我们一直采用的是定时批量同步的方式，每隔几个小时跑一个 Job，把 Salesforce 中有变化的记录拉取出来。这种方式在数据量不大的时候勉强能接受，但随着业务的增长和对数据实时性要求的提高，批量同步的缺点就暴露无遗了：数据延迟高、批次间隔期内无法获取最新状态、以及每次都要判断“哪些数据变了”的复杂逻辑。

最初的设想：Streaming API 能否解决我的实时同步问题？

在寻找解决方案时，我自然而然地把目光投向了 Salesforce 的 Streaming API。当时我对它的理解比较简单粗暴：不就是把 Salesforce 的数据变化推给我嘛，这不正好符合我的需求？我的初步想法是，只要 Salesforce 告诉我哪个 Account 或 Contact 变了，我就可以立即去 DWH 更新它。听起来很美好。

Streaming API 下面其实有几种不同的事件类型，当时我主要关注的是两个：PushTopic Events 和 Change Data Capture (CDC) Events。一开始，我被 PushTopic 的 SOQL-like 语法吸引了。

PushTopic Events：简单易用，但很快发现不足

PushTopic 的概念很直观：你定义一个 SOQL 查询，比如 SELECT Id, Name, LastModifiedDate FROM Account WHERE LastModifiedDate > YESTERDAY，然后只要有符合这个查询条件的 Account 记录发生变化，Salesforce 就会推送一个事件。我当时觉得这简直是为我量身定做的！我可以精确控制哪些字段的变化会触发事件。

// 示例 PushTopic 定义
PushTopic pushTopic = new PushTopic();
pushTopic.Name = 'AccountUpdates';
pushTopic.Query = 'SELECT Id, Name, BillingCity FROM Account';
pushTopic.ApiVersion = 59.0;
pushTopic.NotifyForOperationCreate = true;
pushTopic.NotifyForOperationUpdate = true;
pushTopic.NotifyForOperationDelete = true;
pushTopic.NotifyForOperationUndelete = true;
pushTopic.NotifyForFields = 'All'; // 也可以是 'Referenced' 或 'Select'
insert pushTopic;

然而，在实际验证的过程中，我很快发现了一些问题：

字段限制： PushTopic 只能推送 SOQL 查询中明确指定的字段。如果 DWH 需要同步 Account 上的所有字段，或者未来新增了字段，我就需要手动修改 PushTopic 的 Query，这很不灵活，容易遗漏。
Delete 事件： 虽然 PushTopic 支持 NotifyForOperationDelete，但它只推送被删除记录的 ID，而不是完整的记录快照。对于 DWH 来说，我可能还需要知道删除前的一些关键信息。
SOQL 复杂性： 如果业务逻辑要求基于某些复杂条件触发，那么 PushTopic 的 SOQL 可能会变得很复杂，甚至有些条件是 SOQL 无法表达的。
权限问题： 订阅 PushTopic 的用户需要有查询该 SOQL 中所有字段的权限，这在某些集成场景下也需要仔细考虑。

我意识到，PushTopic 更适合那些 UI 更新、或者只需要知道特定字段变化的轻量级通知场景，而不是我这种需要全面、可靠地同步整个对象状态的 DWH 场景。我的目标是“我知道某个 Account 变了，告诉我它现在所有字段长什么样”，而 PushTopic 只能说“这个 Account 的 Name 字段变了”。这显然不够。

转向 Change Data Capture (CDC)：终于找到了“对的”方案

当我深入研究 Streaming API 文档时，Change Data Capture (CDC) 立即引起了我的注意。CDC 的设计哲学完全符合我的需求：它不是基于 SOQL 查询的，而是针对整个 SObject 的。这意味着，只要你为某个 SObject 启用了 CDC，那么该 SObject 上的任何创建、更新、删除、取消删除操作，都会触发一个事件，并且事件负载会包含所有发生变化的字段值。

CDC 的优势：全面且灵活

全字段追踪： CDC 会将发生变化的字段及其新值包含在事件负载中。如果 DWH 需要同步一个对象的全貌，我只需要在 DWH 端根据这个 ID 去 Salesforce 再拉取一次完整的记录（或直接使用 CDC 事件中提供的所有字段）。这比 PushTopic 的字段限制要灵活得多。
操作类型明确： CDC 事件会明确指出是 CREATE, UPDATE, DELETE 还是 UNDELETE 操作。这对于 DWH 进行相应的增、改、删操作非常方便。
更丰富的元数据： CDC 事件还包含了 ChangeEventHeader，里面有 commitNumber, commitTimestamp, sequenceNumber 等关键信息，特别是 ReplayId，这对于保证事件的可靠传输和处理至关重要。

要启用 CDC，只需在 Salesforce Setup 中找到 "Change Data Capture"，然后把需要追踪的对象（比如 Account, Contact）添加到 "Selected Entities" 列表中即可。配置简单，无需编写代码。

// 示例 CDC 事件负载结构（简化）
{
  "schema": "s_AccountChangeEvent",
  "payload": {
    "ChangeEventHeader": {
      "entityName": "Account",
      "recordIds": ["001xxxxxxxxxxxxxxx"],
      "changeType": "UPDATE",
      "changeOrigin": "com.salesforce.api.soap",
      "commitTimestamp": 1678886400000,
      "commitNumber": 123456789,
      "sequenceNumber": 1,
      "isReplay": false,
      "replayId": 1234567890123456
    },
    "Id": "001xxxxxxxxxxxxxxx",
    "Name": "Updated Account Name",
    "BillingCity": "San Francisco",
    "LastModifiedDate": "2023-03-15T08:00:00Z",
    "ChangeEventHeader_ChangedFields": ["Name", "BillingCity"] // 实际负载会包含所有变动的字段
  },
  "event": {
    "replayId": 1234567890123456
  }
}

选择 CDC 后，我面临的下一个核心问题是如何保证事件的可靠消费。如果我的外部系统挂掉了，或者网络中断了，我怎么知道我错过了哪些事件？或者我怎么避免重复处理事件？

ReplayId：实现可靠事件消费的关键

CDC 事件负载中的 ReplayId 字段，就是解决这个问题的关键。ReplayId 是一个不透明的、按时间顺序递增的数字，Salesforce 为每个发布的事件分配一个唯一的 ReplayId。这个 ID 确保了事件的顺序性和可回溯性。

我的解决方案：

存储 ReplayId： 外部系统在成功处理完一个事件后，必须将其 ReplayId 持久化存储起来。这个存储可以是数据库、文件系统，甚至是 Kafka 的 offset。
断点续传： 当外部系统启动或从故障中恢复时，它会从存储中读取上一次成功处理的 ReplayId。然后，它向 Salesforce Streaming API 发起订阅请求时，指定从这个 ReplayId 之后开始接收事件。

这样，即使外部系统宕机，当它重新上线时，也能从上次中断的地方继续消费事件，确保不会遗漏任何数据。Salesforce 会在订阅请求中，根据提供的 ReplayId 找到对应的事件流位置，然后推送后续的事件。

ReplayId 的局限性：重放窗口

但这里有一个重要的限制：Salesforce 并不是无限期地保留所有历史事件。它有一个“重放窗口”，通常是 24 到 72 小时（具体取决于 Salesforce 的版本和事件类型）。这意味着，如果我的外部系统宕机超过了这个窗口期，那么即使我提供了上次的 ReplayId，Salesforce 也无法回溯到那么久远的事件了。这时候，我就必须采取一个回溯策略：

全量同步： 发生长时间中断后，可能需要回退到传统的全量同步方式，将所有数据重新拉取一遍，以填补缺失。
定期快照： 配合定期的数据快照，可以缩小需要回溯的数据范围。

对于我们的 DWH 场景，这意味着我们需要设计一个健壮的错误处理机制，以及一个当重放窗口失效时能触发全量同步的机制。虽然有点复杂，但这是保证数据一致性的必要牺牲。

订阅机制的选择：CometD 到 Pub/Sub API 的演进

在连接 Salesforce Streaming API 时，我最早接触的是基于 CometD 协议的客户端库。CometD 本质上是一种基于 HTTP 长轮询的模拟双向通信协议，很多语言都有相应的客户端实现，比如 Bayeux 客户端。我们最初使用 Java 编写了一个订阅客户端，通过 CometD 连接到 Salesforce。

// 伪代码：Java CometD 客户端订阅
BayeuxClient client = new BayeuxClient(...);
client.handshake();
client.waitFor(1000, BayeuxClient.State.CONNECTED);

client.getChannel("/data/ChangeEvents").subscribe(new ClientSessionChannel.MessageListener() {
    @Override
    public void onMessage(ClientSessionChannel channel, Message message) {
        // 处理CDC事件
        String replayId = (String) message.get("event").get("replayId");
        Map<String, Object> payload = (Map<String, Object>) message.get("data").get("payload");
        // ... 存储replayId，处理数据
    }
}, new MessageListener() {
    @Override
    public void onMessage(ClientSessionChannel channel, Message message) {
        // 订阅成功/失败通知
    }
}, String.valueOf(lastProcessedReplayId)); // 指定从哪个replayId开始订阅

这种方式运行得不错，但随着 Salesforce 推出了 Pub/Sub API，我开始关注它的潜在优势。

Pub/Sub API (gRPC)：新一代的集成方式

Pub/Sub API 是 Salesforce 基于 gRPC 实现的 Streaming API 接口，它提供了更高效、更现代的客户端-服务器通信方式。相比于 CometD 的长轮询，gRPC 采用了 HTTP/2 协议，支持双向流，通常在性能和资源利用上更有优势。

更低延迟： gRPC 协议通常能提供更低的事件传输延迟。
连接效率： HTTP/2 的多路复用特性减少了连接开销。
类型安全： gRPC 使用 Protobuf 定义消息结构，提供了更好的类型安全和跨语言兼容性。

对于我们团队来说，如果未来有新的实时集成项目，Pub/Sub API 会是首选。但对于已经稳定运行的 CometD 客户端，迁移的成本和收益需要仔细权衡。目前我们还没有迫切的理由进行迁移，但会密切关注其发展和最佳实践。

事件交付限制

无论是 CometD 还是 Pub/Sub API，都需要注意 Streaming API 的 governor limits。例如，每小时的事件交付数量是有限制的。如果 Salesforce 内部发生了大规模的批量更新，一次性产生巨量的 CDC 事件，是有可能触及这些限制的。届时，事件可能会被延迟投递，甚至在极端情况下被丢弃（虽然有 ReplayId，但在系统级别达到上限时，总会有些挑战）。

所以，我们的外部系统设计时，也需要考虑限流和熔断，不能假设 Salesforce 会无限量地推送事件。同时，在 Salesforce 内部，如果可以，尽量避免一次性对大量记录进行高频次的更新，或者将这些操作设计成批处理，并辅以恰当的错误处理和重试机制。

Platform Events：另一种实时选择

在 Streaming API 的家族中，除了 PushTopic 和 CDC，还有 Platform Events。虽然它不是我解决 DWH 同步问题的主力，但我在其他场景下也用过它，值得顺带一提。

Platform Events 就像一个自定义的消息队列，它允许你定义自己的事件对象（类似 SObject），并发布/订阅这些事件。它和 CDC 的主要区别在于：

自定义事件： Platform Events 适用于业务流程中的“事件”概念，例如“订单已支付”、“审批已完成”等，这些事件不一定直接对应某个 SObject 的 CRUD 操作。
解耦： 它是松耦合集成的利器。一个触发器发布事件，多个订阅者可以独立消费。
灵活的 payload： 你可以完全自定义事件的字段。

我曾用 Platform Events 来协调跨系统的业务流程。例如，当 Salesforce 中某个任务状态改变后，我通过 Platform Event 通知另一个外部系统去执行下一步操作，而不是直接调用外部系统的 API。这样可以提高系统的弹性和可伸缩性。

所以，我的经验是：CDC 适用于记录级别的状态同步和数据复制；Platform Events 适用于业务流程级别的事件驱动和解耦集成。 两者各有侧重，可以协同使用。

总结与展望

我的 Streaming API 之旅，从最初对 PushTopic 的误解，到后来拥抱 CDC 的全面性，再到深入理解 ReplayId 的可靠性，以及对 Pub/Sub API 的关注，是一个不断学习和调整的过程。Salesforce Streaming API 确实是一个非常强大的工具，它极大地改变了我们处理实时数据同步和集成的方式。

现在，我们基于 CDC 的 DWH 同步方案运行稳定，极大地提高了数据的新鲜度，为业务决策提供了更及时的支撑。但仍有一些挑战需要持续关注：

规模化问题： 随着 Salesforce 数据量的持续增长，以及外部消费者数量的增加，如何高效地管理订阅连接和事件吞吐量，仍是一个需要不断优化的问题。
错误处理与回溯： 虽然 ReplayId 解决了大部分可靠性问题，但长时间中断后的全量回溯仍然是相对重型且需要人工干预的，未来希望能找到更优雅的解决方案。
监控与告警： 建立健壮的监控系统，及时发现事件延迟、订阅中断或超出限额的情况，是保证系统稳定运行不可或缺的一部分。

总的来说，Streaming API 让我有机会以更现代、更实时的方式思考 Salesforce 的集成策略。它不是一个“即插即用”的银弹，但如果理解了其背后的机制和限制，并加以精心设计，它能为你的 Salesforce 生态系统带来巨大的价值。

Salesforce中文网