日志上报系统Kafka消息队列优化实践（下）

消费模块已经运行一段时间了，如何查看消费的状况呢？

KafkaOffsetMonitor

KafkaOffsetMonitor是一个可以用于监控Kafka的Topic及Consumer消费状况的工具。

1. Visualizations选项卡 > Cluster Overview：查看当前Kafka集群的Broker情况；
2. Topic List选项卡：查看当前Kafka集群上topic列表

有两个Consumer正在消费当前topic.png

点击进入Consumer可以查看该Consumer当前的消费状况。

• 左上角显示了当前Topic的生产速率
• 右上角显示了当前Consumer的消费速率。
• 蓝色线条表示当前Topic中的Message数目
• 灰色线条表示当前Consumer消费的offset位置；
• 红色线条表示蓝色灰色的差值，即当前Consumer滞后于Producer的message数目。

image.png image.png

3. Consumer Groups项卡：查看各partition中的message消费情况

• topic：创建时topic名称
• partition：分区编号
• offset：表示该parition已经消费了多少条message
• logSize：表示该partition已经写了多少条message
• Lag：表示有多少条message没有被消费。
• Owner：表示消费者
• Created：该partition创建时间
• Last Seen：消费状态刷新最新时间。

当前有4个Partition，每个Partition中的message数目分布均匀。其各Partition的Lag(消息堆积)几乎为0，表示该消费者组的消费能力是可以的。刷新页面会发现Lag全变成0，说明消息很快就被消费掉了。

image.png 消息堆积严重情况.png

查看Kafka Topic流量 可以知道当前服务状态，上报数据量的具体情况。

image.png
image.png

总结

1. 根据上述数据，生成速率和消费速率基本一致，目前状态一个消费者完全hold住；
   Q：如果消息堆积严重，消费速率远远低于生产速度，如何提升呢？
   在当前消费组中创建4个消费者，指定每个消费者消费其中一个分区的消息，在保证消息不会被重复消费的前提下提升消费速率。
2. 生成时消息丢失：当重启服务的时候，服务会重新实例化会导致日志被清空，如果此时Flume没有上报会导致部分数据丢失。但Flume秒级别的延迟这种清空丢失的数据非常非常少。我们系统对数据有一定的容错率。
3. 消费时消息丢失：初始化ConsumerGroup的时候我们设置 fromOffset值为 latest。
   默认情况下，当消费者拉取到了分区的某个消息之后，消费者会自动提交了 offset。设置为latest：当各分区下有已提交当offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。
   🤔️ 如果消费者刚拿到这个消息准备进行真正消费的时候，突然挂掉了，消息实际上并没有被消费，但是 offset 却被自动提交了... 这怎么办呢？关闭闭自动提交 offset，每次在真正消费完消息之后之后再自己手动提交 offset 。
4. 如何避免重复消费？同一主题的一条消息只能被同一个消费组下的某一个消费者消费

好文推荐：

如何保证Kafka不丢失消息