用于MongoDB的Debezium连接器

Debezium 的 MongoDB 连接器跟踪 MongoDB 副本集或 MongoDB 分片集群，以获取数据库和集合中的文档更改，并将这些更改记录为 Kafka 主题中的事件。 连接器会自动处理分片集群中分片的添加或删除、每个副本集成员身份的更改、每个副本集中的选举以及等待通信问题解决。

有关与此连接器兼容的 MongoDB 版本的信息，请参阅 Debezium 版本概述。

概述

MongoDB的复制机制提供了冗余和高可用性，是在生产环境中运行MongoDB的首选方式。 MongoDB 连接器捕获副本集或分片集群中的更改。

MongoDB 副本集由一组服务器组成，这些服务器都具有相同数据的副本，复制可确保客户端对副本集主服务器上的文档所做的所有更改都正确应用于其他副本集的服务器，称为辅助副本。 MongoDB复制的工作原理是让主节点在其oplog（或操作日志）中记录更改，然后每个辅助节点读取主节点的oplog，并按顺序将所有操作应用于自己的文档。 将新服务器添加到副本集时，该服务器首先对主服务器上的所有数据库和集合执行快照，然后读取主服务器的 oplog 以应用自启动快照以来可能所做的所有更改。 这个新服务器在赶上主服务器oplog的尾部时成为辅助服务器（并且能够处理查询）。

更改流

虽然Debezium MongoDB连接器不会成为副本集的一部分，但它使用类似的复制机制来获取oplog数据。 主要区别在于连接器不直接读取操作日志。 相反，它将oplog数据的捕获和解码委托给MongoDB更改流功能。 使用更改流，MongoDB服务器将集合中发生的更改公开为事件流。 Debezium 连接器监视流，然后将更改传递到下游。 连接器首次检测到副本集时，它会检查 oplog 以获取上次记录的事务，然后执行主数据库和集合的快照。 连接器完成复制数据后，它会从之前读取的 oplog 位置开始创建一个更改流。

当MongoDB连接器进程发生变化时，它会定期记录事件在oplog流中发起的位置。 当连接器停止时，它会记录它处理的最后一个 oplog 流位置，以便在重新启动后可以从该位置恢复流。 换句话说，连接器可以停止、升级或维护，并在一段时间后重新启动，并且始终准确地从中断的位置继续，而不会丢失单个事件。 当然，MongoDBoplog通常被限制在最大大小，因此如果连接器长时间停止，则oplog中的操作可能会在连接器有机会读取它们之前被清除。 在这种情况下，重新启动后，连接器会检测缺少的oplog操作，执行快照，然后继续流式传输更改。

MongoDB连接器还非常容忍副本集的成员资格和领导权的变化，分片集群内分片的添加或删除以及可能导致通信故障的网络问题。 连接器始终使用副本集的主节点流式传输更改，因此当副本集经过选举并且其他节点成为主节点时，连接器将立即停止流式传输更改，连接到新的主节点，并使用新的主节点开始流式传输更改。 同样，如果连接器无法与副本集主节点通信，它将尝试重新连接（使用指数退避，以免使网络或副本集不堪重负）。 重新建立连接后，连接器将继续流式传输它捕获的最后一个事件的更改。 通过这种方式，连接器会根据副本集成员身份的变化动态调整，并自动处理通信中断。

其他资源

• 复制机制

• 副本集

• 副本集选举

• 分片集群

• 分片添加

• 分片删除

• 更改流

MongoDB连接器的工作原理

连接器支持的 MongoDB 拓扑概述对于规划应用程序非常有用。

配置和部署 MongoDB 连接器时，它首先连接到种子地址的 MongoDB 服务器，并确定有关每个可用副本集的详细信息。 由于每个副本集都有自己独立的 oplog，因此连接器将尝试为每个副本集使用单独的任务。 连接器可以限制它将使用的最大任务数，如果没有足够的任务可用，连接器将为每个任务分配多个副本集，尽管该任务仍将为每个副本集使用单独的线程。

支持的 MongoDB 拓扑

MongoDB连接器支持以下MongoDB拓扑：

MongoDB 副本集

Debezium MongoDB连接器可以从单个MongoDB副本集中捕获更改。 生产副本集至少需要三个成员。

