debian Kafka怎么连集群

在 Debian 上连接 Kafka 集群，通常分三种场景来说：命令行连接、Java 客户端连接、其他语言客户端连接。下面以最常见的 Kafka 2.x / 3.x + Zookeeper 或 KRaft 为例。

一、前提：确认集群信息

连接 Kafka 集群你需要知道：

1. Bootstrap Servers

   broker1:9092,broker2:9092,broker3:9092
   

2. 是否开启认证
   • 无认证（默认）
   • SASL / SSL
3. Topic 是否存在
4. 网络是否通

   telnet broker1 9092
   

二、Debian 命令行连接 Kafka 集群

1. 安装 Kafka 客户端

sudo apt update
sudo apt install -y default-jre

下载 Kafka：

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

2. 使用命令行工具连接集群

列出 Topic

bin/kafka-topics.sh \
  --bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
  --list

生产消息

bin/kafka-console-producer.sh \
  --bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
  --topic test

消费消息

bin/kafka-console-consumer.sh \
  --bootstrap-server broker1:9092,broker2:9092,broker3:9092 \
  --topic test \
  --from-beginning

三、Java 客户端连接 Kafka 集群

1. Maven 依赖

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

2. 生产者示例

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("test", "hello kafka"));
producer.close();

3. 消费者示例

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3: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"));

while (true) {
    consumer.poll(Duration.ofMillis(100))
            .forEach(r -> System.out.println(r.value()));
}

四、带 SASL/SSL 的集群连接（常见生产环境）

1. 客户端配置示例

bootstrap.servers=broker1:9093,broker2:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
ssl.truststore.location=/etc/kafka/client.truststore.jks
ssl.truststore.password=123456
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="user" \
  password="pass";

五、常见问题排查

1. 连不上

# 检查端口
telnet broker1 9092

# 检查 hosts
cat /etc/hosts

2. 报 LEADER_NOT_AVAILABLE

bin/kafka-topics.sh --bootstrap-server ... --describe --topic test

3. Debian 防火墙

sudo ufw allow 9092

六、总结

核心就一句话：

Kafka 客户端只需要 bootstrap.servers，剩下的交给集群自己路由。

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

• 你是 连接已有集群 还是 自己搭集群
• 是否使用 SASL / SSL
• 用 命令行 / Java / Python

我可以直接给你一份可复制的配置。