java监听mysql表变化

import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

import java.util.Properties;

public class MySQLChangeDataCapture {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", "/path/to/storage/offset.dat");
        props.setProperty("offset.storage.topic", "dbhistory.offsets");
        props.setProperty("database.hostname", "localhost");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", "user");
        props.setProperty("database.password", "password");
        props.setProperty("database.server.id", "184054");
        props.setProperty("database.server.name", "dbserver1");
        props.setProperty("database.include.list", "mydb");
        props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
        props.setProperty("database.history.file.filename", "/path/to/dbhistory.dat");
        props.setProperty("include.schema.changes", "false");

        DebeziumEngine<RecordChangeEvent<MyTable>> engine = DebeziumEngine.create(Json.class)
                .using(props)
                .notifying(recordChangeEvent -> {
                    // 处理表变化事件
                    System.out.println(recordChangeEvent);
                })
                .build();

        // 启动监听
        engine.run();
    }
}