利用 Salesforce Streaming API 实现实时数据集成：深度解析

背景与应用场景

大家好，我是一名 Salesforce 集成工程师。在我的日常工作中，核心任务之一就是确保 Salesforce 与企业内的其他系统（如 ERP、数据仓库、移动应用）之间的数据能够高效、准确地同步。传统的集成方式，如周期性轮询 (Polling)，即每隔几分钟调用一次 Salesforce API 来检查数据是否有更新，存在着明显的弊端：

1. 延迟性： 数据同步不是实时的，用户在外部系统中看到的数据可能已经过时。

2. API 消耗： 无论数据是否变化，轮询都会消耗宝贵的 API 调用次数，这在大型企业中会迅速达到 Salesforce 的每日 API 限制。

3. 资源浪费： 频繁的轮询对客户端和服务器都造成了不必要的负载。

为了解决这些问题，Salesforce 提供了强大的 Streaming API。它基于发布-订阅 (Publish-Subscribe) 模型，允许外部客户端订阅 Salesforce 中的事件。一旦 Salesforce 中发生特定事件（例如，一条客户记录被更新），Salesforce 会主动将通知推送给所有已订阅的客户端。这种“推”模型 (Push Technology) 彻底改变了集成的范式，实现了真正的近实时数据同步。

作为集成工程师，我经常在以下场景中应用 Streaming API：

实时仪表盘： 将销售机会 (Opportunity) 的状态变更实时推送到外部 BI 系统的仪表盘上，让销售总监可以即时看到团队业绩的变化。
系统间数据同步： 当一个订单在 Salesforce 中被标记为“已发货”时，立即通知 ERP 系统更新库存和物流信息。
即时通知： 当一个高优先级的工单 (Case) 被创建时，通过集成将通知推送到团队的 Slack 或 Microsoft Teams 频道中。
数据仓库更新： 将 Salesforce 核心对象（如 Account, Contact）的增删改变化实时地流式传输到 Snowflake 或 Redshift 等云数据仓库，避免了传统 ETL 的高延迟。

原理说明

Salesforce Streaming API 并非单一技术，而是一个事件流框架的总称。其底层技术基于 Bayeux protocol，这是一种用于在 Web 客户端和服务器之间传输异步消息的协议。而 CometD 是 Bayeux 协议的一个实现，它通过一种称为“长轮询” (Long Polling) 的技术来模拟服务器向客户端的推送。客户端发起一个请求，服务器会保持该连接打开，直到有事件发生或超时。一旦有事件，服务器立即响应，客户端处理后马上发起下一个长轮询请求。这样就实现了低延迟的消息传递。

Streaming API 提供了多种事件类型，以满足不同的业务需求：

1. PushTopic Events

这是最经典的流式事件。你需要定义一个 PushTopic，它本质上是一个 SOQL 查询语句。当任何记录的创建、更新、删除或反删除操作满足该 SOQL 查询的条件时，Salesforce 就会生成一个通知。例如，你可以创建一个 PushTopic 来监控所有金额超过 100 万美元的销售机会。

2. Change Data Capture (CDC，变更数据捕获)

这是我个人在现代集成项目中首推的事件类型。与 PushTopic 不同，CDC 无需定义 SOQL 查询。你只需在设置中为某个对象（标准或自定义对象）启用 CDC。一旦启用，该对象上所有记录的任何字段变更都会生成一个详细的事件。CDC 的事件体 (Payload) 包含了丰富的元数据，例如变更类型（CREATE, UPDATE, DELETE）、变更的字段以及事务相关的信息。这使得它非常适合用于数据复制和审计场景。

3. Platform Events (平台事件)

平台事件是完全自定义的。你可以像定义一个自定义对象一样，定义一个平台事件的结构（即包含哪些字段）。它遵循事件驱动架构 (Event-Driven Architecture) 的思想。你可以通过 Apex、Flow、或 API 来发布平台事件。这非常适合用于解耦系统间的业务流程。例如，当一个复杂的业务流程在 Salesforce 中完成时，可以发布一个“流程已完成”的平台事件，其他系统订阅该事件后即可触发各自的下游任务。

4. Generic Events (通用事件)

这是一种较旧的机制，允许你将任意字符串消息发布到一个流式通道。由于其功能有限且已被平台事件取代，现在已不推荐使用。

