Kafka与关系型数据库的集成方案与数据同步实践

He ZhenBin 2026-09-18 04:15

一、为什么要把Kafka和关系型数据库搭伙用

咱们先从最接地气的场景说，很多做开发的都碰过这么个事儿：比如你做个电商系统，用户下单后，得干好几件事——给订单表插数据、给库存表减库存、给支付系统发通知、给物流系统同步订单、还要给用户发短信。要是把这些事儿都堆在下单这一个接口里，那接口响应速度肯定慢，而且要是中间某一步失败了，比如减库存失败，订单数据还得回滚，麻烦得很。

这时候Kafka就派上用场了，它就像个快递中转站，把下单这个核心操作（只往关系型数据库插订单）产生的消息，先存到中转站里，后面需要处理的系统（库存、支付、物流）自己去中转站拿消息处理就行，核心接口就快了。但这里有个问题：Kafka里的消息和关系型数据库里的数据，得保持一致啊，比如订单表有了新数据，Kafka里就得有对应的消息，不然库存系统拿不到消息，就不知道要减库存；反过来，要是Kafka里的消息处理完了，关系型数据库里的状态也得更新，比如订单支付成功了，数据库里的订单状态得改成已支付。

所以这俩的集成和数据同步，本质上就是解决“两边数据不能乱”的问题，下面就一步步说具体怎么做。

二、常见的集成方案及适用场景

2.1 方案一：双写（业务代码自己写两边）

这个方案是最容易想到的，就是业务代码在往关系型数据库写数据的时候，同时往Kafka发一条消息。比如用户下单，代码先把订单数据插到MySQL的订单表，然后立刻组装一条Kafka消息，发送到指定的主题。

这个方案的优点是实现简单，不用额外加中间件，业务逻辑自己可控。但缺点也很明显：首先，双写的原子性没法保证，比如插数据库成功了，发Kafka消息失败了，那两边数据就不一致了；其次，要是业务逻辑很复杂，每个需要同步的操作都要写双写代码，很容易漏写，维护成本高；最后，要是数据库里的数据是批量更新的，比如一次改了100条订单的状态，那往Kafka发消息的时候，得发100条，性能会受影响。

这个方案适合什么场景呢？就是业务量不大，数据一致性要求不高的小项目，比如个人博客、小型的内部管理系统，先凑合用，出问题了手动改改就行。

2.2 方案二：数据库的CDC（变更数据捕获）

CDC是啥？说白了就是监控数据库的变化，只要数据库里的数据改了（增、删、改），就自动把这个变化的信息抓出来，然后发送到Kafka里。这个方案就解决了双写的问题，因为它不用业务代码写Kafka消息，是数据库自己的变化被监控到后自动触发的，原子性有保证。

现在主流的关系型数据库，比如MySQL、PostgreSQL，都支持CDC，而且有现成的工具可以直接用，比如针对MySQL的Debezium、Canal，针对PostgreSQL的pg_chameleon。这个方案的优点是数据一致性高，业务代码不用改，维护成本低；缺点是需要额外部署CDC工具，对数据库的配置有要求，比如MySQL要开启binlog（二进制日志），这个会稍微增加一点数据库的性能开销，但一般可以忽略。

这个方案适合大部分中大型项目，尤其是数据一致性要求高的场景，比如电商、金融、物流这些系统，现在很多公司的核心业务都用这个方案。

2.3 方案三：定时同步（全量+增量）

这个方案就是用定时任务，定期把关系型数据库里的数据同步到Kafka里。一般是先做一次全量同步，把数据库里所有符合条件的数据都读出来，发送到Kafka，然后再做增量同步，只同步上次同步之后新增或修改的数据。

这个方案的优点是实现简单，不用改数据库配置，不用额外部署复杂的中间件，只要写个定时任务就行；缺点是实时性差，比如数据库里的数据改了，可能要等几分钟甚至几小时才能同步到Kafka，数据一致性也没法保证，比如同步的时候数据库里的数据又改了，就会出现重复或错误的数据。

这个方案适合什么场景呢？就是对实时性要求不高的场景，比如报表统计、数据分析，只要每天同步一次就行，不用实时更新。

