linux kafka怎么消费数据

在 Linux 下消费 Kafka 数据，常见有 命令行方式 和 编程方式。下面按使用频率给你讲清楚。

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

1️⃣ 前提

• 已安装 Kafka
• 知道：
  • Kafka 地址（如 localhost:9092）
  • Topic 名称（如 test_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-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test_topic \
  --partition 0 \
  --offset 10

三、编程方式消费（示例）

Java（Kafka Consumer）

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
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) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> r : records) {
        System.out.println(r.key() + " : " + r.value());
    }
}

Python（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())

四、常见问题

❓ 消费不到数据

• topic 是否存在：kafka-topics.sh --list
• 是否写错 group
• 是否 offset 已提交

❓ 消费太慢

• 增加 partition
• 增加 consumer 数量（同 group）

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

• Kafka 版本
• 是否用 Docker
• 消费目的是调试还是生产

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