Kafka生产者如何获取并将已发布的Avro消息持久化至数据库？

方案一：生产者端发送后直接持久化 #

Kafka生产者发送消息时，可通过回调机制确认消息发送成功，随后直接将Avro消息写入数据库。需要注意的是，你得在发送前保留原始Avro对象的引用（或者在回调中反序列化字节数据）——因为生产者实际发送的是序列化后的字节流。

代码示例（Java） #

// 假设已通过Avro工具生成实体类User
User avroUser = User.newBuilder()
    .setId(1)
    .setName("Alice")
    .build();

ProducerRecord<String, User> record = new ProducerRecord<>("user-topic", avroUser);

kafkaProducer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            // 消息发送成功，将avroUser写入数据库
            try (Connection conn = getDbConnection()) {
                String sql = "INSERT INTO users (id, name) VALUES (?, ?)";
                PreparedStatement stmt = conn.prepareStatement(sql);
                stmt.setInt(1, avroUser.getId());
                stmt.setString(2, avroUser.getName());
                stmt.executeUpdate();
            } catch (SQLException e) {
                // 处理数据库写入异常
                e.printStackTrace();
            }
        } else {
            // 处理消息发送失败情况
            exception.printStackTrace();
        }
    }
});

如果发送时仅传递了序列化后的字节数组，你需要借助对应的Avro Schema反序列化来还原对象：

// 假设已有Schema对象和字节数据
User avroUser = User.fromByteBuffer(ByteBuffer.wrap(record.value()));

方案二：用Kafka消费者消费后持久化（推荐） #

更贴合Kafka架构设计的做法是让生产者专注于消息发送，用独立的消费者订阅目标Topic，消费Avro消息后再写入数据库。这种方式解耦了生产与持久化逻辑，也便于后续扩展（比如多消费者并行处理）。

关键配置 #

消费者需配置Avro反序列化器及Schema Registry地址：

bootstrap.servers=localhost:9092
group.id=avro-db-sync-group
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
schema.registry.url=http://localhost:8081
specific.avro.reader=true  # 启用特定Avro类的反序列化

代码示例（Java） #

KafkaConsumer<String, User> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("user-topic"));

while (true) {
    ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
    try (Connection conn = getDbConnection()) {
        conn.setAutoCommit(false); // 开启批量事务
        String sql = "INSERT INTO users (id, name) VALUES (?, ?)";
        PreparedStatement stmt = conn.prepareStatement(sql);
        
        for (ConsumerRecord<String, User> record : records) {
            User avroUser = record.value();
            stmt.setInt(1, avroUser.getId());
            stmt.setString(2, avroUser.getName());
            stmt.addBatch();
        }
        
        stmt.executeBatch();
        conn.commit();
        consumer.commitSync(); // 提交消费偏移量
    } catch (SQLException e) {
        // 回滚事务并处理异常
        e.printStackTrace();
    }
}

注意事项 #

• Schema Registry依赖：如果使用Confluent的Avro序列化/反序列化器，必须保证Schema Registry服务正常运行，且生产者、消费者都配置了正确的schema.registry.url。
• 事务一致性：若要保证消息发送与数据库写入的原子性，可使用Kafka事务生产者（配置transactional.id），或在消费者端通过事务确保消费偏移量提交与数据库操作的一致性。
• 异常处理：无论采用哪种方案，都要处理消息发送失败、数据库写入失败、网络异常等场景，避免数据丢失或重复写入。

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