apache-kafka - Kafka Consumer - 具有更高优先级的主题

Question

我正在使用 Kafka Consumer 读取多个主题，我需要其中一个具有更高的优先级。处理需要很多时间，并且(低优先级)主题中总是有很多消息，但我需要尽快处理来自其他主题的消息。

与 Does Kafka support priority for topic or message? 类似的问题但这个使用的是旧 API。

在新的 API (0.10.1.1) 中，有方法

KafkaConsumer::pause(Collection)
KafkaConsumer::resume(Collection)

但我不清楚，如何有效地检测高优先级主题中有新消息，需要暂停其他主题的消费。

有什么想法/例子吗？

Answer 1

最后我解决了这个问题，正如 dawsaw 建议的那样 - 在处理循环中，我存储了我从中读取的所有主题/分区:

• 开始偏移
• endOffsets
• committed - 我不能使用位置，因为我订阅的是主题，而不是分区。

每当 (endOffset - commited) > 0 对于任何优先主题，我都会为非优先主题调用 consumer.pause() 并在 ( endOffset - 已提交)== 0 对于所有优先主题。