Linux Kafka如何连接

“Linux Kafka 如何连接”通常取决于你是谁、用什么方式连接。下面按最常见场景给你一份从入门到实战的 Linux Kafka 连接指南。

一、前提条件（必须）

1️⃣ Kafka 已启动

# 启动 Zookeeper（老版本需要）
bin/zookeeper-server-start.sh config/zookeeper.properties

# 启动 Kafka
bin/kafka-server-start.sh config/server.properties

Kafka 3.x+ 已逐步移除 Zookeeper，但仍兼容。

2️⃣ 确认 Kafka 监听地址

查看 server.properties：

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

✅ 客户端连接的是 advertised.listeners

二、Linux 本地连接方式（最常用）

✅ 方式一：使用 Kafka 自带命令行工具（推荐）

1️⃣ 创建 Topic

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

2️⃣ 生产消息（Producer）

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

输入消息后回车即可发送。

3️⃣ 消费消息（Consumer）

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

✅ 这是 Linux 下最标准的连接方式

三、远程 Linux 服务器连接 Kafka

✅ 场景：Kafka 在远程服务器

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

⚠️ 常见问题：

• 防火墙未放行 9092
• advertised.listeners 配置的是 localhost

✅ 正确示例：

advertised.listeners=PLAINTEXT://192.168.1.100:9092

四、Java 程序连接 Kafka（Linux / 本地 / 远程）

1️⃣ Maven 依赖

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

2️⃣ Producer 示例

Properties props = new Properties();
props.put("bootstrap.servers", "localhost: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-topic", "hello kafka"));
producer.close();

3️⃣ Consumer 示例

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("auto.offset.reset", "earliest");
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-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    records.forEach(r ->
        System.out.println(r.key() + " => " + r.value())
    );
}

五、Docker 中 Linux Kafka 连接

docker exec -it kafka bash

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

六、常见连接失败原因（重点）

七、快速自检命令

# 查看端口
ss -lntp | grep 9092

# 测试连通
telnet 127.0.0.1 9092

八、你可以继续问我

如果你愿意，可以告诉我： 1️⃣ Kafka 是 单机 / 集群 / Docker / K8s？
2️⃣ 是 命令行 / Java / Python / Go 连接？
3️⃣ 报错信息是什么？

我可以直接帮你 定位并解决连接问题 ✅