前期准备

需要提前安装mysql并且开启binlog,需要准备kafka和zookeeper环境

示例采用debezium1.9.0版本

Maven配置

<version.debezium>1.9.0.Final</version.debezium>

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-api</artifactId>

    <version>${version.debezium}</version>

</dependency>

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-embedded</artifactId>

    <version>${version.debezium}</version>

</dependency>

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-connector-mysql</artifactId>

    <version>${version.debezium}</version>

</dependency>

Java代码

import io.debezium.engine.ChangeEvent;

import io.debezium.engine.DebeziumEngine;

import io.debezium.engine.format.Json;

import java.io.IOException;

import java.util.Properties;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.TimeUnit;

public class DebeziumTest {

    private static DebeziumEngine<ChangeEvent<String, String>> engine;

    public static void main(String[] args) throws Exception {

        final Properties props = new Properties();

        props.setProperty("name", "dbz-engine");

        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");

        //offset config begin - 使用文件来存储已处理的binlog偏移量

        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");

//        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.KafkaOffsetBackingStore");

        props.setProperty("offset.storage.file.filename", "D:/tmp/dbz/storage/mysql_offsets.dat");

        props.setProperty("offset.flush.interval.ms", "0");

        //offset config end

        props.setProperty("database.server.name", "mysql-connector");

        props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");

        props.setProperty("database.history.file.filename", "D:/tmp/dbz/storage/mysql_dbhistory.txt");

        props.setProperty("database.server.id", "122110"); //随机设置

        props.setProperty("database.hostname", "localhost");

        props.setProperty("database.port", "3306");

        props.setProperty("database.user", "root");

        props.setProperty("database.password", "123456");

        props.setProperty("database.include.list", "test");//要捕获的数据库名

        props.setProperty("topic.prefix", "mysql-");

//        props.setProperty("table.include.list", "inventory.a");//要捕获的数据表

        props.setProperty("snapshot.mode", "initial");//全量+增量

        props.setProperty("bootstrap.servers", "localhost:9092");

        props.setProperty("topic", "test");

//        props.setProperty("offset.storage.topic", "log-t");

//        props.setProperty("offset.storage.partitions", "1");

//        props.setProperty("offset.storage.replication.factor", "1");

//        KafkaProducerTest test = new KafkaProducerTest("test1841");

        // 使用上述配置创建Debezium引擎，输出样式为Json字符串格式

        engine = DebeziumEngine.create(Json.class)

                .using(props)

                .notifying(record -> {

                    //test.sendMsg(record.value());

                    System.out.println(record.value());//输出到控制台

                })

                .using((success, message, error) -> {

                    if (error != null) {

                        error.printStackTrace();

                        // 报错回调

                        System.out.println("------------error, message:" + message);

                        System.out.println( "exception:" + error);

                    }

                    closeEngine(engine);

                })

                .build();

        ExecutorService executor = Executors.newSingleThreadExecutor();

        executor.execute(engine);

        addShutdownHook(engine);

        awaitTermination(executor);

        System.out.println("------------main finished.");

    }

    private static void closeEngine(DebeziumEngine<ChangeEvent<String, String>> engine) {

        try {

            engine.close();

        } catch (IOException ignored) {

        }

    }

    private static void addShutdownHook(DebeziumEngine<ChangeEvent<String, String>> engine) {

        Runtime.getRuntime().addShutdownHook(new Thread(() -> closeEngine(engine)));

    }

    private static void awaitTermination(ExecutorService executor) {

        if (executor != null) {

            try {

                executor.shutdown();

                while (!executor.awaitTermination(5, TimeUnit.SECONDS)) {

                }

            } catch (InterruptedException e) {

                Thread.currentThread().interrupt();

            }

        }

    }

}

运行效果

随意修改一条mysql表中的数据

修改后

代码日志

{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"int32","optional":true,"field":"age"},{"type":"string","optional":true,"field":"phone"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"name":"io.debezium.time.ZonedTimestamp","version":1,"field":"create_time"}],"optional":true,"name":"mysql_connector.di_test.t_user.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"int32","optional":true,"field":"age"},{"type":"string","optional":true,"field":"phone"},{"type":"string","optional":true,"field":"address"},{"type":"string","optional":true,"name":"io.debezium.time.ZonedTimestamp","version":1,"field":"create_time"}],"optional":true,"name":"mysql_connector.di_test.t_user.Value","field":"after"},{"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":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"mysql_connector.di_test.t_user.Envelope"},"payload":{"before":{"id":2,"name":"lisi2","age":19,"phone":"18444444","address":"北京海淀1","create_time":"2023-12-28T06:47:07Z"},"after":{"id":2,"name":"lisi2","age":19,"phone":"18444444","address":"北京海淀","create_time":"2023-12-28T06:47:07Z"},"source":{"version":"1.9.0.Final","connector":"mysql","name":"mysql-connector","ts_ms":1722423711000,"snapshot":"false","db":"di_test","sequence":null,"table":"t_user","server_id":1,"gtid":null,"file":"binlog.000008","pos":3134,"row":0,"thread":17,"query":null},"op":"u","ts_ms":1722423711663,"transaction":null}}