当主题中没有更多记录时如何在Kafka Consumer中刷新数据批处理

Question

考虑这个 Kafka 消费者，它从主题接收数据，将其缓冲到 PreparedStatement 中，当批处理 10 万条记录时，它会向数据库发出 INSERT 查询。

这在数据仍然传入之前运行良好。但是，例如，当缓冲 20K 记录并且没有更多记录传入时，它仍会等待更多 80K 记录，直到在 flushes 语句中。但是如果在一段时间后停止，我想刷新那些 20K。我怎样才能做到这一点？我看不出有什么方法可以抓住它。

例如，在使用基于 librdkafka 的 php-rdkafka 扩展的 PHP 中，当达到分区末尾时，我会得到 RD_KAFKA_RESP_ERR__PARTITION_EOF，因此在发生这种情况时很容易挂钩缓冲区刷新。

我尝试简化代码，只保留重要部分

public class TestConsumer {

    private final Connection connection;
    private final CountDownLatch shutdownLatch;
    private final KafkaConsumer<String, Message> consumer;
    private int processedCount = 0;

    public TestConsumer(Connection connection) {
        this.connection = connection;
        this.consumer = new KafkaConsumer<>(getConfig(), new StringDeserializer(), new ProtoDeserializer<>(Message.parser()));
        this.shutdownLatch = new CountDownLatch(1);
    }

    public void execute() {
        PreparedStatement statement;
        try {
            statement = getPreparedStatement();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            commit(statement);

            consumer.wakeup();
        }));

        consumer.subscribe(Collections.singletonList("source.topic"));

        try {
            while (true) {
                ConsumerRecords<String, Message> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));

                records.forEach(record -> {
                    Message message = record.value();
                    try {
                        fillBatch(statement, message);
                        statement.addBatch();
                    } catch (SQLException e) {
                        throw new RuntimeException(e);
                    }
                });

                processedCount += records.count();

                if (processedCount > 100000) {
                    commit(statement);
                }
            }
        } catch (WakeupException e) {
            // ignore, we're closing
        } finally {
            consumer.close();
            shutdownLatch.countDown();
        }
    }

    private void commit(PreparedStatement statement) {
        try {
            statement.executeBatch();
            consumer.commitSync();
            processedCount = 0;
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }


    protected void fillBatch(PreparedStatement statement, Message message) throws SQLException {
        try {
            statement.setTimestamp(1, new Timestamp(message.getTime() * 1000L));
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
    }

Answer 1

我理解你的问题是这样的：

• 你想消费来自 Kafka 的消息

• 在内存中堆积最多 10 万条记录

• 批量提交到数据库

• 但您只想等待 t 秒（假设是 10 秒）

这可以通过使用 Kafka 内置的消费者批处理以非常有效和可靠的方式实现。如果您可以以某种方式预测消息的平均大小（以字节为单位）。

在 Kafka 消费者配置中，您可以设置以下内容：

fetch.min.bytes => 这应该是 100k x 消息的平均大小

fetch.max.wait.ms => 这是您的超时时间（以毫秒为单位）（例如 5000 等待 5 秒）

max.partition.fetch.bytes => 最大值。每个分区的数据量。这有助于优化总提取大小

max.poll.records => 单个轮询中返回的最大记录数..可以设置为 100K

fetch.max.bytes => 如果要设置单个请求的上限

这样，如果它们符合定义的字节大小，您最多可以获得 100K 记录，但它会等待可配置的毫秒数。

一旦民意调查返回记录，您可以一次性保存并重复。