三、具体实践：用CDC方案实现MySQL和Kafka的数据同步

下面咱们用具体的例子来演示，用的技术栈是：MySQL 8.0（关系型数据库）、Kafka 3.5（消息队列）、Debezium 2.4（CDC工具）、Docker（用来快速部署这些中间件）。

首先得把这些中间件都部署好，咱们用Docker Compose来部署，这样不用一个个手动装，很方便。

version: '3.8'
services:
  # MySQL服务配置
  mysql:
    image: mysql:8.0
    container_name: mysql-for-kafka
    environment:
      MYSQL_ROOT_PASSWORD: root123 # MySQL root用户密码
      MYSQL_DATABASE: test_db # 初始化创建的数据库
    ports:
      - "3306:3306"
    volumes:
      # 挂载配置文件，开启binlog
      - ./mysql-config:/etc/mysql/conf.d
    command: --default-authentication-plugin=mysql_native_password

  # Kafka服务配置
  kafka:
    image: wurstmeister/kafka:2.13-3.5.1
    container_name: kafka-for-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    depends_on:
      - zookeeper

  # Zookeeper（Kafka依赖的协调服务）
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    container_name: zookeeper-for-kafka
    ports:
      - "2181:2181"

  # Debezium服务配置
  debezium:
    image: debezium/connect:2.4
    container_name: debezium-for-kafka
    ports:
      - "8083:8083"
    environment:
      GROUP_ID: debezium-group
      CONFIG_STORAGE_TOPIC: debezium-configs
      OFFSET_STORAGE_TOPIC: debezium-offsets
      STATUS_STORAGE_TOPIC: debezium-status
      BOOTSTRAP_SERVERS: kafka:9092
    depends_on:
      - kafka
      - mysql

上面这个Docker Compose的配置文件，把四个服务都定义好了，接下来要给MySQL加个配置文件，开启binlog，创建一个mysql-config的文件夹，里面放my.cnf文件：

[mysqld]
# 开启二进制日志，指定日志文件名前缀
log-bin=mysql-bin
# 配置唯一的服务器ID，必须和其他MySQL实例不同
server-id=1
# 配置二进制日志的格式为ROW模式，Debezium要求必须是这个模式
binlog-format=ROW
# 配置binlog的过期时间，比如保留7天
expire_logs_days=7
# 配置binlog的大小，比如单个日志文件最大100M
max_binlog_size=100M

配置好之后，就可以启动这些服务了，在配置文件所在的目录执行下面的命令：

docker-compose up -d

启动成功后，先登录MySQL，创建一个测试表，用来做数据同步测试：

# 进入MySQL容器
docker exec -it mysql-for-kafka mysql -uroot -proot123

然后在MySQL的命令行里执行：

-- 切换到test_db数据库
USE test_db;
-- 创建订单表
CREATE TABLE orders (
    id INT PRIMARY KEY AUTO_INCREMENT COMMENT '订单ID',
    order_no VARCHAR(32) NOT NULL COMMENT '订单编号',
    user_id INT NOT NULL COMMENT '用户ID',
    amount DECIMAL(10,2) NOT NULL COMMENT '订单金额',
    status TINYINT NOT NULL DEFAULT 1 COMMENT '订单状态：1=待支付，2=已支付，3=已取消',
    create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='订单表';

接下来要配置Debezium，让它监控MySQL的orders表，把变化同步到Kafka。Debezium是通过HTTP接口来配置的，执行下面的命令：

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors -d '{
  "name": "mysql-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "root123",
    "database.server.id": "1",
    "database.server.name": "test-mysql",
    "database.whitelist": "test_db",
    "table.whitelist": "test_db.orders",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.test_db",
    "snapshot.mode": "initial"
  }
}'

这个配置的意思是：创建一个叫mysql-orders-connector的连接器，连接到MySQL的test_db数据库，监控orders表，把变化同步到Kafka的test-mysql.test_db.orders主题（Debezium会自动创建这个主题）。

现在来测试一下数据同步，先往orders表插一条数据：

