salesforce零基础学习（一百零五）Change Data Capture

2023-10-06 08:42:58

本篇参考：

https://developer.salesforce.com/docs/atlas.en-us.232.0.api_streaming.meta/api_streaming/using_streaming_api_durability.htm

https://trailhead.salesforce.com/content/learn/modules/change-data-capture

https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm

https://developer.salesforce.com/blogs/2018/08/what-is-change-data-capture.html

https://developer.salesforce.com/docs/atlas.en-us.232.0.object_reference.meta/object_reference/sforce_api_associated_objects_change_event.htm

salesforce零基础学习（九十六）Platform Event浅谈

salesforce零基础学习（八十五）streaming api 简单使用（接近实时获取你需要跟踪的数据的更新消息状态）

我们在前面介绍过 Push Topic 以及 Platform Event这两个 Streaming API，可以想象未来的某天肯定还会补上一篇 Change Data Capture(CDC) 集齐 Streaming API 三件套，今天这篇博客就是对CDC进行浅入浅出。

一. CDC 概念以及什么时候使用

提起CDC以前，还是需要先重新说一下 Streaming Event 以及 Push technology。 Streaming Event(流事件)是一个系统（发布者）向另一个系统（订阅者）发送的即时通知消息。使用推送技术，发布者将数据推送到订阅者，这个操作近乎实时。对Streaming API感兴趣或者不了解的，可以查看上文的第一个链接。Streaming Event在salesforce中主要有三种封装好的feature： Push Topic & Platform Event & Change Data Capture.

所以什么场景下我们推荐使用 Change Data Capture呢？使用CDC有哪些优势，什么场景不建议呢？

使外部系统与Salesforce数据保持同步;
接收Salesforce记录更改的通知，包括创建、更新、删除和取消删除操作;
可以通过CometD或者Apex Trigger去订阅;
捕获所有记录的字段变更;
无论共享规则如何，订阅者都可以广泛访问所有数据;
订阅者基于Field Level Security，仅传递用户有权访问的字段;
加密 change event字段;
在事件的header中获取有关更改的信息，例如更改的来源等，它可以让订阅方更灵活的判断操作数据;
使用事务边界执行数据更新;
使用版本化的事件架构;
以可扩展的方式订阅大量更改;
访问保留的事件最多三天。

以下场景不适用于使用CDC。

根据记录和字段更改执行审核跟踪。
更改数据捕获旨在保持下游系统的同步，而不是单个用户。如果许多用户订阅了CometD客户机，那么并发客户机限制可能会达到。

二. CDC的发布，结构以及订阅

既然我们知道哪些场景推荐使用，那么我们应该清楚如何去发布，以及发送出去的结构和限制等细节知识，用来做一些评估和下游系统的对接。

发布篇：

setup-> change data capture 然后选择我们需要追踪的表即可。这里我们选择了 Account以及 Opportunity，则Account 以及 Opportunity有CUD情况，下游订阅端便可以获取到消息数据。

结构篇：

那下游端订阅需要了解一下CDC推送过去的格式是什么样，否则他们也没法去进行解析操作，所以我们来看一下CDC推送的消息数据的格式。

通过下图我们看到一个CDC的消息数据结构可以简单的分成两部分： header & body。 header用来记录推送的表的信息，比如表名，操作的类型（CUD/UnDelete），操作的表的ID信息等等。body部分即为改动的字段的信息的键值队。针对新增场景，发送所有的非空的字段以及系统字段；针对更新场景，发送所有改变了的字段；针对删除场景，不会有任何的字段。细节可以查看一下官方的API文档。

这里来做一个引申，如果系统中有formula字段，在新增或者更新场景并不会发送过去，所以针对 formula字段，如果使用了CDC需要考虑进行单独的处理，如果前期未识别，后续会增加很多effort来对应。现在看上面这个截图可能有点懵，等会通过trigger订阅打印出来以后可以更好的了解报文内容。

