linux kafka怎么连客户端

在 Linux 上连接 Kafka 客户端，通常分几种情况：命令行客户端（自带工具）、Java 客户端、其他语言客户端（Python/Go 等）。下面按常见用法说明。

一、使用 Kafka 自带命令行客户端（最常用）

1. 前提

• Kafka 已安装（如 /opt/kafka）
• 知道 broker 地址，例如：

  192.168.1.100:9092
  

2. 创建 Topic（示例）

cd /opt/kafka/bin

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

3. 生产消息（生产者）

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

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

4. 消费消息（消费者）

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

二、Java 客户端连接 Kafka

1. Maven 依赖

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

2. 生产者示例

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();

3. 消费者示例

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()));
}

三、Python 客户端（kafka-python）

1. 安装

pip install kafka-python

2. 生产者

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='192.168.1.100:9092')
producer.send('test-topic', b'hello kafka')
producer.close()

3. 消费者

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='192.168.1.100:9092',
    auto_offset_reset='earliest'
)

for msg in consumer:
    print(msg.value)

四、常见连接问题排查

1. 连不上 Kafka

• 检查 broker 是否启动

netstat -an | grep 9092

• 防火墙

iptables -L
# 或
firewall-cmd --list-all

2. 远程连接失败（重要）

Kafka 默认 listeners 配置可能导致远程连不上：

server.properties

listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://你的IP:9092

修改后重启 Kafka。

五、总结

如果你能告诉我：

• Kafka 版本
• 是否本地 / 远程
• 用什么语言

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