Spring Boot集成Debezium快速入门demo

import io.debezium.config.Configuration;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
 
@SpringBootApplication
public class DebeziumSpringBootDemo {
 
    public static void main(String[] args) {
        SpringApplication.run(DebeziumSpringBootDemo.class, args);
    }
 
    @Bean
    public Configuration debeziumConfiguration() {
        // 配置Debezium连接到MySQL数据库的基本参数
        return Configuration.create()
                .with("name", "my-sql-connector")
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("tasks.max", "1")
                .with("database.hostname", "your-db-host")
                .with("database.port", "3306")
                .with("database.user", "your-db-user")
                .with("database.password", "your-db-pass")
                .with("database.server.id", "184054")
                .with("database.server.name", "my-app-connector")
                .with("database.include.list", "my_db")
                .with("database.history.kafka.bootstrap.servers", "kafka:9092")
                .with("database.history.kafka.topic", "my-db-history")
                .with("include.schema.changes", "true")
                .build();
    }
 
    @Bean
    public ChangeEventFormat changeEventFormat() {
        // 使用JSON格式发布变更事件
        return ChangeEventFormat.of(ChangeEventFormat.Target.KAFKA).withKafkaTopicSelector(topicName -> "my_db_topic");
    }
 
    @Bean
    public DebeziumEventHandler debeziumEventHandler() {
        return new DebeziumEventHandler();
    }
 
    @FunctionalInterface
    public interface DebeziumEventHandler {
        void handleEvent(SourceRecord record);
    }
 
    @Bean
    public RecordChangeEvent<SourceRecord> recordChangeEvent(DebeziumEventHandler eventHandler) {
        return new RecordChangeEvent<SourceRecord>() {
            @Override
            public void handleChangeEvent(SourceRecord record) {
                eventHandler.handleEvent(record);
            }
        };
    }
}

在这个示例中，我们创建了一个Spring Boot应用程序，其中包含了Debezium的配置和事件处理逻辑。这个应用程序将连接到MySQL数据库，监控数据变更，并将变更事件发布到Kafka。我们使用\`RecordChange