-- 在MySQL命令行执行
INSERT INTO orders (order_no, user_id, amount) VALUES ('ORD20240520001', 1001, 99.99);

然后去Kafka里看看有没有这条消息，执行下面的命令：

# 进入Kafka容器
docker exec -it kafka-for-kafka /bin/bash
# 消费test-mysql.test_db.orders主题的消息
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-mysql.test_db.orders --from-beginning

你会看到输出一条JSON格式的消息，里面包含了刚才插入的订单数据，还有操作类型（c表示create，也就是新增）。

再测试一下更新操作，在MySQL里执行：

UPDATE orders SET status = 2 WHERE order_no = 'ORD20240520001';

再去Kafka消费消息，会看到一条新的消息，操作类型是u（update），里面包含了更新后的订单状态。

再测试删除操作：

DELETE FROM orders WHERE order_no = 'ORD20240520001';

Kafka里会收到操作类型为d（delete）的消息。

四、反向同步：Kafka数据同步到关系型数据库

刚才说的是数据库到Kafka的同步，那反过来，Kafka里的数据要同步到关系型数据库怎么办？比如我们有个用户行为日志的Kafka主题，要把这些日志存到MySQL的日志表，方便后续查询。

这个时候可以用Kafka Connect的JDBC连接器，或者自己写一个消费者程序，从Kafka消费消息，然后批量插入到数据库里。这里咱们用自己写消费者的方式，技术栈还是Java 17、Kafka 3.5、MySQL 8.0，代码如下：

首先，先引入依赖，用Maven的话，pom.xml里加：

<dependencies>
    <!-- Kafka客户端依赖 -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.5.1</version>
    </dependency>
    <!-- MySQL驱动依赖 -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
    <!-- 连接池依赖，用来管理数据库连接 -->
    <dependency>
        <groupId>com.zaxxer</groupId>
        <artifactId>HikariCP</artifactId>
        <version>5.0.1</version>
    </dependency>
</dependencies>

然后写消费者程序：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaToMysqlConsumer {
    // Kafka主题名，这里假设用户行为日志的主题是user-behavior-log
    private static final String TOPIC = "user-behavior-log";
    // Kafka服务器地址
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";
    // 消费者组ID
    private static final String GROUP_ID = "kafka-to-mysql-group";

    public static void main(String[] args) {
        // 1. 配置Kafka消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        // 键和值的反序列化器，因为消息是JSON字符串，所以用String反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 自动提交偏移量，每次消费完自动记录消费位置
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅主题
        consumer.subscribe(Collections.singletonList(TOPIC));

        // 2. 配置数据库连接池
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/test_db?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8");
        config.setUsername("root");
        config.setPassword("root123");
        config.setMaximumPoolSize(10); // 最大连接数
        HikariDataSource dataSource = new HikariDataSource(config);

        // 3. 消费消息并插入数据库
        try {
            while (true) {
                // 拉取消息，最多等待100ms
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    continue;
                }
                // 批量插入，减少数据库连接开销
                String sql = "INSERT INTO user_behavior_log (user_id, behavior_type, content, create_time) VALUES (?, ?, ?, ?)";
                try (Connection conn = dataSource.getConnection();
                     PreparedStatement pstmt = conn.prepareStatement(sql)) {
                    conn.setAutoCommit(false); // 关闭自动提交，开启事务
                    for (var record : records) {
                        // 解析消息，假设消息是JSON格式：{"userId":1001,"behaviorType":"click","content":"product_123","createTime":"2024-05-20 12:00:00"}
                        String json = record.value();
                        // 这里可以用JSON解析库（比如Jackson）解析，为了简化，这里模拟解析过程
                        int userId = 1001;
                        String behaviorType = "click";
                        String content = "product_123";
                        String createTime = "2024-05-20 12:00:00";

                        // 设置参数
                        pstmt.setInt(1, userId);
                        pstmt.setString(2, behaviorType);
                        pstmt.setString(3, content);
                        pstmt.setString(4, createTime);
                        pstmt.addBatch(); // 添加到批量
                    }
                    // 执行批量插入
                    pstmt.executeBatch();
                    conn.commit(); // 提交事务
                    System.out.println("成功插入" + records.count() + "条日志");
                } catch (SQLException e) {
                    e.printStackTrace();
                    // 这里可以加重试逻辑，比如插入失败，把消息存到死信队列，后续处理
                }
            }
        } finally {
            consumer.close();
            dataSource.close();
        }
    }
}

