debian kafka消费者如何配

下面按Debian 上 Kafka 消费者配置这个主题，分「环境准备 → 配置文件 → 命令行消费 → 代码消费（Java/Python）→ 常见参数」来说明。

一、环境准备（Debian）

1. 安装 Java（Kafka 依赖）

sudo apt update
sudo apt install -y openjdk-17-jdk
java -version

2. 下载并解压 Kafka

wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz
tar -xzf kafka_2.13-3.7.0.tgz
cd kafka_2.13-3.7.0

3. 启动 Kafka（本地测试）

# 启动 Zookeeper（老版本）或 KRaft（新版本）
bin/zookeeper-server-start.sh config/zookeeper.properties

# 另一个终端
bin/kafka-server-start.sh config/server.properties

二、消费者核心配置文件

Kafka 消费者一般不需要单独配置文件，大部分参数通过：

• 命令行参数
• 代码配置
• 或 consumer.properties

示例 consumer.properties

bootstrap.servers=localhost:9092
group.id=test-group
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
auto.offset.reset=earliest
enable.auto.commit=true

三、命令行消费（最常用）

1. 从头消费

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

2. 指定消费者组

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

3. 使用配置文件

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

四、Java 消费者配置示例

Maven 依赖

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.7.0</version>
</dependency>

Java 代码

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");
props.put("auto.offset.reset", "earliest");

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

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

五、Python 消费者（kafka-python）

pip install kafka-python

from kafka import KafkaConsumer

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

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

六、常见重要参数说明

七、排查建议（Debian）

# 查看消费组
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test-group

# 查看 topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

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

• Kafka 版本
• 是否用 KRaft 还是 Zookeeper
• 用 Java / Python / Go
• 是否集群环境

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