topic 设置 replication.factor 参数值大于 1，要求每个 partition 必须有至少 2 个副本；

replication.factor 是 Kafka 中的一个关键配置，它决定了一个 topic 的分区需要被复制到多少个 broker 上。这个参数在创建 topic 时需要指定，并且在 topic 创建后也可以更改。

在 Kafka 配置中，replication.factor 参数通常在 server.properties 文件中设置，或者在创建 topic 时通过命令行参数指定。

1）在 server.properties 中设置默认的topic副本个数：

default.replication.factor=

这个配置项指定了默认的副本因子，如果创建 topic 时没有指定 replication.factor，则会使用这个默认值。

2）在创建 topic 时指定：

kafka-topics.sh --create --topic my_topic --partitions 3 --replication-factor 2 --zookeeper localhost:2181

在这个例子中，my_topic 被设置了 3 个分区和副本因子为 2。这意味着每个分区的数据会在至少两个 broker 上复制。

3）在 topic 创建后更改副本因子：

kafka-topics.sh --alter --topic my_topic --replication-factor 3 --zookeeper localhost:2181

这个命令会更改 my_topic 的副本因子为 3。

注意：replication.factor 的值不能大于集群中 broker 的总数。

acks是生产者客户端的一个配置选项，它表示生产者发送的消息要有多少个副本节点接收到，生产者才认为消息发送成功,其取值可以为0、1、-1(也可以表示为 all)。

1）acks=0

当 acks 设置为 0 时，生产者在发送消息后不需要等待来自kafka server的broker的确认信息，生产者在消息发送出去后就认为写入成功。这种设置具有最高的效率，但最容易导致消息丢失，因为生产者不等待任何确认。

2）acks=1

当 acks 设置为 1 时，只要leader副本已经收到消息并将其写入本地日志中，就会返回ack给生产者，这种方式的安全性较高，但仍然存在一定的消息丢失风险：考虑到这样一个场景，生产者发送的消息确实成功写入到了leader副本，leader副本就会返回成功响应给生产者，但是这条消息还没来得及同步到follower副本中，此时leader副本崩溃，那么这条消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。

相对其它两种设置，ack=1这种配置性能和安全是最均衡的。

3）acks=all

当 acks 设置为 -1 或者all时，leader等待所有同步副本都收到该消息后才会返回ack给生产者。保证了只要有⼀个同步副本存在，消息就不会丢失，这种方式最安全，但性能最差。

前面说的acks=all，leader等待所有同步副本都收到该消息后才会返回ack给生产者，但是这个同步副本的个数并不一定是你设置的topic的所有副本个数，准确的的说这里等待所有同步副本的个数，是指topic的ISR集合个数，注意这个ISR集合个数是会动态变化的，并且由于某种原因可能就会只剩下leader副本自己。

1）同步副本（In-sync Replicas）

在数据同步过程中，follower副本 相对于 leader副本 而言可能会有一定程度上的滞后。根据这个特性，Kafka 中的副本被分为两个集合：ISR（In-Sync Replicas） 和 OSR（Out-of-Sync Replicas）。

ISR集合：所有与 leader 副本保持一定程度同步的副本（包括 leader 副本在内）组成的集合。当Leader副本出现故障后，只有这个集合中的 副本才能被选举为leader（（不过可以修改参数配置来改变这个行为）。）。

OSR集合：与 leader 副本同步滞后过多的副本组成的集合。

leader副本负责维护和跟踪 ISR 集合中所有 follower 副本的同步状态，当某个 follower 副本的滞后程度过高时，leader副本 会将它从 ISR 集合中剔除。当然，如果 OSR 集合中有 follower 同步进度追上了 leader，那么 leader 也会把它从 OSR 集合中转移至 ISR 集合。

如果系统对数据安全性要求高，不允许丢数据，那么需要将acks配置为all。但是仅配置acks=all还是会存在问题。

如果一个分区有3个副本，正常情况ISR里面就是3个副本，但由于网络等问题ISR集合里面的副本数量可能会变化，当ISR集合的副本减少到只有 1 个时，也就是说leader是唯一同步副本，acks=all其实就退化成acks=1了（只需要leader副本保存好消息就返回ack），前面我们已经介绍过，acks = 1是会存在消息丢失的可能。

2）为了防止这种情况发生，可以使用 min.insync.replicas 配置。

min.insync.replicas是broker上的配置，它指定了当acks=all时，ISR 中必须存在的最少同步副本数量。如果 ISR 中的同步副本数量小于这个值，写入操作将会失败，对于不能忍受消息丢失的系统来说，写入失败总比写进去然后丢了强。

min.insync.replicas 是 Kafka 中的一个参数，用于定义请求成功的最小同步副本集合大小。这个参数是为了保证数据的复制安全性，确保在producer发送数据到topic后，至少有min.insync.replicas个副本接收到数据并确认成功写入日志。

如果设置的min.insync.replicas的值大于1，并且Kafka集群中的副本数量也大于或等于这个值，那么在producer发送消息时，至少需要有min.insync.replicas个副本同步成功，才会认为这条消息是成功发送的。

3）这个参数可以在broker级别设置，也可以在topic级别设置。

在broker级别设置，可以在server.properties文件中添加或修改：

