Kafka客户端参数配置建议

acks

1

高可靠：all或者-1

高吞吐：1

收到Server端确认信号个数，表示producer需要收到多少个这样的确认信号，算消息发送成功。acks参数代表了数据备份的可用性。常用选项：

acks=0：表示producer不需要等待任何确认收到的信息，副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1。

acks=1：这意味着至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。如果follower没有成功备份数据，而此时leader又无法提供服务，则消息会丢失。

acks=all或者-1：这意味着leader需要等待ISR中所有备份都成功写入日志。只要任何一个备份存活，数据都不会丢失。min.insync.replicas指定必须确认写入才能被认为成功的副本的最小数量。

retries

0

结合实际业务调整

客户端发送消息的重试次数。值大于0时，这些数据发送失败后，客户端会重新发送。

注意，这些重试与客户端接收到发送错误时的重试没有什么不同。允许重试将潜在的改变数据的顺序，如果这两个消息记录都是发送到同一个partition，则第一个消息失败第二个发送成功，则第二条消息会比第一条消息出现要早。

针对网络闪断场景，生产者建议配置重试能力，推荐重试次数retries=3，重试间隔retry.backoff.ms=1000。

request.timeout.ms

30000

结合实际业务调整

设置一个请求最大等待时间（单位为ms），超过这个时间则会抛Timeout异常。

超时时间如果设置大一些，如127000（127秒），高并发的场景中，能减少发送失败的情况。

block.on.buffer.full

TRUE

TRUE

TRUE表示当我们内存用尽时，停止接收新消息记录或者抛出错误。

默认情况下，这个设置为TRUE。然而某些阻塞可能不值得期待，因此立即抛出错误更好。如果设置为false，则producer抛出一个异常错误：BufferExhaustedException

batch.size

16384

262144

默认的批量处理消息字节数上限。producer将试图批处理消息记录，以减少请求次数。这将改善client与server之间的性能。不会试图处理大于这个字节数的消息字节数。

发送到brokers的请求将包含多个批量处理，其中会包含对每个partition的一个请求。

较小的批量处理数值比较少用，并且可能降低吞吐量（0则会仅用批量处理）。较大的批量处理数值将会浪费更多内存空间，这样就需要分配特定批量处理数值的内存大小。

buffer.memory

33554432

67108864

producer可以用来缓存数据的内存大小。如果数据产生速度大于向broker发送的速度，producer会阻塞或者抛出异常，以“block.on.buffer.full”来表明。

这项设置将和producer能够使用的总内存相关，但并不是一个硬性的限制，因为不是producer使用的所有内存都是用于缓存。一些额外的内存会用于压缩（如果引入压缩机制），同样还有一些用于维护请求。

enable.idempotence

• 3.0之前版本：false
• 3.0及之后版本：true

如果无幂等需求，建议设置为false。

在生产者客户端中开启幂等，然后生产消息，此时您会在消费者客户端或Kafka控制台的“消息查询”中观察到消息offset不连续的现象。这是因为开启了幂等后，在生产消息时会产生一些元数据控制消息，这些控制消息也会生产到该Topic中， 且它们对消费者不可见，从而造成offset不连续的现象。

auto.commit.enable

TRUE

FALSE

如果为真，consumer所fetch的消息的offset将会自动的同步到zookeeper。这项提交的offset将在进程无法提供服务时，由新的consumer使用。

约束： 设置为false后，需要先成功消费再提交，这样可以避免消息丢失。

auto.offset.reset

latest

earliest

没有初始化offset或者offset被删除时，可以设置以下值：

• earliest：自动复位offset为最早
• latest：自动复位offset为最新
• none：如果没有发现offset，则向消费者抛出异常
• anything else：向消费者抛出异常。

如果将此配置设置为latest，新增分区时，生产者可能会在消费者重置初始偏移量之前开始向新增加的分区发送消息，从而导致部分消息丢失。

connections.max.idle.ms

600000

30000

空连接的超时时间（单位为ms），设置为30000可以在网络异常场景下减少请求卡顿的时间。

max.poll.records

500

需小于max.poll.interval.ms时间内的消费处理能力。

消费者单次从Broker中拉取的最大消息条数。

max.poll.interval.ms

300000

如果在两次poll之间存在复杂、耗时的逻辑，需要延长该参数值。

消费者两次拉取消息的最大时间间隔，单位为ms。如果两次拉取消息的时间间隔超过该间隔时间，就视为此次消费失败，消费者被踢出消费组，触发Rebalance。

heartbeat.interval.ms

3000

>=3000

消费者和Kafka之间的心跳间隔，单位为ms。

session.timeout.ms

10000

将该参数值设置为heartbeat.interval.ms的3倍以上。

使用消费组管理offset时，消费者与Broker的会话超时时间，单位为ms。

fetch.max.bytes

1000000

max.request.size < message.max.bytes < fetch.max.bytes

消费者单次从Broker中拉取消息的最大字节数。