若要将 MongoDB 连接器与副本集一起使用，必须将连接器配置中的属性值设置为副本集连接字符串。 当连接器准备好开始从 MongoDB 更改流捕获更改时，它会启动连接任务。 然后，连接任务使用指定的连接字符串建立与可用副本集成员的连接。mongodb.connection.string

MongoDB分片集群

MongoDB分片集群包括：

• 一个或多个分片，每个分片部署为副本集;

• 充当群集配置服务器的单独副本集

• 一个或多个路由器（也称为），客户端连接到这些路由器并将请求路由到相应的分片mongos

  若要将 MongoDB 连接器与分片群集一起使用，请在连接器配置中将属性的值设置为分片群集连接字符串。mongodb.connection.string

MongoDB独立服务器

MongoDB连接器无法监控独立MongoDB服务器的更改，因为独立服务器没有oplog。 如果将独立服务器转换为具有一个成员的副本集，则连接器将正常工作。

逻辑连接器名称

连接器配置属性用作 MongoDB 副本集或分片集群的逻辑名称。 连接器以多种方式使用逻辑名称：作为所有主题名称的前缀，以及在记录每个副本集的更改流位置时作为唯一标识符。topic.prefix

您应该为每个MongoDB连接器提供一个唯一的逻辑名称，以有意义地描述源MongoDB系统。 我们建议逻辑名称以字母或下划线字符开头，其余字符为字母数字或下划线。

执行快照

当任务使用副本集启动时，它使用连接器的逻辑名称和副本集名称来查找描述连接器以前停止读取更改的位置的偏移量。 如果可以找到偏移量并且它仍然存在于oplog中，则任务会立即从记录的偏移位置开始进行流式更改。

但是，如果未找到偏移量或 oplog 不再包含该位置，则任务必须首先通过执行快照来获取副本集内容的当前状态。 此过程首先记录oplog的当前位置并将其记录为偏移量（以及表示快照已启动的标志）。 然后，该任务将继续复制每个集合，生成尽可能多的线程（最多为配置属性的值）以并行执行此工作。 连接器将为它看到的每个文档记录一个单独的读取事件，该读取事件将包含对象的标识符、对象的完整状态以及有关找到对象的 MongoDB 副本集的源信息。 源信息还将包括一个标志，该标志表示事件是在快照期间生成的。snapshot.max.threads

此快照将继续，直到它复制了与连接器筛选器匹配的所有集合。 如果在任务快照完成之前停止连接器，则在重新启动时，连接器将再次启动快照。

临时快照

默认情况下，连接器仅在首次启动后运行初始快照操作。 在此初始快照之后，在正常情况下，连接器不会重复快照过程。 连接器捕获的任何未来更改事件数据仅通过流式处理过程传入。

但是，在某些情况下，连接器在初始快照期间获取的数据可能会过时、丢失或不完整。 为了提供重新捕获集合数据的机制，Debezium 包含一个执行临时快照的选项。 数据库中的以下更改可能是执行临时快照的原因：

• 修改连接器配置以捕获一组不同的集合。

• Kafka 主题将被删除，必须重新生成。

• 由于配置错误或其他问题，会发生数据损坏。

您可以通过启动所谓的临时快照，为之前为其捕获快照的集合重新运行快照。 临时快照需要使用信令收集。 您可以通过向 Debezium 信令集合发送信号请求来启动临时快照。

启动现有集合的即席快照时，连接器会将内容追加到集合已存在的主题。 如果删除了以前存在的主题，如果启用了自动主题创建，Debezium 可以自动创建主题。

即席快照信号指定要包含在快照中的集合。 快照可以捕获数据库的全部内容，也可以仅捕获数据库中集合的子集。

您可以通过向信令集合发送消息来指定要捕获的集合。 将信号类型设置为 ，并提供要包含在快照中的集合的名称，如下表所述：execute-snapshotexecute-snapshotincremental

触发临时快照

您可以通过将具有信号类型的条目添加到信令集合来启动即席快照。 连接器处理邮件后，将开始快照操作。 快照过程读取第一个和最后一个主键值，并将这些值用作每个集合的起点和终点。 根据集合中的条目数和配置的块大小，Debezium 将集合划分为块，并继续连续地为每个块创建快照，一次一个。execute-snapshot

