如何高效统计1分钟区间内指定Kafka主题的发布消息总数量

kafka-console-consumer.sh 不适合该场景：该工具底层会拉取完整消息正文，即使屏蔽控制台输出，消息体仍会走磁盘IO、网络传输流程，开销极高，大流量下甚至会影响集群正常运行。
以下两种方案均基于Broker端偏移量索引计算，无需读取任何消息内容，统计耗时通常在百毫秒级，完全满足高效要求：

方案1：时间戳偏移量差值法（原生无依赖，最推荐） #

原理：Kafka会为每条消息存储写入时间戳，Broker支持直接根据时间戳查询对应分区的消息偏移量，将统计窗口结束时刻的总偏移量减去起始时刻的总偏移量，结果就是窗口内的消息总数。
落地步骤：

• 先计算统计窗口的毫秒级起止时间戳，以统计最近1分钟数据为例，直接在shell中执行以下命令生成时间参数：

  # 起始时间为当前时间往前推1分钟
  START_TS=$(($(date +%s%3N) - 60000))
  # 结束时间为当前时间
  END_TS=$(date +%s%3N)
  

• 查询起始时间点对应主题所有分区的偏移量总和：

  kafka-run-class.sh kafka.tools.GetOffsetShell \
    --broker-list <替换为你的Broker地址，格式为ip1:9092,ip2:9092> \
    --topic <替换为待统计的主题名> \
    --time $START_TS \
    | awk -F ":" '{sum += $3} END {print sum}'
  

  命令输出值记为start_offset
• 查询结束时间点对应主题所有分区的偏移量总和：

  kafka-run-class.sh kafka.tools.GetOffsetShell \
    --broker-list <替换为你的Broker地址> \
    --topic <替换为待统计的主题名> \
    --time $END_TS \
    | awk -F ":" '{sum += $3} END {print sum}'
  

  命令输出值记为end_offset
• 最终1分钟窗口内的消息总数 = end_offset - start_offset

提示：如果统计周期内主题发生了分区增减，先对两次查询返回的分区列表做交集，只计算共有分区的偏移量差值，避免结果误差。如果需要固定每1分钟自动统计，把上述逻辑封装成shell脚本加crontab定时任务即可，资源消耗极低。

方案2：临时消费组Lag统计法（适合持续滚动统计） #

如果需要长期滚动统计每1分钟的消息流入量，可以用专属消费组的Lag值做统计，全程不需要消费消息：

• 首次执行前，先把临时统计消费组的位点重置到主题最新位置，作为统计起点：

  kafka-consumer-groups.sh \
    --bootstrap-server <替换为你的Broker地址> \
    --group kafka_temp_stats_group \
    --topic <替换为待统计的主题名> \
    --reset-offsets --to-latest --execute
  

• 等待1分钟后，查询该消费组在目标主题上的总Lag值，这个值就是过去1分钟新写入的消息总数：

  kafka-consumer-groups.sh \
    --bootstrap-server <替换为你的Broker地址> \
    --group kafka_temp_stats_group \
    --describe \
    | grep <替换为待统计的主题名> \
    | awk '{sum += $5} END {print sum}'
  

• 单次统计完成后，再次执行第一步的位点重置命令，把消费组位点推到最新，即可开始下一个1分钟周期的统计。

内容的提问来源于stack exchange，提问作者Alexandre Cojot