min.insync.replicas=

在topic级别设置，可以使用kafka-configs.sh脚本：

bin/kafka-configs.sh --zookeeper  --entity-type topics --entity-name  --alter --add-config min.insync.replicas=

例如，如果你想设置min.insync.replicas为2，可以在server.properties中这样设置：

min.insync.replicas=2

或者使用kafka-configs.sh脚本：

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name my-topic --alter --add-config min.insync.replicas=2

请注意，设置min.insync.replicas的值大于1前，需要确保Kafka集群的副本数量大于或等于min.insync.replicas的值，否则Kafka集群将无法启动。

max.in.flight.requests.per.connection是Kafka生产者（Producer）配置中的一个关键参数，它指定了在等待来自服务器的响应之前，生产者可以发送到单个服务器的未确认请求的最大数量，

默认值：在Kafka的默认配置中，max.in.flight.requests.per.connection的值通常设置为5。这意味着生产者可以在收到服务器对前一条消息的确认之前，向同一个服务器发送多达5条消息。

作用范围：该参数作用于生产者与Kafka集群中单个Broker的连接。它控制了生产者在等待Broker响应之前，能够并行发送的未确认请求的数量。

增加max.in.flight.requests.per.connection的值可以提高生产者的吞吐量。因为允许生产者并行地发送更多的消息，而不是等待每条消息的确认。这在高负载或批量发送大量消息的场景下尤为有效。

消息顺序保证：当max.in.flight.requests.per.connection大于1时，虽然可以提高吞吐量，但可能会牺牲消息的顺序性保证。因为服务器可能以与发送顺序不同的顺序处理这些请求。这对于需要严格顺序保证的用例（如日志记录）可能不利。

故障恢复：在发生错误时（如网络问题或服务器故障），增加此值可能会导致生产者需要重试更多的消息。因为更多的消息可能处于“在途”状态，这增加了故障恢复的复杂性和时间成本。

严格顺序保证的场景：对于需要严格顺序保证的场景（如金融交易记录），应将max.in.flight.requests.per.connection设置为1。以确保消息严格按照发送顺序被处理。

高吞吐量场景：对于可以容忍一定顺序混乱且追求高吞吐量的场景（如大数据分析），可以适当增加max.in.flight.requests.per.connection的值。但需要注意监控和测试以确保系统稳定性和性能。同时，应启用幂等性（enable.idempotence=true）来避免消息重复。

幂等性启用条件：当启用幂等性时，需要确保max.in.flight.requests.per.connection小于或等于5。这是因为幂等性相关代码中有个ProducerStateEntry实体，用于存储每个producerId对应的Batch，并且只能固定缓存5个BatchMetadata。若max.in.flight.requests.per.connection设为大于5，可能会造成某一批次的元数据被挤出缓存，导致重试时找不到对应的BatchMetadata。

总结：max.in.flight.requests.per.connection是Kafka生产者调优的重要一环。合理配置该参数可以平衡吞吐量和消息顺序性保证之间的关系，满足不同应用场景的需求。设置该参数为1，可以保证最多会丢一个消息。

在 producer 端设置 retries=MAX（很大很大很大的一个值，无限次重试的意思），当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数，设置的很大，当遇到异常就会一直重试，一直到成功收到broker的 ACK，但是设置这个参数大于0可能会导致消息重复， 因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失)有可能导致broker接收到重复的消息。

如何缓解重复消息的问题？

1、Kafka在0.11.0.0之后引入了的幂等性的特性；

2、使用唯一消息标识符：在发送消息时，为每条消息设置一个唯一的标识符（如UUID），然后程序消费端实现幂等性消费，即无论消息被消费多少次，处理结果都应该是一致的。

Kafka在0.11.0.0之后通过如下配置使用幂：

1）enable.idempotence=true。#enable.idempotence配置项表示是否使用幂等性。当enable.idempotence配置为true时，acks必须配置为all。并且建议max.in.flight.requests.per.connection的值小于5。

2）acks=all。

但是幂等性的有一些限制条件：

只能保证 Producer 在单个会话内不丟不重 ，如果 Producer 出现意外挂掉再重启是无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不丢不重）;

幂等性不能跨多个 Topic-Partition，只能保证单个 partition 内的幂等性 ，当涉及多个 Topic-Partition 时，这中间的状态并没有同步。

如果需要跨会话、跨多个 topic-partition 的情况，需要使用 Kafka 的事务性来实现。这种幂等性只是保证了在生产端实现了幂等性，在实际场景中往往需要在消息者端实现幂等性，可以最大程度避免重复消费。

对于Producer，需要设置transactional.id属性。设置了transactional.id属性后，enable.idempotence属性会自动设置为true。

对于Consumer，需要设置isolation.level = read_committed，这样Consumer只会读取已经提交了事务的消息。另外，需要设置enable.auto.commit = false来关闭自动提交Offset功能；

Kafka无法保证绝对不丢数据。虽然Kafka被设计为具有高吞吐量和分布式环境中的数据流处理系统，并且通过多种机制来确保消息的可靠性，但极端情况下，它仍然可能丢失数据。据我目前了解，目前分布式一致性协议 Raft和Paxos，是经过数学理论推演的一致性协议，例如Ocenbase和tidb等分布式数据库。