Kafka consumer configuration / performance issues
我正在尝试将 kafka 作为 AWS SQS 的替代方案。动机主要是提高性能，其中 kafka 将消除一次提取 10 条消息的限制，上限为 256kb。这是我的用例的高级场景。我有一堆正在发送文档以进行索引的爬虫。有效载荷的大小平均约为 1 MB。爬虫调用 SOAP 端点，后者又运行生产者代码以将消息提交到 kafka 队列。消费者应用程序获取消息并处理它们。对于我的测试框，我为主题配置了 30 个分区和 2 个复制。两个 kafka 实例与 1 个 zookeeper 实例一起运行。 kafka 版本为 0.10.0.
在我的测试中，我在队列中发布了 700 万条消息。我创建了一个具有 30 个消费者线程的消费者组，每个分区一个。我最初的印象是，与我通过 SQS 获得的相比，这将大大提高处理能力。不幸的是，事实并非如此。就我而言，数据处理很复杂，平均需要 1-2 分钟才能完成。由于线程无法按时心跳，这导致了一系列分区重新平衡。我可以在日志中看到一堆引用
的消息
Auto offset commit failed for group full_group: Commit cannot be
completed since the group has already rebalanced and assigned the
partitions to another member. This means that the time between
subsequent calls to poll() was longer than the configured
session.timeout.ms, which typically implies that the poll loop is
spending too much time message processing. You can address this either
by increasing the session timeout or by reducing the maximum size of
batches returned in the poll() with max.poll.records.
这会导致同一条消息被多次处理。我尝试使用会话超时、max.poll.records 和轮询时间来避免这种情况，但这会减慢整体处理时间。这是一些配置参数。
|
1
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
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 |
我将消费者轮询时间减少到 100 毫秒。它减少了重新平衡问题，消除了重复处理，但显着减慢了整个过程。最终完成处理所有 600 万条消息需要 35 小时，而使用基于 SQS 的解决方案则需要 25 小时。每个消费者线程平均每次轮询检索 50-60 条消息，尽管其中一些有时轮询 0 条记录。当分区中有大量消息可用时，我不确定这种行为。在随后的迭代中，同一线程能够获取消息。这可能是由于重新平衡造成的吗？
这是我的消费者代码
|
1
2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
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); } |
我了解记录处理部分是我的瓶颈之一。但我敢肯定，这里的一些人也有类似的处理大量处理时间的用例。我想过通过在其专用线程中旋转每个处理器或使用大容量线程池来进行异步处理，但不确定它是否会在系统中产生很大的负载。同时，我看到了几个例子，人们使用暂停和恢复 API 来执行处理以避免重新平衡问题。
在这种情况下，我真的在寻找一些建议/最佳做法。特别是关于 heartbeat、请求超时、最大轮询记录、自动提交间隔、轮询间隔等的推荐配置设置。如果 kafka 不是我用例的正确工具，请告诉我。
您可以首先在与从 Kafka 读取的线程不同的线程中异步处理消息。这种方式自动提交将非常快，并且 Kafka 不会中断您的会话。像这样的东西：
|
1
2 |
private final BlockingQueue<TextAnalysisRequest> requests =
new LinkedBlockingQueue(); |
在阅读线程中：
|
1
2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
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); } |
在处理线程中：
|
1
2 3 4 5 6 7 8 9 10 11 |
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); } } |
还请查看此文档，了解通过 Kafka 发送大量消息的策略：http://blog.cloudera.com/blog/2015/07/deploying-apache-kafka-a-practical-faq/
简而言之，它说 Kafka 在 10K 左右的小消息上表现最好，如果您需要发送较大的消息，最好将它们放在网络存储上，并通过 Kafka 仅发送它们的位置，或者将它们拆分。
原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/267595.html