当前位置：首页 > news >正文

Spring Boot+Debezium：解决 MySQL Binlog监听

news 来源：原创 2024/9/29 7:32:40

在当今的企业应用中，对数据变更进行实时捕获和处理是至关重要的。而Debezium作为一个强大的开源平台，能够实时捕获数据库的变化事件，为实时数据流处理提供了可靠的解决方案。在本文中，我们将探讨如何将Debezium与SpringBoot项目无缝集成，以便对MySQL Binlog进行监听和处理。

为什么选择Debezium？

实时性：Debezium能够实时监控数据库的变更，捕获到变更事件并立即进行处理，使得应用能够及时响应数据的变化。
可靠性：Debezium基于可靠的CDC（Change Data
Capture）技术，能够保证数据的准确性和一致性，确保捕获到的变更事件能够被正确地处理。
易用性：Debezium提供了丰富的API和文档，使得集成到SpringBoot项目中变得简单和容易。

如何整合Debezium到Spring Boot项目中？

1. 添加依赖

首先，我们需要在pom.xml中添加必要的依赖。包括SpringBoot的基础依赖和Debezium的MySQL连接器依赖。

<dependencies><!-- Spring Boot Starter Dependencies --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Debezium Dependencies --><dependency><groupId>io.debezium</groupId><artifactId>debezium-embedded</artifactId><version>1.8.0.Final</version></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-connector-mysql</artifactId><version>1.8.0.Final</version></dependency>
</dependencies>

2. 配置Debezium

在SpringBoot项目的application.properties文件中，配置Debezium连接MySQL所需的参数。

# MySQL connection properties
debezium.mysql.hostname=localhost
debezium.mysql.port=3306
debezium.mysql.user=debezium
debezium.mysql.password=debezium
debezium.mysql.database.history=io.debezium.relational.history.FileDatabaseHistory
debezium.mysql.database.history.file.filename=/tmp/dbhistory.dat# Debezium connector properties
debezium.name=engine
debezium.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.database.hostname=${debezium.mysql.hostname}
debezium.database.port=${debezium.mysql.port}
debezium.database.user=${debezium.mysql.user}
debezium.database.password=${debezium.mysql.password}
debezium.database.server.id=85744
debezium.database.server.name=my-app-connector
debezium.database.include.list=mydatabase
debezium.table.include.list=mydatabase.mytable
debezium.database.history=${debezium.mysql.database.history}
debezium.database.history.file.filename=${debezium.mysql.database.history.file.filename}

3. 创建Debezium引擎配置类

创建一个Spring配置类，用于初始化和配置Debezium引擎。

package com.example.debezium.config;import io.debezium.config.Configuration;
import io.debezium.embedded.Connect;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.format.Json;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.Properties;@Configuration
public class DebeziumConfig {@Value("${debezium.name}")private String name;@Value("${debezium.connector.class}")private String connectorClass;@Value("${debezium.database.hostname}")private String hostname;@Value("${debezium.database.port}")private String port;@Value("${debezium.database.user}")private String user;@Value("${debezium.database.password}")private String password;@Value("${debezium.database.server.id}")private String serverId;@Value("${debezium.database.server.name}")private String serverName;@Value("${debezium.database.include.list}")private String includeList;@Value("${debezium.table.include.list}")private String tableIncludeList;@Value("${debezium.database.history}")private String databaseHistory;@Value("${debezium.database.history.file.filename}")private String databaseHistoryFile;@Beanpublic Configuration debeziumConfiguration() {Properties props = new Properties();props.setProperty("name", name);props.setProperty("connector.class", connectorClass);props.setProperty("database.hostname", hostname);props.setProperty("database.port", port);props.setProperty("database.user", user);props.setProperty("database.password", password);props.setProperty("database.server.id", serverId);props.setProperty("database.server.name", serverName);props.setProperty("database.include.list", includeList);props.setProperty("table.include.list", tableIncludeList);props.setProperty("database.history", databaseHistory);props.setProperty("database.history.file.filename", databaseHistoryFile);return Configuration.from(props);}@Beanpublic EmbeddedEngine debeziumEngine(Configuration config) {return EmbeddedEngine.create().using(config).notifying(record -> {// Handle the change event hereSystem.out.println(record);}).build();}
}

4. 创建监听器

编写一个监听器类，用于处理捕获到的数据库变化事件。

package com.example.debezium.listener;import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.Json;
import org.springframework.stereotype.Component;@Component
public class DebeziumListener implements EmbeddedEngine.CompletionCallback {@Overridepublic void handle(RecordChangeEvent<SourceRecord> event, EmbeddedEngine.Context context) {// 解析并处理事件String value = event.record().value().toString();System.out.println("Received event: " + value);}@Overridepublic void handle(Throwable throwable, EmbeddedEngine.Context context) {// 错误处理throwable.printStackTrace();}
}

5. 启动Debezium引擎

在SpringBoot的主类中启动Debezium引擎。

package com.example.debezium;import io.debezium.embedded.EmbeddedEngine;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class DebeziumApplication implements CommandLineRunner {@Autowiredprivate EmbeddedEngine debeziumEngine;public static void main(String[] args) {SpringApplication.run(DebeziumApplication.class, args);}@Overridepublic void run(String... args) throws Exception {new Thread(debeziumEngine).start();}
}