Debian Kafka 生产者配置技巧是什么

5月 7th, 2025

在Debian上配置Kafka生产者时，需要关注以下几个要点：

安装和配置Kafka

安装Java运行环境(JDK)：因为Kafka是用Scala编写的，需要JVM。
```
sudo apt-get install openjdk-8-jdk
```

下载并解压Kafka安装包：

wget https://downloads.apache.org/kafka/2.5.2/kafka_2.12-2.5.2.tgz
tar -zxvf kafka_2.12-2.5.2.tgz

配置环境变量，指定Kafka的安装目录和命令文件所在目录。

Kafka生产者基本配置

bootstrap.servers：设置连接Kafka的初始连接服务器地址，如果是集群，则可以通过此初始连接发现集群中的其他broker。
```
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
```
acks：控制消息的持久性和可靠性。常用配置包括acks1（仅等待主分区确认）和acks_all（等待所有ISR副本分区确认）。
```
props.put("acks", "all");
```

key.serializer 和 value.serializer：设置消息key和value的序列化器。

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

retries：设置消息发送失败时的重试次数。
```
props.put("retries", 3);
```
其他重要配置：
- batch.size：设置批量发送消息的大小。
- linger.ms：设置消息发送的延迟时间，以毫秒为单位。
- compression.type：设置消息压缩的类型，如gzip、snappy等。
- max.in.flight.requests.per.connection：设置每个连接允许的最大未确认请求数，用于控制消息的传输延迟和吞吐量。

配置示例

以下是一个Kafka生产者配置的示例，使用Java编写：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.send(new ProducerRecord<>("my-topic", "key", "value"));
        }
    }
}