Debezium 支持
Debezium引擎，变更数据捕获 (CDC) 入站通道适配器。DebeziumMessageProducer 允许捕获数据库变更事件，将其转换为消息，并稍后流式传输到出站通道。
您需要将 Spring Integration Debezium 依赖项包含到您的项目中
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-debezium</artifactId>
<version>6.3.5</version>
</dependency>compile "org.springframework.integration:spring-integration-debezium:6.3.5"您还需要包含一个针对您的输入数据库的Debezium 连接器依赖项。例如，要将 Debezium 与 PostgreSQL 一起使用，您需要 PostgreSQL Debezium 连接器。
-
Maven
-
Gradle
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-postgres</artifactId>
<version>${debezium-version}</version>
</dependency>compile "io.debezium:debezium-connector-postgres:{debezium-version}"|
将 |
入站 Debezium 通道适配器
Debezium 适配器需要一个预配置的DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>>实例。
|
debezium-supplier 提供了一个开箱即用的 |
|
Debezium Java DSL 可以从提供的 |
此外，可以使用以下配置属性调整DebeziumMessageProducer
-
contentType- 允许处理JSON（默认）、AVRO和PROTOBUF消息内容。contentType必须与为提供的DebeziumEngine.Builder配置的SerializationFormat保持一致。 -
enableBatch- 当设置为false（默认）时，Debezium 适配器将为从源数据库接收到的每个ChangeEvent数据变更事件发送新的Message。如果设置为true，则适配器会为从 Debezium 引擎接收到的每个ChangeEvent批次发送一个Message到下游。此类有效负载不可序列化，需要自定义序列化/反序列化实现。 -
enableEmptyPayload- 启用对墓碑（又名删除）消息的支持。在数据库行删除时，Debezium 可以发送一个墓碑变更事件，该事件具有与已删除行相同的键和Optional.empty的值。默认为false。 -
headerMapper- 自定义HeaderMapper实现，允许将ChangeEvent头选择并转换为Message头。默认的DefaultDebeziumHeaderMapper实现提供了一个setHeaderNamesToMap的设置器。默认情况下，所有头都将被映射。 -
taskExecutor- 为 Debezium 引擎设置自定义TaskExecutor。
以下代码片段演示了此通道适配器的各种配置
使用 Java 配置进行配置
以下 Spring Boot 应用程序显示了如何使用 Java 配置配置入站适配器的示例。
@SpringBootApplication
public class DebeziumJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(DebeziumJavaApplication.class)
.web(WebApplicationType.NONE)
.run(args);
}
@Bean
public MessageChannel debeziumInputChannel() {
return new DirectChannel();
}
@Bean
public MessageProducer debeziumMessageProducer(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder,
MessageChannel debeziumInputChannel) {
DebeziumMessageProducer debeziumMessageProducer =
new DebeziumMessageProducer(debeziumEngineBuilder);
debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
return debeziumMessageProducer;
}
@ServiceActivator(inputChannel = "debeziumInputChannel")
public void handler(Message<?> message) {
Object destination = message.getHeaders().get(DebeziumHeaders.DESTINATION); (1)
String key = new String((byte[]) message.getHeaders().get(DebeziumHeaders.KEY)); (2)
String payload = new String((byte[]) message.getPayload()); (3)
System.out.println("KEY: " + key + ", DESTINATION: " + destination + ", PAYLOAD: " + payload);
}
}| 1 | 事件的目标逻辑目标的名称。通常，目标由topic.prefix配置选项、数据库名称和表名称组成。例如：my-topic.inventory.orders。 |
| 2 | 包含已更改表的键的模式和已更改行的实际键。键模式及其对应的键有效负载都包含已更改表中PRIMARY KEY（或唯一约束）中每一列的字段，时间为连接器创建事件时。 |
| 3 | 与键类似，有效负载具有模式部分和有效负载值部分。模式部分包含描述有效负载值部分的 Envelope 结构的模式，包括其嵌套字段。创建、更新或删除数据的操作的变更事件都具有具有 Envelope 结构的值有效负载。 |
|
|
同样，我们可以将DebeziumMessageProducer配置为批量处理传入的变更事件。
@Bean
public MessageProducer debeziumMessageProducer(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder,
MessageChannel debeziumInputChannel) {
DebeziumMessageProducer debeziumMessageProducer = new DebeziumMessageProducer(debeziumEngineBuilder);
debeziumMessageProducer.setEnableBatch(true);
debeziumMessageProducer.setOutputChannel(debeziumInputChannel);
return debeziumMessageProducer;
}
@ServiceActivator(inputChannel = "debeziumInputChannel")
public void handler(List<ChangeEvent<Object, Object>> payload) {
System.out.println(payload);
}Debezium Java DSL 支持
spring-integration-debezium通过Debezium工厂和DebeziumMessageProducerSpec实现提供了一个方便的 Java DSL 流式 API。
Debezium Java DSL 的入站通道适配器是
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder = ...
IntegrationFlow.from(
Debezium.inboundChannelAdapter(debeziumEngineBuilder)
.headerNames("special*")
.contentType("application/json")
.enableBatch(false))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))或者从本地 Debezium 配置属性创建一个DebeziumMessageProducerSpec实例，并默认为JSON序列化格式。
Properties debeziumConfig = ...
IntegrationFlow
.from(Debezium.inboundChannelAdapter(debeziumConfig))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))以下 Spring Boot 应用程序提供了一个使用 Java DSL 配置入站适配器的示例。
@SpringBootApplication
public class DebeziumJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(DebeziumJavaApplication.class)
.web(false)
.run(args);
}
@Bean
public IntegrationFlow debeziumInbound(
DebeziumEngine.Builder<ChangeEvent<byte[], byte[]>> debeziumEngineBuilder) {
return IntegrationFlow
.from(Debezium
.inboundChannelAdapter(debeziumEngineBuilder)
.headerNames("special*")
.contentType("application/json")
.enableBatch(false))
.handle(m -> System.out.println(new String((byte[]) m.getPayload())))
.get();
}
}
