linux kafka怎么消费

在 Linux 上消费 Kafka 消息，常见方式有以下几种（从最常用到编程方式）：

一、使用 Kafka 自带命令行工具（最常用）

Kafka 安装包里自带 kafka-console-consumer.sh。

1️⃣ 从最新位置消费

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic

2️⃣ 从最早消息开始消费

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning

3️⃣ 指定消费者组

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --group my-group

4️⃣ 显示 key

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --property print.key=true

二、查看消费进度（Offset）

bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group my-group

三、使用 Kafka 工具（带 UI）

适合不熟悉命令行的场景：

• kafdrop
• AKHQ
• Kafka Tool（Windows/Linux）

四、用 Python 消费（示例）

安装

pip install kafka-python

消费代码

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    group_id='my-group'
)

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

五、用 Java 消费（核心片段）

KafkaConsumer<String, String> consumer =
    new KafkaConsumer<>(props);

consumer.subscribe(Collections.singletonList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> r : records) {
        System.out.println(r.value());
    }
}

六、常见问题

❓ 消费不到消息？

• 检查 topic 是否存在
• 是否用了 --from-beginning
• 消费者组是否已有 offset

❓ 连接失败？

telnet localhost 9092

确认 Kafka 正在运行

如果你能告诉我：

• Kafka 版本
• 是否用 Docker
• 想用命令行还是代码

我可以给你更精确的示例。