Kafka消费者配置/性能问题

Question

对于我的测试，我在队列中发布了700万条消息。我创建了一个包含30个消费者线程消费者组，每个分区一个。我最初的印象是，与通过SQS获得的相比，这将大大加快处理能力。不幸的是，情况并非如此。在我的例子中，数据处理是复杂的，平均需要1-2分钟才能完成，这导致了一系列分区重新平衡，因为线程不能按时运行。我在日志里看到一堆消息

组FULL_GROUP的自动偏移量提交失败：无法完成提交，因为该组已重新平衡并将分区分配给另一个成员。这意味着对poll（）的后续调用之间的时间比配置的session.timeout.ms长，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加会话超时，或者通过使用max.poll.records减小poll（）中返回的批的最大大小来解决这一问题。

这会导致同一消息被多次处理。我尝试使用会话超时、max.poll.records和轮询时间来避免这种情况，但这会减慢整个处理BigTime的速度。下面是一些配置参数。

metadata.max.age.ms = 300000
max.partition.fetch.bytes = 1048576
bootstrap.servers = [kafkahost1:9092, kafkahost2:9092]
enable.auto.commit = true
max.poll.records = 10000
request.timeout.ms = 310000
heartbeat.interval.ms = 100000
auto.commit.interval.ms = 1000
receive.buffer.bytes = 65536
fetch.min.bytes = 1
send.buffer.bytes = 131072
value.deserializer = class com.autodesk.preprocessor.consumer.serializer.KryoObjectSerializer
group.id = full_group
retry.backoff.ms = 100
fetch.max.wait.ms = 500
connections.max.idle.ms = 540000
session.timeout.ms = 300000
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
metrics.sample.window.ms = 30000
auto.offset.reset = latest

这是我的消费者代码

while (true) {
    try{
        ConsumerRecords records = consumer.poll(100);
        for (ConsumerRecord record : records) {
            if(record.value()!=null){
                TextAnalysisRequest textAnalysisObj = record.value();
                if(textAnalysisObj!=null){
                    // Process record
                    PreProcessorUtil.submitPostProcessRequest(textAnalysisObj);
                }
            }
        }
    }catch(Exception ex){
        LOGGER.error("Error in Full Consumer group worker", ex);
    }

我真的在寻找一些建议/最佳实践在这种情况下。特别是关于心跳、请求超时、最大轮询记录、自动提交间隔、轮询间隔等的推荐配置设置。如果kafka不适合我的用例，请让我知道。

Answer 1

您可以从异步处理消息开始，在一个独立的线程中，而不是从Kafka读取消息的线程。这样自动提交将非常快，Kafaka不会削减您的会话。类似这样的事情：

    private final BlockingQueue<TextAnalysisRequest> requests = 
new LinkedBlockingQueue();

在读取线程中：

while (true) {
    try{
        ConsumerRecords records = consumer.poll(100);
        for (ConsumerRecord record : records) {
            if(record.value()!=null){
                TextAnalysisRequest textAnalysisObj = record.value();
                if(textAnalysisObj!=null){
                    // Process record
                    requests.offer(textAnalysisObj);
                }
            }
     }    
}
catch(Exception ex){
    LOGGER.error("Error in Full Consumer group worker", ex);
}

在处理线程中：

            while (!Thread.currentThread().isInterrupted()) {
                try {
                    TextAnalysisRequest textAnalysisObj = requests.take();
                    PreProcessorUtil.submitPostProcessRequest(textAnalysisObj);
                } catch (InterruptedException e) {
                    LOGGER.info("Process thread interrupted", e);
                    Thread.currentThread().interrupt();
                } catch (Throwable t) {
                    LOGGER.warn("Unexpected throwable while processing.", t);
                }
            }