kafka生产者debian如何配

下面给你一份在 Debian 上配置 Kafka 生产者的实用指南，覆盖系统环境、Kafka 安装、生产者配置、命令行测试和 Java 示例。

一、Debian 环境准备

1. 安装 Java（Kafka 依赖）

Kafka 需要 JVM（推荐 JDK 8 / 11 / 17）

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

验证：

java -version

二、安装 Kafka（Debian）

1. 下载 Kafka

cd /opt
sudo wget https://downloads.apache.org/kafka/3.7.1/kafka_2.13-3.7.1.tgz
sudo tar -xzf kafka_2.13-3.7.1.tgz
sudo mv kafka_2.13-3.7.1 kafka

2. 启动 Kafka（单机）

cd /opt/kafka

# 启动 Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &

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

三、创建 Topic（生产者前必备）

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

查看 Topic：

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

四、Kafka 生产者配置（核心）

1. 命令行生产者（最简单）

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

输入消息即可：

hello kafka
debian producer test

2. 生产者配置文件（推荐）

创建 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 \
  --bootstrap-server localhost:9092 \
  --topic test-topic \
  --producer.config producer.properties

五、Java 生产者示例（Debian 上开发）

1. Maven 依赖

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

2. Java 代码

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerDemo {
    public static void main(String[] args) {
        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);

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

        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Sent to partition " + metadata.partition());
            } else {
                exception.printStackTrace();
            }
        });

        producer.close();
    }
}

六、常见 Debian 上的问题

1. 无法连接 localhost:9092

检查：

netstat -tunlp | grep 9092

2. 云服务器注意

server.properties 中修改：

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

七、生产环境建议

✅ 使用 acks=all
✅ 开启 retries
✅ 使用批量发送
✅ 监控 producer 指标
✅ 多 broker + replication

如果你需要：

systemd 托管 Kafka
Python / Go 生产者
SSL / SASL 安全配置
多节点 Kafka 集群

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