13.9. 配置 Debezium 连接器以使用 outbox 模式
outbox 模式是在多个(micro)服务之间安全可靠地交换数据的方法。开箱即用模式实施可避免服务内部状态(通常在数据库中保留)和需要相同数据的服务所消耗的事件之间的不一致。
要在 Debezium 应用程序中实施 outbox 模式,请将 Debezium 连接器配置为:
- 捕获 outbox 表中的更改
- 应用 Debezium outbox 事件路由器单一消息转换(SMT)
配置为应用 outbox SMT 的 Debezium 连接器应该只捕获 outbox 表中发生的更改。如需更多信息,请参阅 有选择地应用转换的选项。
只有在每个 outbox 表具有相同的结构时,连接器才可以捕获多个 outbox 表中的更改。
请参阅 带有 Outbox Pattern 的 Reliable microservicess Data Exchange,以了解 outbox 模式很有用的原因及其工作方式。
outbox 事件路由器 SMT 与 MongoDB 连接器不兼容。
MongoDB 用户可以运行 MongoDB outbox 事件路由器 SMT。
以下主题提供详情:
- 第 13.9.1 节 “Debezium outbox 消息示例”
- 第 13.9.2 节 “Debezium outbox 事件路由器 SMT 期望的 outbox 表结构”
- 第 13.9.3 节 “基本 Debezium outbox 事件路由器 SMT 配置”
- 第 13.9.4 节 “用于有选择地应用 Outbox 事件路由器转换的选项”
- 第 13.9.5 节 “在 Debezium outbox 消息中使用 Avro 作为有效负载格式”
- 第 13.9.6 节 “在 Debezium outbox 消息中发出其他字段”
- 第 13.9.7 节 “扩展转义的 JSON 字符串作为 JSON”
- 第 13.9.8 节 “用于配置 outbox 事件路由器转换的选项”
13.9.1. Debezium outbox 消息示例
要了解 Debezium outbox 事件路由器 SMT 是如何配置的,请查看以下 Debezium outbox 消息示例:
# Kafka Topic: outbox.event.order # Kafka Message key: "1" # Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc" # Kafka Message Timestamp: 1556890294484 { "{\"id\": 1, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}" }
配置为应用 outbox 事件路由器 SMT 的 Debezium 连接器通过转换 Debezium 原始消息来生成上述消息,如下所示:
# Kafka Message key: "406c07f3-26f0-4eea-a50c-109940064b8f" # Kafka Message Headers: "" # Kafka Message Timestamp: 1556890294484 { "before": null, "after": { "id": "406c07f3-26f0-4eea-a50c-109940064b8f", "aggregateid": "1", "aggregatetype": "Order", "payload": "{\"id\": 1, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}", "timestamp": 1556890294344, "type": "OrderCreated" }, "source": { "version": "2.3.7.Final", "connector": "postgresql", "name": "dbserver1-bare", "db": "orderdb", "ts_usec": 1556890294448870, "txId": 584, "lsn": 24064704, "schema": "inventory", "table": "outboxevent", "snapshot": false, "last_snapshot_record": null, "xmin": null }, "op": "c", "ts_ms": 1556890294484 }
此 Debezium outbox 消息示例基于 默认的 outbox 事件路由器配置,它假定基于聚合的 outbox 表结构和事件路由。要自定义行为,outbox 事件路由器 SMT 提供了大量 配置选项。
13.9.2. Debezium outbox 事件路由器 SMT 期望的 outbox 表结构
要应用 default outbox 事件路由器 SMT 配置,您的 outbox 表被认为具有以下列:
Column | Type | Modifiers --------------+------------------------+----------- id | uuid | not null aggregatetype | character varying(255) | not null aggregateid | character varying(255) | not null type | character varying(255) | not null payload | jsonb |
column | 效果 |
---|---|
|
包含事件的唯一 ID。在 outbox 消息中,这个值是一个标头。例如,您可以使用此 ID 删除重复的消息。 |
|
包含 SMT 附加到连接器发出 outbox 消息的主题名称中的值。默认行为是,这个值替换了 |
|
包含事件密钥,它为有效负载提供 ID。SMT 使用这个值作为 emitted outbox 消息中的键。这对于在 Kafka 分区中维护正确顺序非常重要。 |
|
outbox 更改事件的表示。默认结构是 JSON。默认情况下,Kafka message 值只由 |
其他自定义列 |
outbox 表中的任何其他列 可以添加到 outbox 事件, 可以在 payload 部分或作为消息标头中添加。 |
13.9.3. 基本 Debezium outbox 事件路由器 SMT 配置
要将 Debezium 连接器配置为支持 outbox 模式,请配置 outbox.EventRouter
SMT。要获取 SMT 的默认行为,请在没有指定任何选项的情况下将其添加到连接器配置中,如下例所示:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
自定义配置
连接器可能会发出许多类型的事件信息(例如: heartbeat 消息、tombstone 消息或有关事务或模式更改的元数据信息)。要将转换应用到源自 outbox 表中的事件,请定义一个 有选择地将转换应用到这些事件的 SMT predicate 语句。
13.9.4. 用于有选择地应用 Outbox 事件路由器转换的选项
除了 Debezium 连接器在数据库更改时发出的更改事件消息外,连接器还会发出其他类型的信息,包括心跳消息,以及有关 schema 更改和事务的元数据消息。由于这些其他消息的结构与 SMT 设计的更改事件消息的结构不同,因此最好将连接器配置为有选择地应用 SMT,以便它只处理预期的数据更改消息。您可以使用以下任一方法配置连接器来有选择地应用 SMT:
- 为转换配置 SMT predicate。
-
对 SMT 使用
route.topic.regex
配置选项。
13.9.5. 在 Debezium outbox 消息中使用 Avro 作为有效负载格式
outbox 事件路由器 SMT 支持任意有效负载格式。outbox 表中的 payload
列值以透明的方式传递。使用 JSON 的替代方法是使用 Avro。这对消息格式监管很有用,并确保 outbox 事件模式以向后兼容的方式发展。
源应用程序如何为 outbox 消息有效负载生成 Avro 格式内容超出了本文档的范围。可以利用 KafkaAvroSerializer
类序列化 GenericRecord
实例。要确保 Kafka message 值是准确的 Avro 二进制数据,请将以下配置应用到连接器:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter value.converter=io.debezium.converters.BinaryDataConverter
默认情况下,有效负载
列值( Avro 数据)是唯一的消息值。配置 BinaryDataConverter
,因为值转换器将 payload
列值按原样传播到 Kafka 消息值中。
Debezium 连接器可以配置为发送心跳、事务元数据或模式更改事件(支持因连接器而异)。这些事件无法通过 BinaryDataConverter
序列化,因此必须提供额外的配置,因此转换器知道如何序列化这些事件。例如,以下配置演示了使用没有模式的 Apache Kafka JsonConverter
:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter value.converter=io.debezium.converters.BinaryDataConverter value.converter.delegate.converter.type=org.apache.kafka.connect.json.JsonConverter value.converter.delegate.converter.type.schemas.enable=false
delegate Converter
实现通过 delegate.converter.type
选项指定。如果转换器需要任何其他配置选项,也可以指定它们,如使用 schemas.enable=false
显示的模式禁用。
从 Debezium 版本 1.9 开始,converter io.debezium.converters.ByteBufferConverter
已被弃用,并在 2.0 中删除。另外,在使用 Kafka Connect 时,必须在升级到 Debezium 2.x 前更新连接器的配置。
13.9.6. 在 Debezium outbox 消息中发出其他字段
您的 outbox 表可能包含您要添加到发出出出出出消息的值的列。例如,一个 outbox 表,它在 aggregatetype
列中具有 购买顺序
的值,另一个列, eventType
,其可能的值是 order-created
和 order-shipped
。可以使用语法 column:placement:alias
来添加其他字段。
放置
允许的值有: - header
- envelope
- partition
要在 outbox 消息标头中发出 eventType
列值,请配置 SMT,如下所示:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter transforms.outbox.table.fields.additional.placement=eventType:header:type
结果将是 Kafka 消息的标头,类型
作为其键,eventType
列的值作为其值。
要在 outbox 消息 envelope 中发出 eventType
列值,请配置 SMT,如下所示:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter transforms.outbox.table.fields.additional.placement=eventType:envelope:type
要控制在其上生成 outbox 消息的分区,请配置 SMT,如下所示:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter transforms.outbox.table.fields.additional.placement=partitionColumn:partition
请注意,对于 分区
放置,添加别名将无效。
13.9.7. 扩展转义的 JSON 字符串作为 JSON
您可能注意到,Debezium outbox 消息包含以 String 表示 的有效负载
。因此,当这个字符串实际上是 JSON 时,它会出现在结果 Kafka 信息中,如下所示:
# Kafka Topic: outbox.event.order # Kafka Message key: "1" # Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc" # Kafka Message Timestamp: 1556890294484 { "{\"id\": 1, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}" }
outbox 事件路由器允许您将此消息内容扩展为 "real" JSON,使用从 JSON 文档本身分离的编译模式。这样,Kafka 信息的结果类似如下:
# Kafka Topic: outbox.event.order # Kafka Message key: "1" # Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc" # Kafka Message Timestamp: 1556890294484 { "id": 1, "lineItems": [{"id": 1, "item": "Debezium in Action", "status": "ENTERED", "quantity": 2, "totalPrice": 39.98}, {"id": 2, "item": "Debezium for Dummies", "status": "ENTERED", "quantity": 1, "totalPrice": 29.99}], "orderDate": "2019-01-31T12:13:01", "customerId": 123 }
要启用此转换,您必须将 table.expand.json.payload
设置为 true,并使用如下所示的 JsonConverter
:
transforms=outbox,... transforms.outbox.type=io.debezium.transforms.outbox.EventRouter transforms.outbox.table.expand.json.payload=true value.converter=org.apache.kafka.connect.json.JsonConverter
13.9.8. 用于配置 outbox 事件路由器转换的选项
下表描述了您可以为 outbox 事件路由器 SMT 指定的选项。在表格中,Group 列指示 Kafka 的配置选项分类。
选项 | 默认 | 组 | 描述 |
---|---|---|---|
| 表 |
决定在 outbox 表中有
outbox 表中的所有更改都应该是 | |
| 表 |
指定包含唯一事件 ID 的 outbox 表列。此 ID 将存储在 | |
| 表 | 指定包含 event 键的 outbox 表列。当此列包含值时,SMT 会使用该值作为 emitted outbox 信息中的键。这对于在 Kafka 分区中维护正确的顺序非常重要。 | |
表 | 默认情况下,emitted outbox 消息中的时间戳是 Debezium 事件时间戳。要在 outbox 消息中使用不同的时间戳,请将此选项设置为包含您要发出出出消息的时间戳的 outbox 表列。 | ||
| 表 | 指定包含事件有效负载的 outbox 表列。 | |
| 表 |
指定是否应进行 String 有效负载的 JSON 扩展。如果没有找到内容,或者在解析错误时保留内容,则会保留"as is"。 | |
| 表 |
当启用 JSON 扩展属性
| |
table, Envelope | 指定您要添加到 outbox 消息标头或信封的一个或多个 outbox 表列。指定以逗号分隔的对列表。在每个对中,指定列的名称,以及是否希望该值在标头中还是信封。使用冒号分隔对中的值,例如:
要为列指定别名,请将别名指定为 trio,例如:
第二个值是放置,它必须始终是 配置示例包括在 emitting additional fields in Debezium outbox messages 中。 | ||
table, Schema | 设置后,这个值将用作 schema 版本,如 Kafka Connect Schema Javadoc 所述。 | ||
| 路由器 | 指定 outbox 表中的列名称。默认行为是此列中的值成为连接器发出 outbox 消息的主题名称的一部分。示例位于 预期 outbox 表的描述 中。 | |
| 路由器 |
指定 outbox SMT 在 RegexRouter 中应用到 outbox 表记录的正则表达式。这个正则表达式是设置 | |
| 路由器 |
指定连接器发出 outbox 消息的主题名称。默认主题名称为
| |
| 路由器 |
指明一个空或 |