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

本篇参考：

我们在前面介绍过 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以内等等。如果用到了这个模型，详细查看官方文档进行夯实即可。篇中有错误欢迎指出，有不懂欢迎留言。

本文链接：https://www.cnblogs.com/zero-zyq/p/14984517.html

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

salesforce零基础学习（一百零五）Change Data Capture的更多相关文章

随机推荐

热门专题

目录导航