?看這篇文章之前，建議先看這篇文章，主要是采用數據庫的binlog進行監聽

mysql實時監聽canal+kafka_怪只怪滿眼盡是人間煙火-CSDN博客1.首先安裝mysql2.然后安裝kafka然后安裝kafkahttps://qushen.blog.csdn.net/article/details/1226820003.安裝并配置canalcanal.deployer-1.1.5.tar.gz??????https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz...https://qushen.blog.csdn.net/article/details/122759354

pom依賴

<properties><java.version>1.8</java.version><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><debezium.version>1.8.0.Final</debezium.version></properties><!-- https://mvnrepository.com/artifact/io.debezium/debezium-api --><dependency><groupId>io.debezium</groupId><artifactId>debezium-api</artifactId><version>${debezium.version}</version></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-embedded</artifactId><version>${debezium.version}</version><exclusions><exclusion><artifactId>slf4j-log4j12</artifactId><groupId>org.slf4j</groupId></exclusion></exclusions></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-connector-mysql</artifactId><version>${debezium.version}</version></dependency>

java代碼

import io.debezium.engine.ChangeEvent; import io.debezium.engine.DebeziumEngine; import io.debezium.engine.format.Json;import java.util.Properties; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;/*** @author qushen* @create 2022/2/24 11:30*/ public class debeziumCDC {public static void main(String[] args) {final Properties props = new Properties();/* begin connector properties */props.setProperty("database.hostname", "192.168.23.144"); // ipprops.setProperty("database.port", "3306"); // 端口props.setProperty("database.user", "root"); // 用戶名props.setProperty("database.password", "123456"); // 密碼props.setProperty("database.server.id", "123456789"); // value可以任意修改,別重復props.setProperty("database.server.name", "qushen"); // 可以任意修改,別重復props.setProperty("database.serverTimezone", "UTC"); // 時區// 1. 使用database.whitelist，只設置數據庫（會通知全庫的CDC信息）// 2. 使用table.whitelist，設置庫名和表名（會通知單個庫的單個表的CDC信息）props.setProperty("database.whitelist", "qushen"); // 庫名 // props.setProperty("table.whitelist", "db_inventory_cdc.tb_products_cdc"); // 庫.表名props.setProperty("name", "instala-core");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", "/tmp/offsets.dat");props.setProperty("offset.flush.interval.ms", "60000");props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");props.setProperty("database.history.file.filename", "/tmp/dbhistory.dat");// props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName()); // props.setProperty("offset.flush.interval.ms", String.valueOf(10000L)); // props.setProperty("key.converter.schemas.enable", "false"); // props.setProperty("value.converter.schemas.enable", "true"); // props.setProperty("include.schema.changes", "true"); // props.setProperty("tombstones.on.delete", "false"); // props.setProperty("database.history", FileDatabaseHistory.class.getCanonicalName()); // props.setProperty("database.history.store.only.monitored.tables.ddl", "true"); // props.setProperty("database.history.instance.name", UUID.randomUUID().toString()); // props.setProperty("database.history.skip.unparseable.ddl", "true");DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class).using(props).notifying(record -> {// record中會有操作的類型（增、刪、改）和具體的數據// key是主鍵System.out.println("record.key() = " + record.key());System.out.println("record.value() = " + record.value());}).build();Executors.newSingleThreadExecutor().execute(engine);// Run the engine asynchronously ...ExecutorService executor = Executors.newSingleThreadExecutor();executor.execute(engine);} }

總結

以上是生活随笔為你收集整理的Spring Boot集成Debezium监控数据库变化的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。