腾讯云消息队列 CKafka 是一种高吞吐量的分布式消息系统，在众多企业级应用中被广泛使用。数据高可靠性是 CKafka 的重要特性之一，本文档介绍 CKafka 数据高可靠的多方面机制。
多副本及选举机制
多副本
多副本设计可增强系统可用性、可靠性，CKafka 推荐消息主题 3 副本，生产使用至少 2 副本。
通常 Replica 会被均匀分布到整个集群 Broker 节点上，Replica 的分配算法如下：
1. 将所有 Broker（假设共 n 个 Broker）和待分配的 Partition 排序。
2. 将第 i 个 Partition 分配到第（i mod n）个 Broker 上。
3. 将第 i 个 Partition 的第 j 个 Replica 分配到第（(i + j) mod n）个 Broker 上。
Leader Election 选举机制
消息队列 CKafka 版在 ZooKeeper 中动态维护了一个 ISR（in-sync replicas），ISR 里的所有 Replica 都确保跟上了 Leader。只有在 ISR 里的成员才有被选为 Leader 的可能。
假设 ISR 中 f + 1 个 Replica，一个 Partition 能在保证不丢失已 commit 的消息的前提下容忍 f 个 Replica 的失败。
假设共有 2f + 1 个 Replica（包含 Leader 和 Follower），commit 之前必须保证有 f + 1 个 Replica 复制完消息，为了保证正确选出新的 Leader，fail 的 Replica 不能超过 f 个。
Leader 切换原理
在建立一个新 Topic 时，CKafka Broker 集群会进行每个 partition 的 leader 分配，将当前 Topic 的 partition 均匀分布在每个 Broker 上。
但在使用一段时间后，可能会出现 partition 在 Broker 上分配不均，或是出现客户端在生产消费中抛出 BrokerNotAvailableError，NotLeaderForPartitionError 等异常。
这通常都是由于 partition 发生了 leader 切换导致的，典型场景如下：
当某个 partition leader 所在的 Broker 发生某些意外情况，例如网络中断，程序崩溃，机器硬件故障导致无法与 Broker controller 通信时，当前 Topic partition 将会发生 leader 切换，leader 将迁移至 follower partition 上。
当 Kafka 集群设置 auto.leader.rebalance.enable = true进行自动 rebalance，或是人工增加/削减 Broker 并手动触发 rebalance 时，由于涉及到 partition 自动平衡，此时也会出现 leader 切换。
当由于 Broker 意外中断，导致 leader 切换时：
如果客户端设置 ack = all，并且 min.insync.replicas > 1 ，由于消息同时在 leader partition 和 follower partition 都确认，因此消息不会丢失。
如果客户端设置 ack = 1，此时可能会出现设置在 replica.lag.time.max.ms时间中的消息未同步到 follower partition，可能导致消息丢失。
当由于 Broker 正常，手动/自动(如实例升级、单可用区切换跨可用区、实例迁移等)发起 rebalance 导致 leader 切换时，不会导致消息丢失，原因如下：
如果客户端设置 ack = all，并且 min.insync.replicas > 1 ，由于消息同时在 leader partition 和 follower partition 都确认，因此消息不会丢失。
如果客户端设置 ack = 1 ，leader 切换将会自动同步 partition 中的 offset，因此消息不会丢失。
acks 参数配置
1. acks = all（或 -1）
当生产者将 acks 参数设置为 all（或 -1）时，表示生产者需要等待所有的 ISR 副本都确认接收到消息后，才会认为消息发送成功。
这种配置提供了最高的数据可靠性，但可能会增加消息发送的延迟。例如，在一个包含 3 个副本的分区中，生产者发送消息后，需要等待 Leader 副本和两个 Follower 副本都确认接收到消息，消息才会被认为发送成功。
2. acks = 1
当 acks 参数设置为 1 时，生产者只需要等待 Leader 副本确认接收到消息后，就会认为消息发送成功。Follower 副本的同步情况不影响生产者的判断。
这种配置在数据可靠性和消息发送延迟之间提供了一种平衡。如果 Leader 副本在消息确认后发生故障，而此时 Follower 副本还没有完全同步消息，可能会导致消息丢失。
3. acks = 0
当 acks 参数设置为 0 时，生产者不需要等待任何副本的确认，就可以认为消息发送成功。这种配置下消息发送的延迟最低，但数据可靠性也最低。如果Broker 节点在接收到消息之前发生故障，消息将会丢失。
副本数、acks、min.insync.replicas 参数的关系
关键规则与配置建议
1. 副本数与min.insync.replicas 的关联性
强制约束：min.insync.replicas ≤ 副本数，否则 Broker 会拒绝写入请求。
推荐配置：
生产环境：副本数 = 3 & min.insync.replicas = 2（平衡容灾与性能）
金融场景：副本数 = 5 & min.insync.replicas = 3（强一致性要求）
2. acks 与 min.insync.replicas 的联动规则
acks = all的生效条件：
必须配合 min.insync.replicas ≥ 2，否则当 ISR 中仅剩 Leader 时，acks = all 退化为 acks = 1，无法保证数据安全。
acks = 1 的潜在风险：
若 Leader 副本故障且 Follower 未同步，消息可能丢失（需通过 retries 和幂等性补偿）。
常见副本数、acks、min.insync.replicas 搭配使用场景即数据可靠性
|
任意值 | 0 | 任意值 | 生产者不等待确认 消息可能未写入任何副本即丢失 | 最低：无数据可靠性保障 |
2 | all | 1 | 退化为 acks = 1（ISR 可能仅含Leader） 容忍 1 个 Broker 故障 | 低：Leader 故障时数据可能丢失 |
3 | 1 | 1 | 生产者仅需 Leader 确认 ISR 可能仅含 Leader 容忍 2 个 Broker 故障 | 中等：Leader 故障且无 Follower 同步时可能丢失数据 |
3 | all | 2 | 生产者需等待 ISR 中至少 2 个副本确认 ISR 包含 Leader 及至少 1 个 Follower 容忍 1 个 Broker 故障 | 高：ISR 中至少 2 个副本存活时数据不丢失 |
3 | all | 3 | 生产者需等待所有 3 个副本确认 ISR 必须包含所有副本 容忍 0 个 Broker 故障 | 最高：极端情况下仅允许 0 个节点故障 |
5 | all | 3 | 生产者需等待 3 个副本确认 ISR 包含 Leader 及 2 个 Follower 容忍 2 个 Broker 故障 | 高：多数副本存活时数据安全 |
数据丢失的场景及解决方法
本节将分别通过生产端、服务端（CKafka）和消费端介绍影响消息队列 CKafka 数据可靠性的因素，并提供对应的解决方法。
生产端数据丢失如何处理？
数据丢失原因
生产者将数据发送到消息队列 CKafka 版时，数据可能因为网络抖动而丢失，此时消息队列 CKafka 版未收到该数据。可能情况：
网络负载高或者磁盘繁忙时，生产者又没有重试机制。
磁盘超过购买规格的限制，例如实例磁盘规格为 9000GB，在磁盘写满后未及时扩容，会导致数据无法写入到消息队列 CKafka 版。
突发或持续增长峰值流量超过购买规格的限制，例如实例峰值吞吐规格为 100MB/s，在长时间峰值吞吐超过限制后未及时扩容，会导致数据写入消息队列 CKafka 版变慢，生产者有排队超时机制时，导致数据无法写入到消息队列 CKafka 版。
解决方法
生产者对自己重要的数据，开启失败重试机制。
针对磁盘使用，在配置实例时设置好监控和 告警策略 ，可以做到事先预防。
遇到磁盘写满时，可以在控制台及时升配（消息队列 CKafka 版非独占实例间升配为平滑升配不停机且也可以单独升配磁盘）或者通过修改消息保留时间降低磁盘存储。 为了尽可能减少生产端消息丢失，您可以通过 buffer.memory 和 batch.size（以字节为单位）调优缓冲区的大小。缓冲区并非越大越好，如果由于某种原因生产者宕机了，那么缓冲区存在的数据越多，需要回收的垃圾越多，恢复就会越慢。应该时刻注意生产者的生产消息数情况、平均消息大小等（消息队列 CKafka 监控中有丰富的监控指标）。
配置生产端 acks。
当 producer 向 leader 发送数据时，可以通过 request.required.acks 参数以及 min.insync.replicas 设置数据可靠性的级别。
建议配置的参数值
此参数值仅供参考，实际数值需要依业务实际情况而定。
重试机制：message.send.max.retries=3;retry.backoff.ms=10000;
高可靠的保证：request.required.acks=-1;min.insync.replicas=2;
高性能的保证：request.required.acks=0;
可靠性+性能：request.required.acks=1;
服务端（CKafka）数据丢失如何处理？
数据丢失原因
partition 的 leader 在未完成副本数 followers 的备份时就宕机，即使选举出了新的 leader 但是数据因为未来得及备份就丢失。
开源 Kafka 的落盘机制为异步落盘，也就是数据是先存在 PageCache 中的，当还没有正式落盘时，Broker 出现断开连接或者重启或者故障时，PageCache 上的数据由于没有来得及落盘进而丢失。
磁盘故障导致已经落盘的数据丢失。
解决方法
开源 Kafka 是多副本的，官方推荐通过副本来保证数据的完整性，此时如果是多副本，同时出现多副本多 Broker 同时挂掉才会丢数据，比单副本数据的可靠性高很多，所以消息队列 CKafka 版强制 Topic 是双副本，可配置3副本。
消息队列 CKafka 版服务配置了更合理的参数 log.flush.interval.messages 和 log.flush.interval.ms，对数据进行刷盘。
消息队列 CKafka 版对磁盘做了特殊处理，保证部分磁盘损坏时也不会影响数据的可靠性。
建议配置的参数值
非同步状态的副本可以选举为 leader：unclean.leader.election.enable=false // 关闭
消费端数据丢失如何处理？
数据丢失原因
还未真正消费到数据就提交 commit 了 offset，若过程中消费者挂掉，但 offset 已经刷新，消费者错过了一条数据，需要消费分组重新设置 offset 才能找回数据。
消费速度和生产速度相差太久，而消息保存时间太短，导致消息还未及时消费就被过期删除。
解决方法
合理配置参数 auto.commit.enable，等于 true 时表示自动提交。建议使用定时提交，避免频繁 commit offset。
监控消费者的情况，正确调整数据的保留时间。监控当前消费 offset 以及未消费的消息条数，并配置告警，防止由于消费速度过慢导致消息过期删除。
数据丢失排查方案
在本地打印分区 partition 和偏移量 offset 进行排查
打印信息代码如下：
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(Topic, messageKey, messageStr));
RecordMetadata recordMetadata = future.get();
log.info("partition: {}", recordMetadata.partition());
log.info("offset: {}", recordMetadata.offset());
如果能够打印出 partition 和 offset，则表示当前发送的消息在服务端已经被正确保存。此时可以通过消息查询的工具去查询相关位点的消息即可。
如果打印不出 partition 和 offset，则表示消息没有被服务端保存，客户端需要重试。