ubuntu下kafka如何连接

在 Ubuntu 下使用 Kafka，通常分为 Kafka 连接 Kafka（生产/消费） 和 客户端连接 Kafka 两种情况。下面从最常见场景一步步说明。

一、前提：Kafka 已安装并启动

1️⃣ 安装 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️⃣ 启动 Zookeeper（Kafka 依赖）

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

3️⃣ 启动 Kafka

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

✅ 默认 Kafka 监听：

localhost:9092

二、本地连接 Kafka（最常见）

1️⃣ 创建 Topic

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

2️⃣ 生产者连接（发送消息）

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

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

3️⃣ 消费者连接（接收消息）

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

✅ 到这一步，本地 Kafka 已成功连接。

三、远程连接 Kafka（重点）

问题原因

Kafka 默认只监听 localhost，远程无法连接。

解决方法：修改 server.properties

vim config/server.properties

✅ 关键配置：

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

例如：

advertised.listeners=PLAINTEXT://192.168.1.100:9092

然后重启 Kafka：

bin/kafka-server-stop.sh
bin/kafka-server-start.sh -daemon config/server.properties

四、客户端连接 Kafka（Java 示例）

Maven 依赖

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

生产者示例

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

消费者示例

Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.100: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-topic"));

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

五、常见问题排查

❌ 连接超时

• 防火墙未放行 9092

sudo ufw allow 9092

❌ 连接成功但无法消费

• advertised.listeners 配置错误
• 客户端使用了 localhost

❌ Kafka 启动失败

tail -f logs/server.log

六、快速总结

如果你能告诉我：

• ✅ Kafka 是本机还是服务器？
• ✅ 用命令行还是 Java / Python / Go？
• ✅ 是否 Docker 部署？

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