这个代码的逻辑很简单：先配置Kafka消费者，订阅指定的主题；然后配置数据库连接池，方便管理数据库连接；最后循环拉取消息，批量插入到数据库里，用事务保证批量插入的原子性，要么全部成功，要么全部失败。

五、技术优缺点及注意事项

5.1 各方案的优缺点总结

方案	优点	缺点	适用场景
双写	实现简单，不用额外中间件	原子性无保证，易漏写，批量操作性能差	小项目，一致性要求低
CDC	一致性高，业务代码无感知，维护成本低	需要额外部署工具，数据库配置有要求，轻微性能开销	中大型项目，一致性要求高
定时同步	实现简单，不用改数据库配置	实时性差，一致性无保证	非实时场景，比如报表

5.2 注意事项

数据一致性：不管用哪种方案，都要考虑异常情况的处理，比如双写时发Kafka失败了怎么办，CDC时数据库崩溃了怎么办，反向同步时插入数据库失败了怎么办。可以加重试机制，或者把异常消息存到死信队列，后续人工或自动处理。
性能优化：批量操作可以大大提升性能，比如CDC时可以批量同步，反向同步时可以批量插入数据库；另外，Kafka的主题分区数要合理设置，分区数太少会影响并发性能，太多会增加Kafka的管理开销。
数据过滤和转换：很多时候，我们不需要把数据库里的所有字段都同步到Kafka，比如敏感字段（密码、手机号）可以过滤掉，或者对数据做转换，比如把数据库里的时间戳转换成字符串，方便下游系统处理。
版本兼容：不同版本的中间件可能存在兼容问题，比如Debezium和MySQL的版本、Kafka和Debezium的版本，要提前查官方文档，确保版本兼容。
监控和告警：要对同步过程做监控，比如监控Kafka的消息堆积量、CDC工具的运行状态、数据库的连接数，要是出现异常，及时告警，避免数据不一致的问题长时间没发现。

六、文章总结

Kafka和关系型数据库的集成，本质上是解决“核心业务和下游业务解耦”以及“数据一致性”的问题，不同的方案有不同的适用场景，要根据自己的业务需求来选。

如果是小项目，追求快速上线，对一致性要求不高，可以选双写方案；如果是中大型项目，对一致性和实时性要求高，优先选CDC方案；如果是对实时性要求不高的场景，比如报表统计，可以选定时同步方案。

在实践过程中，要注意异常处理、性能优化、版本兼容和监控告警，确保整个同步过程稳定可靠。

围绕“Kafka与关系型数据库的集成方案与数据同步实践”参与讨论

0 条

Kafka与关系型数据库的集成方案与数据同步实践

一、为什么要把Kafka和关系型数据库搭伙用

二、常见的集成方案及适用场景

2.1 方案一：双写（业务代码自己写两边）

2.2 方案二：数据库的CDC（变更数据捕获）

2.3 方案三：定时同步（全量+增量）

三、具体实践：用CDC方案实现MySQL和Kafka的数据同步

四、反向同步：Kafka数据同步到关系型数据库

五、技术优缺点及注意事项

5.1 各方案的优缺点总结

5.2 注意事项

六、文章总结

评论

关联文章

AI助手

一、为什么要把Kafka和关系型数据库搭伙用

二、常见的集成方案及适用场景

2.1 方案一：双写（业务代码自己写两边）

2.2 方案二：数据库的CDC（变更数据捕获）

2.3 方案三：定时同步（全量+增量）

三、具体实践：用CDC方案实现MySQL和Kafka的数据同步

四、反向同步：Kafka数据同步到关系型数据库

五、技术优缺点及注意事项

5.1 各方案的优缺点总结

5.2 注意事项

六、文章总结

同分类面试题

关联文章

AI助手