Debezium 支持
Debezium 引擎、变更数据捕获 （CDC） 入站通道适配器。
这DebeziumMessageProducer允许捕获数据库更改事件，将其转换为消息，并在以后流式传输到出站通道。
你需要在你的项目中包含 Spring 集成的 Debezium 依赖项：
-
Maven
-
Gradle
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-debezium</artifactId>
<version>6.4.2</version>
</dependency>compile "org.springframework.integration:spring-integration-debezium:6.4.2"您还需要为输入 Database 包含一个 debezium 连接器依赖项。 例如，要将 Debezium 与 PostgreSQL 一起使用，您将需要 postgres 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消息内容。 contentTypemust与SerializationFormat为提供的DebeziumEngine.Builder. -
enableBatch- 当设置为false（默认），则 Debezium 适配器会将新的Message对于每个ChangeEvent从源数据库接收到的数据更改事件。 如果设置为true然后，适配器会向下游发送一个Message对于每批ChangeEvent从 Debezium 引擎接收。 这样的有效负载是不可序列化的，需要自定义序列化/反序列化实现。 -
enableEmptyPayload- 启用对逻辑删除（又名删除）消息的支持。 在数据库行删除时，Debezium 可以发送一个逻辑删除更改事件，该事件与已删除的行具有相同的键，值为Optional.empty. 默认为false. -
headerMapper-习惯HeaderMapper实现，允许选择和转换ChangeEvent标头导入Message头。 默认的DefaultDebeziumHeaderMapperimplementation 为setHeaderNamesToMap. 默认情况下，所有标头都已映射。 -
taskExecutor- 设置自定义TaskExecutor用于 Debezium 引擎。
以下代码片段演示了此 channel adapter 的各种配置：
使用 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.prefixconfiguration 选项、数据库名称和表名。例如：my-topic.inventory.orders. |
| 2 | 包含已更改表的 key 的架构和已更改行的实际 key。
键架构及其相应的键有效负载都包含已更改表的PRIMARY KEY（或唯一约束）。 |
| 3 | 与键一样，payload 也有一个 schema 部分和一个 payload value 部分。 的 schema 部分包含描述 payload value 部分的 Envelope 结构的架构，包括其嵌套字段。 创建、更新或删除数据的作的 Change 事件都具有具有 envelope 结构的值 payload。 |
|
这 |
同样，我们可以配置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通过Debeziumfactory 和DebeziumMessageProducerSpec实现。
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())))或者创建一个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();
}
}