目前，操作类型仅触发增量快照。 更多信息，请参见增量快照。execute-snapshot

增量快照

为了提供管理快照的灵活性，Debezium 包含一个补充快照机制，称为增量快照。 增量快照依赖于 Debezium 机制将信号发送到 Debezium 连接器。 增量快照基于 DDD-3 设计文档。

在增量快照中，Debezium 不是像初始快照那样一次捕获数据库的完整状态，而是分阶段捕获一系列可配置块中的每个集合。 您可以指定希望快照捕获的集合以及每个区块的大小。 区块大小决定了快照在数据库上的每次读取操作期间收集的行数。 增量快照的默认区块大小为 1024 行。

随着增量快照的进行，Debezium 使用水印来跟踪其进度，并维护其捕获的每个收集行的记录。 与标准的初始快照过程相比，这种分阶段捕获数据的方法具有以下优势：

• 您可以在流式数据捕获的同时并行运行增量快照，而不是将流式传输到快照完成。 在整个快照过程中，连接器继续从更改日志中捕获近乎实时的事件，并且两个操作都不会阻止另一个操作。

• 如果增量快照的进度中断，您可以恢复它而不会丢失任何数据。 该过程恢复后，快照将从停止的位置开始，而不是从头开始重新捕获集合。

• 您可以随时按需运行增量快照，并根据需要重复该过程以适应数据库更新。 例如，在修改连接器配置以将集合添加到其 collection.include.list 属性后，可以重新运行快照。

增量快照过程

运行增量快照时，Debezium 会按主键对每个集合进行排序，然后根据配置的块大小将集合拆分为块。 它逐块工作，然后捕获块中的每个集合行。 对于它捕获的每一行，快照都会发出一个事件。 该事件表示区块快照开始时行的值。READ

随着快照的进行，其他进程可能会继续访问数据库，从而可能修改集合记录。 为了反映此类更改，将按常规将 、 或 操作提交到事务日志中。 同样，正在进行的 Debezium 流进程会继续检测这些变更事件，并向 Kafka 发送相应的变更事件记录。INSERTUPDATEDELETE

Debezium 如何解决具有相同主键的记录之间的冲突

在某些情况式处理发出的 or 事件是按顺序接收的。 也就是说，流式处理过程可能会发出一个事件，该事件在快照捕获包含该行的事件的区块之前修改收集行。 当快照最终发出行的相应事件时，其值已被取代。 为了确保以正确的逻辑顺序处理不按顺序到达的增量快照事件，Debezium 采用缓冲方案来解决冲突。 只有在解决了快照事件和流事件之间的冲突后，Debezium 才会向 Kafka 发出事件记录。UPDATEDELETEREADREAD

快照窗口

为了帮助解决延迟到达事件和修改同一集合行的流事件之间的冲突，Debezium 采用了所谓的快照窗口。 快照窗口划分增量快照捕获指定集合区块的数据的时间间隔。 在块的快照窗口打开之前，Debezium 遵循其常规行为，并从事务日志中直接向目标 Kafka 主题下游发出事件。 但是，从特定块的快照打开的那一刻起，直到它关闭，Debezium 都会执行重复数据消除步骤来解决具有相同主键的事件之间的冲突。READ

对于每个数据收集，Debezium 都会发出两种类型的事件，并将这两种事件的记录存储在单个目标 Kafka 主题中。 它直接从表中捕获的快照记录作为操作发出。 同时，随着用户不断更新数据收集中的记录，并且事务日志会更新以反映每次提交，Debezium 会针对每次更改发出或操作。READUPDATEDELETE

当快照窗口打开并且 Debezium 开始处理快照块时，它会将快照记录传送到内存缓冲区。 在快照窗口期间，缓冲区中事件的主键将与传入流事件的主键进行比较。 如果未找到匹配项，则流事件记录将直接发送到 Kafka。 如果 Debezium 检测到匹配项，它将丢弃缓冲事件，并将流记录写入目标主题，因为流事件在逻辑上取代静态快照事件。 区块的快照窗口关闭后，缓冲区仅包含不存在相关事务日志事件的事件。 Debezium 将这些剩余事件发送到集合的 Kafka 主题。READREADREADREAD

