Kafka消费者组在Debian上如何配置

5月 5th, 2025

在Debian系统上配置Kafka消费者组，你需要遵循以下步骤：

安装Java：
Kafka需要Java运行环境，因此首先确保你的Debian系统上安装了Java。你可以使用以下命令来安装OpenJDK：
```
sudo apt update
sudo apt install openjdk-11-jdk
```
下载并解压Kafka：
从Apache Kafka官方网站下载最新版本的Kafka，并将其解压到你希望的目录。
```
wget https://downloads.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0
```
请根据实际情况替换下载链接中的版本号。

启动Zookeeper和Kafka服务器：
Kafka使用Zookeeper进行集群管理，所以你需要先启动Zookeeper，然后启动Kafka服务器。

# 启动Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# 在另一个终端中启动Kafka服务器
bin/kafka-server-start.sh config/server.properties

创建主题（可选）：
如果你还没有创建Kafka主题，可以使用以下命令创建一个：
```
bin/kafka-topics.sh --create --topic your_topic_name --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1
```
替换your_topic_name为你想要的主题名称。

配置消费者组：
创建一个消费者配置文件，例如consumer.properties，并添加以下内容：

bootstrap.servers=localhost:9092
group.id=your_group_id
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

替换your_group_id为你的消费者组ID。

编写消费者应用程序：
使用你喜欢的编程语言编写消费者应用程序。以下是一个简单的Java消费者示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.load(SimpleConsumer.class.getResourceAsStream("/consumer.properties"));

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("your_topic_name"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                records.forEach(record -> {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                });
            }
        } finally {
            consumer.close();
        }
    }
}