订阅篇：

salesforce针对CDC支持两种订阅方式： CometD以及Apex Trigger，针对 CometD不做介绍，PushTopic & Platform Event & CDC都是支持CometD订阅方式，如果涉及到下游系统，直接根据官方文档中的demo一步一步配置，很轻松的就可以实现。这里主要是介绍一下 Apex Trigger方式订阅。

要知道，如果我们使用 PushTopic方式订阅，如果下游系统真的没收到数据是一个很麻烦的事情，我们没法去和他们解释到底是salesforce没有广播这条数据，还是订阅端问题，会有适当的扯皮操作，但是使用CDC我们完全不会有这个疑问，因为当广播出去以后，我们可以通过trigger去实现订阅从而实现tracking。不是所有的表都支持CDC，所以我们可以去查看一下官方文档来确定一下。针对支持的表，trigger的写法和 ApexTrigger很相似，区别就是监控的表为 [Object]ChangeEvent。如果是标注你的表，则直接使用名称，比如 AccountChangeEvent。如果是自定义表，则中间需要加上两个下划线，比如 CustomObject__ChangeEvent。并且只允许 after insert使用。下面的例子是针对 OpportunityChangeEvent进行监听，如果是Stage为Close Won情况下，创建一个Task。通过这个trigger，我们也可以打印出来实际的这个结构。

trigger OpportunityChangeTrigger on OpportunityChangeEvent (after insert) {
    List<Task> tasks = new List<Task>();
  // Iterate through each event message.
  for (OpportunityChangeEvent event : Trigger.New) {
    // Get some event header fields
    EventBus.ChangeEventHeader header = event.ChangeEventHeader;
    system.debug('event : ' + JSON.serialize(event));
    if (header.changetype == 'UPDATE') {
        System.debug('List of all changed fields:');
        for (String field : header.changedFields) {
            if (null == event.get(field)) {
                System.debug('Deleted field value (set to null): ' + field);
            } else {
                System.debug('Changed field value: ' + field + '. New Value: '
                    + event.get(field));
            }
        }
    }
    if ((header.changetype=='UPDATE') && (event.isWon==true)) {
       // Create a task 
       Task tk = new Task();
        tk.Subject = 'Follow up on won opportunities: ' + header.recordIds;
        tk.OwnerId = header.CommitUser;
        tasks.add(tk);
    }
  }
  // Insert all tasks in bulk.
  if (tasks.size() > 0) {
    insert tasks;
  }
}

需要知道的是，如果我们希望在debug log中查看到CDC相关的订阅信息，需要将 Traced Entity Type设置成 Automated Process。

我们新建一条 Opportunity的情况下，以下是创建的内容。

可以看一下message 的结构，其中包括了当前的类型，以及变更的字段以及其字段对应的值，通过header我们可以看到这条记录执行的是CREATE的操作。不是所有的字段都展示在这里，只有内容非空的才会在body中。

{
    "IsClosed": false, 
    "attributes": {
        "url": "/services/data/v52.0/sobjects/OpportunityChangeEvent/4881111", 
        "type": "OpportunityChangeEvent"
    }, 
    "IsWon": false, 
    "IsSplit": false, 
    "CloseDate": "2021-07-10", 
    "ReplayId": "4881111", 
    "ForecastCategory": "Pipeline", 
    "Name": "测试CDC 0710", 
    "IsExcludedFromTerritory2Filter": false, 
    "Probability": 10, 
    "OwnerId": "0053g000000lqx1AAA", 
    "LastModifiedDate": "2021-07-10T15:58:07.000+0000", 
    "Id": "1CExx000000KTnbGAG", 
    "HasOpportunityLineItem": false, 
    "StageName": "Prospecting", 
    "IsPrivate": false, 
    "ForecastCategoryName": "Pipeline", 
    "LastModifiedById": "0053g000000lqx1AAA", 
    "CreatedById": "0053g000000lqx1AAA", 
    "ChangeEventHeader": {
        "recordIds": [
            "0063g00000AQHUwAAP"
        ], 
        "nulledFields": [], 
        "diffFields": [], 
        "commitUser": "0053g000000lqx1AAA", 
        "entityName": "Opportunity", 
        "changeType": "CREATE", 
        "commitNumber": 11077894054001, 
        "changedFields": [], 
        "commitTimestamp": 1625932687000, 
        "changeOrigin": "com/salesforce/api/soap/52.0;client=SfdcInternalAPI/", 
        "transactionKey": "00038c39-bc36-b76b-776e-d489f264045f", 
        "sequenceNumber": 1
    }, 
    "CreatedDate": "2021-07-10T15:58:07.000+0000"
}