连接器对每个快照区块重复该过程。

触发增量快照

目前，启动增量快照的唯一方法是将临时快照信号发送到源数据库上的信令集合。

通过使用 MongoDB 方法将信号提交到信令集合。insert()

Debezium 检测到信令集合中的更改后，它会读取信号，并运行请求的快照操作。

您提交的查询指定要包含在快照中的集合，并选择性地指定快照操作的类型。 目前，快照操作的唯一有效选项是默认值 。incremental

要指定要包含在快照中的集合，请提供一个列出集合的数组或用于匹配集合的正则表达式数组，例如，data-collections
{"data-collections": ["public.Collection1", "public.Collection2"]}

增量快照信号的数组没有默认值。 如果阵列为空，Debezium 会检测到不需要执行任何操作，并且不会执行快照。data-collectionsdata-collections

先决条件

• 信令已启用。
  

  • 源数据库上存在信令数据收集。

  • 信令数据收集在 signal.data.collection 属性中指定。

使用源信令通道触发增量快照

1. 将快照信号文档插入信令集合：

   <signalDataCollection>.insert({"id" : _<idNumber>,"type" : <snapshotType>, "data" : {"data-collections" ["<collectionName>", "<collectionName>"],"type": <snapshotType>}});

   例如

   db.debeziumSignal.insert({ 
   "type" : "execute-snapshot",  
   "data" : {
   "data-collections" ["\"public\".\"Collection1\"", "\"public\".\"Collection2\""], 
   "type": "incremental"} 
   });

   命令中 、 和参数的值对应于信令集合的字段。idtypedata

   示例中的参数说明如下表所示：

   表 2.MongoDB insert（） 命令中用于向信令集合发送增量快照信号的字段描述

   项目	价值	描述
   1	db.debeziumSignal	指定源数据库上信令集合的完全限定名称。
   2	零	该参数指定分配为信号请求标识符的任意字符串。 前面示例中的 insert 方法省略了可选参数的使用。 由于文档没有显式为参数分配值，因此MongoDB自动分配给文档的任意id将成为信号请求的标识符。 使用此字符串标识将消息记录到信令集合中的条目。 Debezium 不使用此标识符字符串。 相反，在快照期间，Debezium 会生成自己的字符串作为水印信号。_idid_ididid
   3	execute-snapshot	指定参数 指定信号要触发的操作。type
   4	data-collections	信号字段的必需组件，用于指定集合名称或正则表达式的数组，以匹配要包含在快照中的集合名称。 该数组列出了按集合的完全限定名称匹配集合的正则表达式，其格式与在 signal.data.collection 配置属性中指定连接器的信令集合的名称的格式相同。data
   5	incremental	信号字段的可选组件，用于指定要运行的快照操作的类型。 目前，唯一有效的选项是默认值 . 如果未指定值，连接器将运行增量快照。typedataincremental

以下示例显示了连接器捕获的增量快照事件的 JSON。

示例：增量快照事件消息

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 
    },
    "op":"r", 
    "ts_ms":"1620393591654",
    "transaction":null
}

使用卡夫卡信令通道的过程

您可以向已配置的 Kafka 主题发送消息，以请求连接器运行临时增量快照。

Kafka 消息的键必须与连接器配置选项的值匹配。topic.prefix

消息的值是带有 和 字段的 JSON 对象。typedata

信号类型为 ，并且该字段必须具有以下字段：execute-snapshotdata

执行快照 Kafka 消息的示例：

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

具有附加条件的临时增量快照

Debezium 使用该字段来选择集合内容的子集。additional-condition

通常，当 Debezium 运行快照时，它会运行 SQL 查询，例如：

SELECT * FROM <tableName> ….

当快照请求包含 时，会将 追加到 SQL 查询中，例如：additional-conditionadditional-condition

SELECT * FROM <tableName> WHERE <additional-condition> ….

例如，给定一个包含列（主键）和 的集合，如果希望快照仅包含以下内容，则在请求快照时，可以附加一条语句来筛选内容：productsidcolorbrandcolor='blue'additional-condition

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue'"}}`

您可以使用该语句传递基于多个列的条件。 例如，使用与上一示例中相同的集合，如果希望快照仅包含集合中的内容，并且 ，可以发送以下请求：additional-conditionproductsproductscolor='blue'brand='MyBrand'

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue' AND brand='MyBrand'"}}`

