消息队列 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 节点在接收到消息之前发生故障，消息将会丢失。

七、关键规则与配置建议

1. 副本数与 min.insync.replicas 的关联性

·强制约束：min.insync.replicas ≤ 副本数，否则 Broker 会拒绝写入请求。

·推荐配置：

·生产环境：副本数 = 3 & min.insync.replicas = 2（平衡容灾与性能）

·金融场景：副本数 = 3 & 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 搭配使用场景即数据可靠性

九、数据丢失的场景及解决方法

本节将分别通过生产端、服务端（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 设置数据可靠性的级别。

·acks 及相关配置参考副本数、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;

十五、数据丢失原因

·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 才能找回数据。

·消费速度和生产速度相差太久，而消息保存时间太短，导致消息还未及时消费就被过期删除。

20、解决方法

·合理配置参数 auto.commit.enable，等于 true 时表示自动提交。建议使用定时提交，避免频繁 commit offset。

·监控消费者的情况，正确调整数据的保留时间。监控当前消费 offset 以及未消费的消息条数，并配置告警，防止由于消费速度过慢导致消息过期删除。

22、在本地打印分区 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，则表示消息没有被服务端保存，客户端需要重试。