如何将经业务处理后存入RDBMS表的记录发布至Kafka Topic？

针对你现有工作流（上游生成数据→Transformer处理→存入RDBMS），要把表中结果发布到Kafka Topic，以下是几种实用的实现方案：

方案1：数据库CDC（变更数据捕获）方式 #

• 核心逻辑：监听RDBMS表的增改操作，自动捕获变更数据并推送至Kafka
• 实现步骤：
  • 使用Debezium、Canal这类CDC工具，配置对应数据库的连接器，指定监听目标表
  • 示例Debezium MySQL连接器配置片段：

    name=mysql-workflow-connector
    connector.class=io.debezium.connector.mysql.MySqlConnector
    tasks.max=1
    database.hostname=your-db-host
    database.port=3306
    database.user=your-db-user
    database.password=your-db-pass
    database.server.id=19260817
    database.server.name=workflow-db
    database.include.list=your_target_db
    table.include.list=your_target_db.your_result_table
    topic.prefix=workflow-cdc
    

  • 优势：实时性强，无需修改现有业务代码，自动同步所有数据变更
  • 局限：需部署维护CDC组件，数据库需开启binlog（MySQL）或对应日志功能，对数据库权限有要求

方案2：在Transformer模块新增Kafka推送逻辑 #

• 核心逻辑：在Transformer处理完数据并写入RDBMS的环节，同步将处理结果发送到Kafka
• 实现步骤：
  • 集成Kafka客户端（如Java的kafka-clients、Python的confluent-kafka），在数据成功写入DB后调用生产者API发送消息
  • 示例Java代码片段：

    // 假设processedResult是处理后的业务对象
    ObjectMapper objectMapper = new ObjectMapper();
    String message = objectMapper.writeValueAsString(processedResult);
    ProducerRecord<String, String> kafkaRecord = new ProducerRecord<>(
        "your-target-topic", 
        processedResult.getUniqueId(), 
        message
    );
    // 发送消息并处理回调
    kafkaProducer.send(kafkaRecord, (metadata, ex) -> {
        if (ex != null) {
            // 记录失败日志或触发重试
            log.error("Push to Kafka failed for record ID: {}", processedResult.getUniqueId(), ex);
        }
    });
    

  • 优势：逻辑直接，无需额外组件，可通过事务或本地消息表保证DB写入与Kafka推送的一致性
  • 局限：需要修改现有业务代码，耦合了业务逻辑与消息推送逻辑

方案3：定时拉取表数据推送 #

• 核心逻辑：通过定时任务定期查询RDBMS中未推送的数据，推送至Kafka后标记状态
• 实现步骤：
  • 在结果表新增is_pushed（布尔型）或push_timestamp（时间型）字段，标记数据是否已推送
  • 用Quartz、Spring Task或系统级定时任务（如Linux crontab）执行查询+推送+更新操作
  • 示例查询SQL：

    SELECT * FROM your_result_table WHERE is_pushed = 0 ORDER BY create_time LIMIT 200;
    

  • 优势：实现简单，对现有系统侵入极小，适合非实时、低频次的业务场景
  • 局限：实时性差，需处理重复推送问题（可通过Kafka消息键做幂等），定时任务需考虑并发冲突

内容的提问来源于stack exchange，提问作者SPD