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

Question

考虑从主题接收数据的Kafka消费者，将其缓冲成PravaReDealStand，当100K记录被批处理时，它将插入查询向DB发出。

直到数据仍在输入之前，这种方法都能很好地工作。但是，例如，当缓冲了20K条记录并且没有更多的记录传入时，它仍然会等待更多的80K条记录，直到in刷新语句。但如果一段时间后停止运行，我想冲洗这些20K。我该怎么做？我看不出有什么办法能抓住它。

例如，在PHP中，使用基于库的php-rdkafka扩展，当分区结束时，我会得到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的消息

把它们堆在记忆里100K记录

批量提交到数据库

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

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

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

fetch。最小字节数=

fetch。麦克斯，等等。ms=

max.partition。取来字节=

max.poll。记录=

fetch。最大字节数=

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

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