将这条数据进行update操作，设置amount，并且修改一下Stage Name

我们可以看一下message详情架构如下：body中拥有了 Amount信息以及修改的字段的信息。我们可以看到现在的change Type是 UPDATE。在header的changedFields区域就可以看到这次修改的哪些字段。

{
    "IsClosed": false, 
    "attributes": {
        "url": "/services/data/v52.0/sobjects/OpportunityChangeEvent/4890418", 
        "type": "OpportunityChangeEvent"
    }, 
    "Amount": 101.0, 
    "LastModifiedDate": "2021-07-11T05:40:20.000+0000", 
    "IsSplit": false, 
    "HasOpportunityLineItem": false, 
    "LastAmountChangedHistoryId": "0083g00000KllvXAAR", 
    "IsExcludedFromTerritory2Filter": false, 
    "Id": "1CExx000000KWDiGAO", 
    "ReplayId": "4890418", 
    "StageName": "Needs Analysis", 
    "IsPrivate": false, 
    "LastStageChangeDate": "2021-07-11T05:40:20.000+0000", 
    "IsWon": false, 
    "ChangeEventHeader": {
        "recordIds": [
            "0063g00000AQHUwAAP"
        ], 
        "nulledFields": [], 
        "diffFields": [], 
        "commitUser": "0053g000000lqx1AAA", 
        "entityName": "Opportunity", 
        "changeType": "UPDATE", 
        "commitNumber": 11078205948865, 
        "changedFields": [
            "StageName", 
            "Amount", 
            "ExpectedRevenue", 
            "LastModifiedDate", 
            "LastStageChangeDate", 
            "LastAmountChangedHistoryId"
        ], 
        "commitTimestamp": 1625982020000, 
        "changeOrigin": "com/salesforce/api/soap/52.0;client=SfdcInternalAPI/", 
        "transactionKey": "0003bba5-8a9b-2ad2-c3c3-a15a295b2bb5", 
        "sequenceNumber": 1
    }, 
    "ExpectedRevenue": 20.2
}

三. CDC 、 PushTopic、Platform Event 区别

我们作为开发人员还好，可能架构选型以后，我们了解做了就好。但是哪天我们做到了架构，需要我们选型相关的，我们如何去选型呢？这三个有什么区别或者优缺点，如何去取舍？下图是我们官方的一个比较，详情链接：https://developer.salesforce.com/docs/atlas.en-us.232.0.api_streaming.meta/api_streaming/event_comparison.htm?search_text=Platform%20Event

除了这种 high level的比较，一定要确定好选型以后的细节考虑。比如 PushTopic的query的长度限制，Change Data Capture formula字段变更不会传过去，而且超过指定的数量需要花钱等等。high level决定选型，细节决定了你的effort，缺一不可。

总结：篇中浅入浅出介绍了一下CDC的使用，至此streaming api 广播订阅的三个模型都已经有简单介绍。很多细节介绍也没有展开，比如trigger一次进入数据的数据量必须2000以内等等。如果用到了这个模型，详细查看官方文档进行夯实即可。篇中有错误欢迎指出，有不懂欢迎留言。

码农公寓

相关文章