kafak producer 配置介绍

梦里梦外; 2021-06-22 15:37 375阅读 0赞

producer的配置中，bootstrap.servers、key.serializer、value.serializer 这三个是必须的。接下来我们介绍一些重要的配置，更详细的配置，请参考官网：

[https://kafka.apache.org/documentation/\#configuration][https_kafka.apache.org_documentation_configuration]

**1、buffer.memory、max.block.ms、batch.size、linger.ms配置：**

新版kafka Producer使用的是异步、批量方式发送消息，即调用send方法后实际上是将该消息放入了内存缓冲区，然后再有一个后台线程不断地从缓冲区取消息批量封装成request发送到broker，发送成功后会回调send方法的callback（如果没有，就不用回调了）。

kafka客户端会累积一定量的消息后统一组装成一个批量消息发出，这个的触发条件是： 消息量达到了batch.size的大小或者等待批量的时间超过了linger.ms时间。

1）buffer.memory：

设置缓冲区大小（单位字节），默认值：33554432（32M）。

如果消息的发送速度超过了将记录发送到服务器的速度，则生产者将阻塞max.block.ms，超过此时间将引发异常。

2）max.block.ms：

设置最大阻塞时间，单位ms，默认值：60000（1分钟）。

send方法、partitionsFor这些方法可以由于缓冲区已满或元数据不可用而被阻塞。

3）batch.size：

为了提高发送性能，减少请求次数，producer会把要发送到同一个partition的批量消息作为batch发送，batch.size 是用来设置batch的字节大小，达到该值后会批量发送一个request，默认值：16384（16K）。

如果 batch.size 太小则可能会降低吞吐量，设置太大则可能会导致消息无法及时发送。若将该值设为0，则不会进行批处理。

4）max.request.size

设置客户端向broker发送单个请求的最大值（单位字节），即：消息批大小，默认值：1048576（1M）。避免出现超大的请求，造成网络不可用。

注：服务端对请求（记录批大小）有自己的上限（如果启用了压缩，则在压缩后）可能与此不同。

5）linger.ms：

设置将消息批量封装成request发送到broker的延时，超过该之后，会批量发送一个request，默认值：0。类似于TCP中的Nagle算法。

当分区的记录达到batch.size值，无论此设置如何，都会立即发送该记录；假设分区积累的字节数少于batch.size，我们将等待linger.ms时间，以等待更多的消息一起批量发送。

注：默认情况下，即使缓冲区中还有其他未使用的空间，缓冲区中的消息也可以立即发送。

**2、acks配置：**

producers一般采用异步、批量的方式向kafka发送消息，通常在发送完消息后会得到一个响应：

*  metadata数据，其中会包含offset值等信息；
 *  或者发送过程中遇到的错误(exception)；

1）acks=0：

消息从内存缓冲区被封装成请求（批量），发送到broker后不会等待broker的响应。所以，producer无法知道消息是否发生成功，有数据丢失的可能性，但此时系统的吞吐量最大。此外，acks=0时无法使用producer的retry功能。

此时客户端不等待broker的响应，使用同步模式或callback模式发送消息，在返回的metadata数据中offset都是-1，且不会收到发送中异常信息。

2）acks=1：

当leader partition收到请求并且存到本地后，broke发送一个成功的响应给producer；如果leader partition掉线或请求失败，producer会进行重试。

这种方式有较好的可靠性（partition leader成功存储到本地且响应producer，但在同步到follower partition前宕机，会造成消息丢失），同时有较好的性能（因为客户端producer会等待直到broker确认收到消息）。

此时客户端等待broker的响应，使用同步模式或callback模式发送消息，在返回的metadata数据中offset是正确的值，如果有发送中异常，也会捕获到。

3）acks=-1或all：

producer发送请求后，leader broker不仅会将消息写入本地日志，同时还会等待ISR中所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给producer。这种方式有最高的可靠性，和较差的性能。

此时客户端等待broker的响应，使用同步模式或callback模式发送消息，在返回的metadata数据中offset是正确的值，如果有发送中异常，也会捕获到。

**3、retries配置：**

producer在收到错误的响应后进行重试的次数，这个动作是kafka client自动触发的（对于那些能够重试的异常），默认值：0。

此外，启用重试后，会带来producer 消息重复的可能性。解决这个问题，可以使用0.11版本服务端的幂等性。

1）retry.backoff.ms配置：

在重试之前需要等待的时间，默认是100。为了避免在某些故障情况下，短时间内大量的重试。

2）max.in.flight.requests.per.connection：

客户端在阻塞之前将通过单个连接发送的未确认请求的最大数量。如果将此设置设置为大于1且发送失败，则由于重试（例如，如果启用重试），存在消息重新排序的风险。默认值：5。

该值设置越大，会消耗一些内存，同时吞吐量会更高；设置越小，会节省内存，吞吐量会变低。

我们看一个场景：如果使用异步+回调方式发送消息，我们不需要等待获取response，会最终在回调函数中获取response。这时，你能否知道可以发送多少条这样的消息吗（不需要等待response）？这个问题等价于，允许多少个正在进行中的请求仍未被确认？这个问题本质就是max.in.flight.requests.per.connection参数的含义。

所以，如果你关心消息的顺序，同时要求可靠性（开启重试），那么有两种选择：

*  使用同步的方式发送消息；
 *  max.in.flight.requests.per.connection = 1；

3）哪些情况会触发kafka client自动retry：

Kafka Producer 一般会发生两类错误：

*  可重试错误：这类错误可以通过重发消息来解决，比如对于连接错误，可以通过再次建立连接来解决；“无主（no leader）”错误则可以通过重新为分区选举首领来解决。KafkaProducer 可以被配置成自动重试。
 *  无法通过重试解决的错误：比如“消息太大”异常，直接抛出异常；重试后仍无法解决问题，应用程序会收到一个重试异常。

【实验】

[https_kafka.apache.org_documentation_configuration]: https://kafka.apache.org/documentation/#configuration