linux kafka怎样消费

在 Linux 环境下消费 Kafka 消息，常见方式有以下几种，从最常用命令行到编程方式都给你说明。

一、使用 Kafka 自带命令行消费（最常用）

1️⃣ 前提

• 已安装 Kafka
• Kafka 服务已启动
• 知道：
  • Kafka 地址（如 localhost:9092）
  • Topic 名称

2️⃣ 从最新位置消费

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test_topic

3️⃣ 从最早消息开始消费（重要）

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test_topic \
  --from-beginning

4️⃣ 指定消费者组（推荐）

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test_topic \
  --group my_group \
  --from-beginning

查看消费进度：

kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group my_group

5️⃣ 输出为 JSON（可选）

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test_topic \
  --property print.key=true \
  --property key.separator=":"

二、使用 Kafka 工具（带 UI，方便）

常见工具

• Kafka Tool
• OffsetExplorer
• Kafka Eagle
• CMAK（Kafka Manager）

适合：

• 调试
• 查看消息内容
• 管理 offset

三、使用 Python 消费 Kafka（示例）

安装

pip install kafka-python

消费示例

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "test_topic",
    bootstrap_servers="localhost:9092",
    group_id="my_group",
    auto_offset_reset="earliest"
)

for msg in consumer:
    print(msg.value.decode())

四、使用 Java 消费（简单示例）

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("test_topic"));

while (true) {
    consumer.poll(Duration.ofMillis(100))
            .forEach(record -> System.out.println(record.value()));
}

五、常见问题排查

❌ 消费不到消息

• 是否用 --from-beginning
• topic 是否真的有数据
• group 是否已消费过（offset 已提交）

❌ 连接失败

telnet localhost 9092

❌ 查看所有 topic

kafka-topics.sh --bootstrap-server localhost:9092 --list

如果你愿意，可以告诉我：

• Kafka 版本
• 是否用 Docker
• 是否集群
• 用什么语言消费

我可以给你更精准的命令或示例。