停止增量快照

还可以通过向源数据库上的集合发送信号来停止增量快照。 通过将文档插入到信号集合中来提交停止快照信号。 Debezium 检测到信令集合中的更改后，它会读取信号，并在增量快照操作正在进行时停止该操作。

您提交的查询指定 的快照操作，以及要删除的当前正在运行的快照的集合（可选）。incremental

先决条件

• 信令已启用。
  

  • 源数据库上存在信令数据收集。

  • 信令数据收集在 signal.data.collection 属性中指定。

使用源信令通道停止增量快照

1. 将停止快照信号文档插入信令集合：

   <signalDataCollection>.insert({"id" : _<idNumber>,"type" : "stop-snapshot", "data" : {"data-collections" ["<collectionName>", "<collectionName>"],"type": "incremental"}});

   例如

   db.debeziumSignal.insert({ 
   "type" : "stop-snapshot",  
   "data" : {
   "data-collections" ["\"public\".\"Collection1\"", "\"public\".\"Collection2\""], 
   "type": "incremental"} 
   });

   信号命令中的 、 和参数的值对应于信令集合的字段。idtypedata

   示例中的参数说明如下表所示：

   表 4.插入命令中的字段说明，用于将停止增量快照文档发送到信令集合

   项目	价值	描述
   1	db.debeziumSignal	指定源数据库上信令集合的完全限定名称。
   2	零	前面示例中的 insert 方法省略了可选参数的使用。 由于文档没有显式为参数分配值，因此MongoDB自动分配给文档的任意id将成为信号请求的标识符。 使用此字符串标识将消息记录到信令集合中的条目。 Debezium 不使用此标识符字符串。_idid
   3	stop-snapshot	该参数指定信号要触发的操作。type
   4	data-collections	信号字段的可选组件，用于指定集合名称或正则表达式的数组，以匹配要从快照中删除的集合名称。 该数组列出了按集合的完全限定名称匹配集合的正则表达式，其格式与在 signal.data.collection 配置属性中指定连接器的信令集合的名称的格式相同。 如果省略字段的此组件，信号将停止正在进行的整个增量快照。datadata
   5	incremental	指定要停止的快照操作类型的信号字段的必需组件。 目前，唯一有效的选项是 。 如果未指定值，则信号无法停止增量快照。dataincrementaltype

使用 Kafka 信令通道停止增量快照

您可以向配置的 Kafka 信令主题发送信号消息以停止临时增量快照。

Kafka 消息的键必须与连接器配置选项的值匹配。topic.prefix

消息的值是带有 和 字段的 JSON 对象。typedata

信号类型为 ，并且该字段必须具有以下字段：stop-snapshotdata

以下示例显示了一个典型的 Kafka 消息：stop-snapshot

Key = `test_connector`

