kafka消费者debian咋配

下面给你一份**在 Debian 上配置 Kafka 消费者（命令行 + Java 示例）**的实用说明，从环境到跑通一条消息。

一、前置条件（Debian）

1. 安装 Java

Kafka 需要 JDK 8 或以上：

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

2. 安装 Kafka（Debian）

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

二、启动 Kafka（本地测试）

1. 启动 Zookeeper（Kafka 3.3+ 可不用，但简单起见）

/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties

2. 启动 Kafka Broker

/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties

三、创建 Topic

/opt/kafka/bin/kafka-topics.sh \
  --create \
  --topic test-topic \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

四、命令行消费者（最常用）

1. 启动消费者

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

参数说明：

• --from-beginning：从最早消息开始消费
• 不加则只消费启动后的新消息

2. 测试生产者

另开一个终端：

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

输入：

hello kafka

消费者终端会显示：

hello kafka

五、消费者配置（重要参数）

常见消费者配置（consumer.properties）：

bootstrap.servers=localhost:9092
group.id=test-group
auto.offset.reset=earliest
enable.auto.commit=true

使用方式：

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

六、Java 消费者（Debian 项目示例）

1. Maven 依赖

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

2. 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-topic"));

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

七、常见问题（Debian）

1. 连接不上

• 检查 server.properties 中：

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://<服务器IP>:9092

2. 消费者不消费

• group.id 是否重复
• auto.offset.reset 是否设置

如果你是要：

• 用 systemd 管理 Kafka
• 配 SASL / SSL
• 用 Python / Go 消费者

告诉我，我可以直接给你对应配置。