示例代码

作为集成工程师，我们不仅要在 Salesforce 侧进行配置，更要关注客户端如何消费这些事件。以下是一些来自 Salesforce 官方文档的代码示例，展示了如何创建事件源和如何从外部客户端进行订阅。

1. 使用 Apex 创建一个 PushTopic

此代码段演示了如何通过 Apex 创建一个 PushTopic，用于监控客户 (Account) 记录的创建和更新，前提是这些客户的员工数量 (NumberOfEmployees) 大于 1000。

PushTopic pushTopic = new PushTopic();
pushTopic.Name = 'LargeAccounts';
pushTopic.Query = 'SELECT Id, Name, NumberOfEmployees FROM Account WHERE NumberOfEmployees > 1000';
pushTopic.ApiVersion = 58.0;
// NotifyOnFields specifies which record operations trigger a notification.
// In this case, it's for create and update operations.
pushTopic.NotifyForOperationCreate = true;
pushTopic.NotifyForOperationUpdate = true;
pushTopic.NotifyForOperationUndelete = false;
pushTopic.NotifyForOperationDelete = false;
// For updates, a notification is generated only when field values are changed (NotifyForFields is set to 'Referenced').
pushTopic.NotifyForFields = 'Referenced';

insert pushTopic;

注释：`NotifyForFields = 'Referenced'` 是一个关键设置，它表示只有当 SOQL 查询中引用的字段（Id, Name, NumberOfEmployees）的值发生变化时，才会针对更新操作发送通知，这可以有效减少不必要的事件推送。

2. 定义和发布一个平台事件

首先，我们需要在 Salesforce 中通过 UI 或 Metadata API 定义一个名为 `Cloud_News__e` 的平台事件。假设它有两个自定义字段：`Location__c` (Text) 和 `Urgent__c` (Checkbox)。

然后，我们可以使用 Apex 来发布这个事件。这通常发生在某个业务逻辑触发点，比如一个 Apex Trigger 或一个 LWC 控制器中。

// Create a list of events to publish
List<Cloud_News__e> newsEvents = new List<Cloud_News__e>();
newsEvents.add(new Cloud_News__e(
    Location__c='West',
    Urgent__c=true,
    News_Content__c='Major service update in the West region.'
));
newsEvents.add(new Cloud_News__e(
    Location__c='East',
    Urgent__c=false,
    News_Content__c='Minor service update in the East region.'
));

// Call the EventBus.publish method to publish the events
List<Database.SaveResult> results = EventBus.publish(newsEvents);

// Inspect the results
for (Database.SaveResult sr : results) {
    if (sr.isSuccess()) {
        System.debug('Successfully published event.');
    } else {
        for(Database.Error err : sr.getErrors()) {
            System.debug('Error returned: ' +
                        err.getStatusCode() +
                        ' - ' +
                        err.getMessage());
        }
    }
}

注释：`EventBus.publish()` 方法是异步执行的，它将事件放入队列等待发布。返回的 `SaveResult` 可以让你检查发布操作是否成功提交到队列中，而不是事件是否已被消费。

3. 使用 Java 客户端 (CometD) 订阅事件

这是集成工程师最关心的部分——外部系统如何连接。这个官方示例展示了如何使用 Java 和 CometD 库来订阅一个通道，例如 PushTopic 的 `/topic/LargeAccounts` 或 CDC 的 `/data/AccountChangeEvent`。

// This example is conceptual and relies on a CometD library like org.cometd.client
// You would need to set up a Maven/Gradle project with the necessary dependencies.

// Step 1: Set up authentication and get a session ID
// (Code for making an OAuth 2.0 call to Salesforce to get an access token is omitted for brevity)
String sessionId = "YOUR_SALESFORCE_SESSION_ID";
String instanceUrl = "https://your-instance.my.salesforce.com";

// Step 2: Configure the BayeuxClient (CometD client)
BayeuxClient client;
String channel = "/topic/LargeAccounts"; // The channel to subscribe to

// Transport configuration
ClientTransport transport = new LongPollingTransport(null, httpClient);
client = new BayeuxClient(instanceUrl + "/cometd/58.0", transport);

