使用Shell脚本消费Kafka消息的脚本严谨性及最优实践咨询

• 缺少变量判空逻辑：如果 grep 未匹配到业务 Topic，$TOPIC 为空会直接导致后续消费者命令报错
• 地址硬编码：ZooKeeper、bootstrap-server 地址固定写死，更换环境后无法直接使用
• 无前置错误捕获：如果 kafka-topics.sh 执行失败（比如 ZooKeeper 连接异常），没有提前终止逻辑，会直接把空值传给 TOPIC 变量
• 不兼容低版本 Kafka：0.9.x 及更早版本的控制台消费者默认使用旧消费者实现，必须添加 --new-consumer 参数才能连接 bootstrap-server，否则会执行报错
• Shell 写法不规范：变量引用没有加双引号，遇到含特殊字符/空格的内容会出现异常；反引号的命令替换写法可读性差，更推荐用$()格式

如果仅用于功能可用性验证、监控探活、快速测试这类轻量化场景，用 Shell 调用 Kafka 自带的控制台脚本是完全合理的实践，不需要额外开发其他语言代码，运维成本极低。
但如果是要做业务逻辑处理、高吞吐量消费、精确偏移量管理这类复杂场景，不建议用 Shell 实现：控制台消费者的输出格式化难度高、错误处理能力弱、偏移量管理不灵活，性能也远低于官方原生客户端。

#!/bin/bash
# 可配置参数，根据实际环境修改
KAFKA_HOME="/home/kafka/kafka"
ZK_ADDR="localhost:2181"
BOOTSTRAP_ADDR="localhost:9092"
QUERY_TIMEOUT=3

# 前置依赖检查
if [ ! -x "${KAFKA_HOME}/bin/kafka-topics.sh" ] || [ ! -x "${KAFKA_HOME}/bin/kafka-console-consumer.sh" ]; then
    echo "Kafka 执行脚本不存在或无执行权限"
    exit 1
fi

# 获取第一个非内置的业务Topic
TOPIC="$(${KAFKA_HOME}/bin/kafka-topics.sh --zookeeper "${ZK_ADDR}" --list 2>/dev/null | grep -v "__consumer_offsets" | grep -m 1 "[a-zA-Z]")"
if [ -z "${TOPIC}" ]; then
    echo "未找到可用的业务 Topic"
    exit 1
fi

# 检测Kafka版本，判断是否需要加--new-consumer参数
KAFKA_VERSION="$(${KAFKA_HOME}/bin/kafka-topics.sh --version 2>&1 | awk -F'[ .]' '{print $1"."$2}')"
EXTRA_CONSUMER_ARGS=""
# 0.9及更早版本需要加--new-consumer参数
if [ $(echo "${KAFKA_VERSION} <= 0.9" | bc) -eq 1 ]; then
    EXTRA_CONSUMER_ARGS="--new-consumer"
fi

# 执行消费验证
timeout -k ${QUERY_TIMEOUT}s ${QUERY_TIMEOUT}s "${KAFKA_HOME}/bin/kafka-console-consumer.sh" \
    --bootstrap-server "${BOOTSTRAP_ADDR}" \
    --topic "${TOPIC}" \
    --from-beginning \
    --max-messages 1 \
    ${EXTRA_CONSUMER_ARGS} \
    &>/dev/null

if [ $? -eq 0 ]; then
    echo "消费验证成功"
else
    echo "消费验证失败"
fi

• 轻量化验证/监控场景：直接使用上述优化后的 Shell 脚本即可，逻辑简单无额外依赖，适配绝大多数Kafka版本
• 复杂业务消费场景：推荐使用 Kafka 官方原生客户端（Java/Python/Go 等）实现，可以精确管理偏移量、自定义消费逻辑、配置错误重试机制，性能和稳定性远高于 Shell 调用控制台脚本的方式

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