kafka-clients之ProducerConfig
在Kafka 2.4.0版本中，ProducerConfig类提供了许多配置参数，用于控制生产者行为。这些参数可以调整Kafka生产者的性能、可靠性和行为。以下是每一个参数的详细说明：
基本配置
- bootstrap.servers
- 描述：Kafka集群的地址列表，用于初始化与Kafka集群的连接。
- 默认值：无
- 示例：
localhost:9092,localhost:9093
- client.id
- 描述：标识生产者的ID，便于在服务端记录日志或追踪。
- 默认值：空
- 示例：
my-producer-1
- key.serializer
- 描述：将消息的键序列化为字节的类，必须是
org.apache.kafka.common.serialization.Serializer的实现。 - 默认值：无
- 示例：
org.apache.kafka.common.serialization.StringSerializer
- 描述：将消息的键序列化为字节的类，必须是
- value.serializer
- 描述：将消息的值序列化为字节的类，必须是
org.apache.kafka.common.serialization.Serializer的实现。 - 默认值：无
- 示例：
org.apache.kafka.common.serialization.StringSerializer
- 描述：将消息的值序列化为字节的类，必须是
分区与负载均衡
- partitioner.class
- 描述：定义消息发送到Kafka主题时分区的策略，默认使用内置的分区器，也可以自定义。
- 默认值：
org.apache.kafka.clients.producer.internals.DefaultPartitioner - 示例：自定义类如
com.example.MyPartitioner
- acks
- 描述：控制生产者发送消息的确认级别。
0：生产者不等待服务器的响应，消息有可能丢失。1：Leader副本确认消息后返回ack。all：所有副本都确认消息后返回ack，提供最高的持久性保证。
- 默认值：
1 - 示例：
all
- 描述：控制生产者发送消息的确认级别。
- retries
- 描述：发送消息失败时生产者的重试次数。
- 默认值：
2147483647(无限重试) - 示例：
5
- retry.backoff.ms
- 描述：两次重试之间的时间间隔（毫秒）。
- 默认值：
100 - 示例：
200
性能调优
- batch.size
- 描述：生产者发送给同一分区的消息批次大小，以字节为单位。达到批次大小后会立即发送消息，即使
linger.ms没有达到。 - 默认值：
16384（16 KB） - 示例：
32768（32 KB）
- 描述：生产者发送给同一分区的消息批次大小，以字节为单位。达到批次大小后会立即发送消息，即使
- linger.ms
- 描述：生产者在发送消息之前等待的时间，以增加批量大小。即使没有达到
batch.size，生产者也会在达到linger.ms后发送消息。 - 默认值：
0 - 示例：
5
- 描述：生产者在发送消息之前等待的时间，以增加批量大小。即使没有达到
- buffer.memory
- 描述：生产者用于缓冲消息的内存总量（字节）。
- 默认值：
33554432（32 MB） - 示例：
67108864（64 MB）
- compression.type
- 描述：消息的压缩类型。
none：不压缩。gzip：使用Gzip压缩。snappy：使用Snappy压缩。lz4：使用LZ4压缩。zstd：Kafka 2.1.0 引入的Zstandard压缩。
- 默认值：
none - 示例：
gzip
- 描述：消息的压缩类型。
- max.in.flight.requests.per.connection
- 描述：在单个连接上生产者可以发送但尚未确认的最大请求数。较低的值可以减少重试时消息的顺序错误。
- 默认值：
5 - 示例：
1
可靠性与错误处理
- enable.idempotence
- 描述：是否启用幂等性。启用后，Kafka保证每条消息只会写入一次，从而避免重复消息。
- 默认值：
false - 示例：
true
- transactional.id
- 描述：如果需要使用事务，则必须设置该参数，用于唯一标识一个事务性生产者。
- 默认值：空
- 示例：
my-transactional-producer
- delivery.timeout.ms
- 描述：消息发送超时时间，包含重试时间。超时后消息将被丢弃。
- 默认值：
120000（2分钟） - 示例：
300000（5分钟） - 注意该配置必须符合下面的要求，否责会抛出配置异常：delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
Caused by: org.apache.kafka.common.config.ConfigException: delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
at org.apache.kafka.clients.producer.KafkaProducer.configureDeliveryTimeout(KafkaProducer.java:501)
at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:392)
... 3 more
- request.timeout.ms
- 描述：生产者等待服务器响应的最大时间，超时后重试。
- 默认值：
30000（30秒） - 示例：
60000（1分钟）
- max.block.ms
- 描述：在缓冲区已满的情况下，生产者阻塞的最大时间。超过该时间将抛出异常。
- 默认值：
60000（1分钟） - 示例：
300000（5分钟）
其他配置
- max.request.size
- 描述：生产者发送的单个请求的最大大小，单位为字节。
- 默认值：
1048576（1 MB） - 示例：
2097152（2 MB）
- connections.max.idle.ms
- 描述：生产者的空闲连接关闭时间。
- 默认值：
540000（9分钟） - 示例：
300000（5分钟）
- send.buffer.bytes
- 描述：TCP发送缓冲区大小，单位为字节。如果设置为
-1，则使用操作系统默认设置。 - 默认值：
131072（128 KB） - 示例：
65536（64 KB）
- 描述：TCP发送缓冲区大小，单位为字节。如果设置为
- receive.buffer.bytes
- 描述：TCP接收缓冲区大小，单位为字节。如果设置为
-1，则使用操作系统默认设置。 - 默认值：
32768（32 KB） - 示例：
65536（64 KB）
- 描述：TCP接收缓冲区大小，单位为字节。如果设置为
安全配置
- security.protocol
- 描述：Kafka客户端与Broker之间通信的协议。
PLAINTEXT：不加密。SSL：SSL加密。SASL_PLAINTEXT：使用SASL认证但不加密。SASL_SSL：使用SASL认证并加密。
- 默认值：
PLAINTEXT - 示例：
SSL
- 描述：Kafka客户端与Broker之间通信的协议。
- ssl.keystore.location
- 描述：SSL密钥库的文件路径，用于客户端认证。
- 默认值：空
- 示例：
/path/to/keystore.jks
- ssl.truststore.location
- 描述：SSL信任库的文件路径，用于验证Broker的SSL证书。
- 默认值：空
- 示例：
/path/to/truststore.jks
- sasl.mechanism
- 描述：SASL身份验证机制。
GSSAPI：Kerberos。PLAIN：用户名/密码认证。SCRAM-SHA-256或SCRAM-SHA-512：更安全的用户名/密码认证。
- 默认值：
GSSAPI - 示例：
PLAIN
- 描述：SASL身份验证机制。
这些参数能够帮助配置Kafka生产者的各个方面，从基本的连接配置到性能优化、安全设置等，确保生产者能够根据应用的需求高效、安全地发送消息。
消息丢弃场景
极端情况下，59秒时消息阻塞获取到资源了，进入缓冲区。但是一直没能发送成功，假设重试次数为无限大。那么这条消息会在2分钟后丢弃
即默认配置下：大约 3 分钟后会导致消息丢失。下面是详细的原因分析。
详细分析
**max.block.ms = 59 秒**- 假设在第 59 秒时，
send()操作成功地获取到缓冲区资源，并将消息放入缓冲区，从而避免max.block.ms的超时触发。 - 由于已经成功进入缓冲区，消息的后续发送过程不再受
max.block.ms影响，而是转向delivery.timeout.ms的控制。
- 假设在第 59 秒时，
**delivery.timeout.ms = 2 分钟**- 在消息进入缓冲区后，生产者会开始尝试将其发送到 Kafka 集群。
- 假设网络异常或集群不可用，生产者会反复重试发送该消息。即使重试次数无限大，
delivery.timeout.ms限制了整个发送过程的最大时间。 - 因此，在消息进入缓冲区的 2 分钟后，即大约第 3 分钟（59 秒 + 2 分钟）时，
delivery.timeout.ms超时将会触发，导致消息被丢弃，并向回调抛出TimeoutException或相关异常。
指数回退配置案例
相关配置
**reconnect.backoff.ms**：初始等待时间，默认值为50毫秒。**reconnect.backoff.max.ms**：等待时间的最大值，默认值为1000毫秒（1秒）。
指数退避公式
Kafka客户端使用指数退避策略，但不会将延迟时间翻倍。相反，它会增加一个固定值的时间，使其随着重试次数线性上升。可以使用公式：
wait_time=min(reconnect.backoff.ms×(attempt+1),reconnect.backoff.max.ms)
其中：
attempt是当前的重试次数，从0开始。
逐步退避的计算过程
以下是每次重试时的等待时间，直到达到最大值1000ms为止。
- 第1次重试（attempt = 0）：50 ms × (0 + 1) = 50 ms
- 第2次重试（attempt = 1）：50 ms × (1 + 1) = 100 ms
- 第3次重试（attempt = 2）：50 ms × (2 + 1) = 150 ms
- 第4次重试（attempt = 3）：50 ms × (3 + 1) = 200 ms
- 第5次重试（attempt = 4）：50 ms × (4 + 1) = 250 ms
- 第6次重试（attempt = 5）：50 ms × (5 + 1) = 300 ms
- …
- 第20次重试（attempt = 19）：50 ms × (19 + 1) = 1000 ms（已达最大值）
从第20次重试开始，等待时间将保持在1000毫秒，不再增长。
说明
这种递增方式可有效避免高频率重试引起的资源消耗，同时通过指数方式尽量缩短恢复时间。在高并发、网络不稳定或Kafka服务偶尔中断的情况下，合理配置这两个参数可以显著改善系统的恢复性。
原文地址：https://blog.csdn.net/u010597819/article/details/143640025
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！