ubuntu kafka如何配置生产者

在 Ubuntu 上配置 Kafka 生产者，通常分三步：安装 Kafka → 启动服务 → 配置并使用生产者。下面以 Kafka 3.x（KRaft 或 Zookeeper 模式均可） 为例说明。

一、前提：安装并启动 Kafka

1. 安装 Java

sudo apt update
sudo apt install openjdk-17-jdk -y
java -version

2. 下载并解压 Kafka

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

二、启动 Kafka（以 KRaft 为例，更简单）

1. 生成 Cluster ID

KAFKA_CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)

2. 格式化存储

bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/kraft/server.properties

3. 启动 Kafka

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

如果用 Zookeeper：

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

三、创建 Topic（生产者需要先有 Topic）

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

查看 Topic：

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

四、命令行生产者配置（最常用）

1. 最简单方式（使用默认配置）

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

然后直接输入消息：

hello kafka
this is a test

2. 使用生产者配置文件

Kafka 自带示例配置：

config/producer.properties

示例配置内容：

bootstrap.servers=localhost:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
acks=all
retries=3
linger.ms=10
batch.size=16384

使用配置文件启动：

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

五、Java 生产者配置示例（常用）

Maven 依赖

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

Java 代码

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");
props.put("acks", "all");
props.put("retries", 3);

KafkaProducer<String, String> producer =
    new KafkaProducer<>(props);

ProducerRecord<String, String> record =
    new ProducerRecord<>("test-topic", "key1", "hello kafka");

producer.send(record);
producer.close();

六、常见配置说明

七、验证是否成功

消费者查看消息：

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

如果你需要：

• 远程生产者配置
• SSL / SASL 安全配置
• Docker 中的 Kafka 生产者
• Python（kafka-python）示例

可以直接告诉我你的使用场景。