Value = `{"type":"stop-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

流式处理更改

副本集的连接器任务记录偏移量后，它使用偏移量来确定 oplog 中应开始流式传输更改的位置。 然后，任务（取决于配置）要么连接到副本集的主节点，要么连接到副本集范围的更改流，并从该位置开始流式传输更改。 它处理所有创建、插入和删除操作，并将它们转换为 Debezium 更改事件。 每个更改事件都包括操作日志中发现操作的位置，连接器会定期将其记录为其最近的偏移量。 记录偏移的时间间隔由 offset.flush.interval.ms 控制， 是 Kafka Connect 工作线程配置属性。

当连接器正常停止时，将记录上次处理的偏移量，以便在重新启动时，连接器将准确地从中断的位置继续。 但是，如果连接器的任务意外终止，则任务可能在上次记录偏移量之后但在记录最后一个偏移量之前处理并生成了事件;重新启动后，连接器从上次记录的偏移量开始，可能会生成一些与崩溃之前生成的事件相同的事件。

如前所述，连接器任务始终使用副本集的主节点从 oplog 流式传输更改，确保连接器尽可能看到最新的操作，并且可以以比使用辅助节点更低的延迟捕获更改。 当副本集选择新的主节点时，连接器会立即停止流式传输更改，连接到新的主节点，并从位于相同位置的新主节点开始流式传输更改。 同样，如果连接器在与副本集成员通信时遇到任何问题，它会尝试重新连接，方法是使用指数退避，以免使副本集不堪重负，连接后，它将继续从上次中断的位置流式传输更改。 通过这种方式，连接器能够动态调整副本集成员身份的变化，并自动处理通信故障。

总而言之，MongoDB连接器在大多数情况下继续运行。通信问题可能会导致连接器等待问题解决。

前映像支持

在 MongoDB 6.0 及更高版本中，您可以配置更改流以发出文档的前映像状态，以填充 MongoDB 更改事件的字段。 要在 MongoDB 中启用前映像的使用，必须使用 、 或 来设置集合的 。 要使 Debezium MongoDB 能够在更改事件中包含前映像，请将连接器的 设置为其中一个选项。beforechangeStreamPreAndPostImagesdb.createCollection()createcollModcapture.mode*_with_pre_image

主题名称

MongoDB 连接器将所有插入、更新和删除操作的事件写入每个集合中的文档到单个 Kafka 主题。 Kafka 主题的名称始终采用 logicalName 的形式。数据库名称。collectionName，其中 logicalName 是使用配置属性指定的连接器的逻辑名称，databaseName 是发生操作的数据库的名称，collectionName 是受影响文档所在的 MongoDB 集合的名称。topic.prefix

例如，考虑一个 MongoDB 副本集，其中包含包含四个集合的数据库：、、 和 。 如果为监视此数据库的连接器指定了逻辑名称 ，则该连接器将生成有关以下四个 Kafka 主题的事件：inventoryproductsproducts_on_handcustomersordersfulfillment

• fulfillment.inventory.products

• fulfillment.inventory.products_on_hand

• fulfillment.inventory.customers

• fulfillment.inventory.orders

请注意，主题名称不包含副本集名称或分片名称。 因此，对分片集合（其中每个分片包含集合文档的子集）的所有更改都指向同一 Kafka 主题。

您可以将 Kafka 设置为根据需要自动创建主题。 如果没有，则必须使用 Kafka 管理工具在启动连接器之前创建主题。

分区

MongoDB 连接器不会明确确定如何对事件的主题进行分区。 相反，它允许 Kafka 确定如何根据事件键对主题进行分区。 您可以通过在 Kafka Connect 工作线程配置中定义实现的名称来更改 Kafka 的分区逻辑。Partitioner

Kafka 仅对写入单个主题分区的事件保持总顺序。 按键对事件进行分区确实意味着具有相同键的所有事件始终转到同一分区。 这可确保特定文档的所有事件始终完全有序。

事务元数据

Debezium 可以生成表示事务元数据边界的事件，并丰富变更数据事件消息。

对于每个事务和，Debezium 都会生成一个包含以下字段的事件：BEGINEND

status

BEGIN或END

id

唯一事务标识符的字符串表示形式。

event_count（对于活动）END

事务发出的事件总数。

data_collections（对于活动）END

和 的数组对，提供源自给定数据收集的更改发出的事件数。data_collectionevent_count

以下示例显示了一个典型消息：

{
  "status": "BEGIN",
  "id": "1462833718356672513",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "1462833718356672513",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "rs0.testDB.collectiona",
      "event_count": 1
    },
    {
      "data_collection": "rs0.testDB.collectionb",
      "event_count": 1
    }
  ]
}

除非通过 topic.transaction 选项被覆盖， 事务事件将写入名为 <topic.prefix> 的主题。.transaction

更改数据事件扩充

启用事务元数据后，将使用新字段丰富数据消息。 此字段以字段组合的形式提供有关每个事件的信息：Envelopetransaction

id

唯一事务标识符的字符串表示形式。

total_order

事件在事务生成的所有事件中的绝对位置。

data_collection_order

事件在事务发出的所有事件中的每个数据收集位置。

下面是消息外观的示例：

{
  "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"annek@noanswer.org\"}",
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "1462833718356672513",
    "total_order": "1",
    "data_collection_order": "1"
  }
}

数据更改事件

{
 "schema": { 
   ...
  },
 "payload": { 
   ...
 },
 "schema": { 
   ...
 },
 "payload": { 
   ...
 },
}

{
  "_id": 1004,
  "first_name": "Anne",
  "last_name": "Kretchmar",
  "email": "annek@noanswer.org"
}

{
  "schema": { 
    "type": "struct",
    "name": "fulfillment.inventory.customers.Key", 
    "optional": false, 
    "fields": [ 
      {
        "field": "id",
        "type": "string",
        "optional": false
      }
    ]
  },
  "payload": { 
    "id": "1004"
  }
}

{
  "_id": 1004,
  "first_name": "Anne",
  "last_name": "Kretchmar",
  "email": "annek@noanswer.org"
}

{
    "schema": { 
      "type": "struct",
      "fields": [
        {
          "type": "string",
          "optional": true,
          "name": "io.debezium.data.Json", 
          "version": 1,
          "field": "after"
        },
        {
          "type": "string",
          "optional": true,
          "name": "io.debezium.data.Json",
          "version": 1,
          "field": "patch"
        },
        {
          "type": "struct",
          "fields": [
            {
              "type": "string",
              "optional": false,
              "field": "version"
            },
            {
              "type": "string",
              "optional": false,
              "field": "connector"
            },
            {
              "type": "string",
              "optional": false,
              "field": "name"
            },
            {
              "type": "int64",
              "optional": false,
              "field": "ts_ms"
            },
            {
              "type": "boolean",
              "optional": true,
              "default": false,
              "field": "snapshot"
            },
            {
              "type": "string",
              "optional": false,
              "field": "db"
            },
            {
              "type": "string",
              "optional": false,
              "field": "rs"
            },
            {
              "type": "string",
              "optional": false,
              "field": "collection"
            },
            {
              "type": "int32",
              "optional": false,
              "field": "ord"
            },
            {
              "type": "int64",
              "optional": true,
              "field": "h"
            }
          ],
          "optional": false,
          "name": "io.debezium.connector.mongo.Source", 
          "field": "source"
        },
        {
          "type": "string",
          "optional": true,
          "field": "op"
        },
        {
          "type": "int64",
          "optional": true,
          "field": "ts_ms"
        }
      ],
      "optional": false,
      "name": "dbserver1.inventory.customers.Envelope" 
      },
    "payload": { 
      "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"annek@noanswer.org\"}", 
      "source": { 
        "version": "2.3.3.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": false,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 31,
        "h": 1546547425148721999
      },
      "op": "c", 
      "ts_ms": 1558965515240 
    }
  }

{
    "schema": { ... },
    "payload": {
      "op": "u", 
      "ts_ms": 1465491461815, 
      "before":"{\"_id\": {\"$numberLong\": \"1004\"},\"first_name\": \"unknown\",\"last_name\": \"Kretchmar\",\"email\": \"annek@noanswer.org\"}", 
      "after":"{\"_id\": {\"$numberLong\": \"1004\"},\"first_name\": \"Anne Marie\",\"last_name\": \"Kretchmar\",\"email\": \"annek@noanswer.org\"}", 
      "updateDescription": {
        "removedFields": null,
        "updatedFields": "{\"first_name\": \"Anne Marie\"}", 
        "truncatedArrays": null
      },
      "source": { 
        "version": "2.3.3.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": false,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 1,
        "h": null,
        "tord": null,
        "stxnid": null,
        "lsid":"{\"id\": {\"$binary\": \"FA7YEzXgQXSX9OxmzllH2w==\",\"$type\": \"04\"},\"uid\": {\"$binary\": \"47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=\",\"$type\": \"00\"}}",
        "txnNumber":1
      }
    }
  }

{
    "schema": { ... },
    "payload": {
      "op": "d", 
      "ts_ms": 1465495462115, 
      "before":"{\"_id\": {\"$numberLong\": \"1004\"},\"first_name\": \"Anne Marie\",\"last_name\": \"Kretchmar\",\"email\": \"annek@noanswer.org\"}",
      "source": { 
        "version": "2.3.3.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": true,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 6,
        "h": 1546547425148721999
      }
    }
  }

{
  "name": "inventory-connector", 
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", 
    "mongodb.connection.string": "mongodb://192.168.99.100:27017/?replicaSet=rs0", 
    "topic.prefix": "fullfillment", 
    "collection.include.list": "inventory[.]*" 
  }
}