linux kafka怎么消费消息

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

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

Kafka 安装包中自带消费脚本，适合快速验证和调试。

1. 启动一个简单的消费者（从头消费）

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

说明：

• --bootstrap-server：Kafka broker 地址
• --topic：要消费的主题
• --from-beginning：从最早消息开始消费（不加重试只消费新消息）

2. 指定消费者组（推荐）

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

查看消费者组状态：

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

二、消费指定分区的消息

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --partition 0 \
  --offset 10

三、使用 Kafka 工具脚本（旧版本）

如果你使用的是老版本（< 2.0）：

kafka-console-consumer.sh \
  --zookeeper localhost:2181 \
  --topic test-topic \
  --from-beginning

四、使用编程方式消费（示例）

1. 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(Collections.singletonList("test-topic"));

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

2. Python（kafka-python）

pip install kafka-python

from kafka import KafkaConsumer

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

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

五、常见问题排查

1. 消费不到消息？

• 确认 topic 是否存在

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

• 确认有生产者写入
• 检查 --from-beginning 或 auto_offset_reset

2. 消费组卡住？

kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group test-group \
  --reset-offsets --to-earliest --execute

如果你能告诉我：

• Kafka 版本
• 是否使用 SASL / SSL
• 是否要实时监控还是写程序

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