// Add the session ID to the Authorization header for authentication
client.addExtension(new ClientSession.Extension() {
    @Override
    public boolean sendMeta(ClientSession session, Mutable message) {
        message.getExt(true).put("oauthToken", sessionId);
        return true;
    }
});

// Step 3: Handshake with the server
client.handshake(handshakeReply -> {
    if (handshakeReply.isSuccessful()) {
        // Step 4: Subscribe to the channel after a successful handshake
        client.getChannel(channel).subscribe((c, message) -> {
            // This is where you process the incoming event message
            System.out.println("Received event: " + message.getDataAsMap());
        });
    } else {
        System.err.println("Handshake failed: " + handshakeReply.toString());
    }
});

注释：这是一个简化的示例。一个生产级的客户端还需要实现完整的 OAuth 2.0 认证流程来获取 `sessionId`，以及健壮的重连逻辑和错误处理机制。`message.getDataAsMap()` 将会返回事件的具体内容，包括 `sobject` 数据和 `eventId` 等信息。

注意事项

在设计和实施基于 Streaming API 的集成方案时，必须考虑以下几点：

1. 权限与可见性： 订阅事件的用户必须拥有 "API Enabled" 权限。对于 PushTopic 和 CDC 事件，用户还需要对相关的对象和字段拥有读取权限。如果用户看不到某个字段，那么该字段的变更也不会出现在事件通知中。

2. API 与事件交付限制： Salesforce 对事件的发布和交付有明确的限制。例如，24 小时内可以交付的事件数量是有限的（具体数量取决于你的 Salesforce 版本和附加许可）。作为集成工程师，你需要仔细评估业务的事件量，并在 Salesforce Setup 的“公司信息”页面监控“每日流式 API 事件”的使用情况，防止超出限制。

3. 事件的持久性与 ReplayId： Salesforce 的事件总线 (Event Bus) 会将事件保留一段时间（通常是 24 到 72 小时）。每个事件都有一个唯一的、递增的 ReplayId。这是一个至关重要的概念！你的客户端应该在每次成功处理事件后，持久化存储最后收到的 `ReplayId`。如果客户端因任何原因（如网络中断、重启）断开连接，它在重新订阅时可以指定这个 `ReplayId`，Salesforce 会从该 ID 之后的所有错过的事件开始重新推送。这保证了事件的“至少一次”交付，防止数据丢失。

4. 选择正确的事件类型：

如果你的需求是基于复杂的记录条件（类似 SOQL 的 `WHERE` 子句），并且只需要部分字段，PushTopic 是个不错的选择。
如果你需要获取一个对象上所有记录、所有字段的详细变更信息，用于数据复制或审计，Change Data Capture (CDC) 是更现代、更强大的选择。
如果你需要的是解耦系统间的业务流程，触发与 Salesforce 数据记录不直接相关的逻辑，那么平台事件是最佳方案。

总结与最佳实践

对于我们集成工程师来说，Salesforce Streaming API 是一个强大的工具，它让我们能够构建响应迅速、高效且可扩展的集成解决方案，彻底摆脱了传统轮询模式的束缚。

最佳实践总结：

优先考虑 CDC： 对于数据同步场景，优先选择 Change Data Capture，因为它提供了更丰富、更标准的变更事件格式。
设计弹性的客户端： 你的订阅客户端必须是“有状态”的。它需要妥善处理连接、认证、订阅、断线重连，并最重要地，正确管理 `ReplayId` 以保证数据完整性。
监控使用情况： 密切关注你的事件交付限制。如果预计事件量巨大，考虑购买 Salesforce 的附加产品，如 "High Volume Platform Events"。
使用共享通道： 如果多个客户端需要订阅相同的事件，让它们都订阅同一个通道，而不是为每个客户端创建独立的 PushTopic，这样可以节约资源。
不要在事件中包含大量数据： 事件通知本身应该是轻量级的。如果需要获取完整的记录数据，可以在收到通知后，使用事件中提供的记录 ID，通过 REST API 回调 Salesforce 获取最新数据。这种模式称为“声明变更，而非状态” (Claim Check Pattern)。

通过遵循这些原则，你可以充分利用 Streaming API 的强大功能，构建出稳定、可靠、实时的企业级集成架构。

精通 Salesforce Email Studio：咨询顾问指南之 AMPscript 与数据扩展实现动态个性化邮件

十月 07, 2025

阅读全文

Salesforce中文网