Apache Kafka（五）- Safe Kafka Producer

本文主要是介绍Apache Kafka（五）- Safe Kafka Producer，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Kafka Safe Producer

在应用Kafka的场景中，需要考虑到在异常发生时（如网络异常），被发送的消息有可能会出现丢失、乱序、以及重复消息。

对于这些情况，我们可以创建一个“safe producer”，用于规避这些问题。下面我们会先介绍对于这几种情况的说明以及配置，最后给出一个配置示例。

1. acks 详述

之前我们介绍过 Kafka Producer 的 acks 有三种模式，下面我们进一步介绍一下这三种模式：

1.1. acks = 0（no acks）

使用acks=0 时，也就意味着：

在发送一条message 后，不需要response
如果broker 下线或是发生了故障，则我们不会知道，并且会丢失数据，因为broker不会返回任何response 给producers

acks=0 的工作方式如下图，不需要收到任何 ack：

一般使用 acks=0 的场景为：可以接受可能丢失数据。例如：

指标信息收集
日志收集（可接受偶尔丢失几条log数据）

1.2. acks=1（leader acks）

使用acks=1 时：

producer需要获取leader 的response，才能确认消息已被收到。但是replication是否收到则不会保证（会在后台执行replication）
如果producer 没有收到ack，则可能会retry

acks=1 的工作方式如下图，Producer需要收到每条消息的ack：

如果leader broker下线或是发生故障，但是replicas还没有复制发送的数据，则我们也会有数据丢失
默认是这个模式

1.3. acks=all（replicas acks）

使用acks=all 时：

需要Leader与Replicas的ack
增加了latency与更高的“数据不丢失”安全性
如果有足够的replicas，则不会有数据丢失

acks=all的一个工作方式如下，每个replica都需要回复ack，才能保证一个write的写入：

如果是需要完全不丢失数据，则这个设置是有必要考虑的。

在设置 acks=all（也就是replica acks）时，必须与另一个参数一起用，也就是：min.insync.replicas：

min.insync.replicas参数可以在broker level设置或是topic level设置（override）
min.insync.replicas=2表示的是：至少2个brokers是ISR（in-sync replicas）（包括leader），且必须响应表示它们有数据，否则就会返回报错。设置此参数为2也是最常见的配置。

假设设置参数replication.factor=3, min.insync=2, acks=all，则最多只能容许1一个broker异常，否则producer在发送数据时会收到报错。

假设有3个brokers，min.insync.replicas=2，若其中有两个broker异常，则Producer会收到“NOT_ENOUGH_REPLICAS”的异常。

2. Producer Retry

为了防止一些瞬时的错误（例如NotEnoughReplicasException）影响整个应用，一般我们需要处理一些异常，以避免数据丢失。在Producer中也有重试的配置，默认为0，可以手动调整它的值，最高可以到Integer.MAX_VALUE。

在重试时，默认情况下，会有可能造成消息发送时乱序。因为一般发送失败的消息会被re-queue，然后再次发送，所以会造成部分消息乱序。

此情况在key-based 序列消息中，尤为严重。因为所有具有相同key的messages会被送往同一个partition，而若是有消息被requeue，然后重传，则会打乱这个partition中的部分key顺序。

对于这种情况，我们可以设置参数max.in.flight.requests.per.connection控制：同一时刻，有多少个produce请求可以被并行发起：

默认为5
如果为了完全确保重试后的消息也能保持严格有序，则可以设置此参数为1（但是可能会影响throughput）

不过在 Kafka >= 1.0.0中，对于此场景会有更好的解决方案，本文之后的部分会提及。

3. Idempotent Producer

在重传场景中，会遇到一个常见的问题是：由于网络的原因，Producer会在Kafka中引入重复的messages。

如下图所示：

一个正常的request请求为：

Producer 发送消息到Kafka
Kafka commit 这条消息
Kafka发送ack回Producer

但是一个产生重复消息的请求过程为：

Producer发送消息到Kafka
Kafka commit 这条消息
Kafka发送ack回Producer时，由于网络原因，ack未到达Producer端
Producer过一段时间后开始重传消息
Kafka commit这条重复的消息并返回ack给Producer

从Producer的角度看，它仅正常发送了一次消息，因为它只收到了一次 ack。从Kafka的角度看，它收到了两次消息，所以commit了两次。

在Kafka >= 0.11之后，可以定义一个“idempotent producer”，可以解决由网络问题造成的重复消息。如下图所示：

对于一个idempotent producer 来说，处理重复消息的请求过程为：

Producer 发送一条消息
Kafka收到消息并commit
Kafka发送回Producer的ack由于网络问题未到达Producer
Producer重试发送消息，在Producer>=0.11 的版本中，消息里会带上一个produce request id
Kafka收到消息后，通过对比produce request id，可以辨别出这条消息是一条重复的消息，所以不会再次commit，并会再次发送一个ack

Idempotent producers可以很好的保证一个稳定，以及无重复数据的pipeline。

伴随Idempotent producers一起被设置的参数有：

retries = Integer.MAX_VALUE (2^31-1 = 2147483647)，也就是说基本会在出错时无限重传
max.in.flight.requests=1 (Kafka >= 0.11 & < 1.1) ，也就是说在这些版本中，若是设置max.in.flight.requests > 1时仍会有可能产生乱序数据
或者 max.in.flight.requests=5 (Kafka >= 1.1 –> High Performance)，也就是说在高于1.1版本的Kafka中，设置max.in.flight.requests=5也可以在保证不乱序的同时，保证并行的高性能

对于Idempotent Producer的配置，仅需配置类似以下参数即可：

properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

4. Safe Producer 配置总结

上面介绍了创建一个safe producer 所需的配置，下面我们总结一下在不同版本的Kafka中所需要做的配置：

Kafka < 0.11

ack=all (procuder level)：确保在发送ack前，数据已经正常备份
min.insync.replicas=2 (broker/topic level)：确保至少有两个in ISR 的brokers 有数据后再回送ack
retires=MAX_INT (producer level)：确保在发生瞬时问题时，可以无限次重试
max.in.flight.requests.per.connection=1 (producer level)：确保每次仅有一个请求发送，防止在重试时产生乱序数据

Kafka >= 0.11

enable.idempotence=true (producer level) + min.insync.replicas=2 (broker/topic level)
- 隐含的配置为 acks=all, retries=MAX_INT, max.in.flight.requests.per.connection=5 (default)
- 可以在保证消息顺序的同时，提高performance

这里必须要提到的是：运行一个“safe producer”可能会影响系统的throughput与latency，所以在应用到生产系统前，必须先做测试以判断影响。

5. safe producer 示例

我们按照之前的步骤启动一个由Java编写的Kafka Producer，并查看输出的配置，可以看到默认的部分参数为：

acks = 1

enable.idempotence = false

max.in.flight.requests.per.connection = 5

retries = 2147483647

现在我们显式地加上以下参数：

// create a safe Producer
properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
properties.setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
properties.setProperty(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));

然后查看producer配置的部分输出为：