网站建设交印花税,昆明模板建站代理,阿里巴巴官网app,做下载网站用阿里云的什么产品系列文章目录
上手第一关#xff0c;手把手教你安装kafka与可视化工具kafka-eagle Kafka是什么#xff0c;以及如何使用SpringBoot对接Kafka 架构必备能力——kafka的选型对比及应用场景 Kafka存取原理与实现分析#xff0c;打破面试难关 防止消息丢失与消息重复——Kafka可…系列文章目录
上手第一关手把手教你安装kafka与可视化工具kafka-eagle Kafka是什么以及如何使用SpringBoot对接Kafka 架构必备能力——kafka的选型对比及应用场景 Kafka存取原理与实现分析打破面试难关 防止消息丢失与消息重复——Kafka可靠性分析及优化实践 系列文章目录一、可靠性的考量角度二、分区副本1. 分区副本的含义2. AR 与 ISR机制 三、ACKS设置四、重试机制五、幂等性设计六、消费偏移量七、可靠性不足分析总结 在上一章内容中我们解析了Kafka在读写层面上的原理介绍了很多Kafka在读出与写入时的各种设计初步理解了Kafka大吞吐量的原因本期我们将带领大家从另一个角度即从可靠性方面来分析Kafka的机制与原理 作者简介战斧从事金融IT行业有着多年一线开发、架构经验爱好广泛乐于分享致力于创作更多高质量内容 本文收录于 kafka 专栏有需要者可直接订阅专栏实时获取更新 高质量专栏 云原生、RabbitMQ、Spring全家桶 等仍在更新欢迎指导 Zookeeper Redis dubbo docker netty等诸多框架以及架构与分布式专题即将上线敬请期待 一、可靠性的考量角度
其实我们在《RabbitMQ 能保证消息可靠性吗》 一文中我们已经阐述了对于MQ类主键可靠性应该从哪些角度去判断。总结下来其实就是
消息不会意外丢失消息不会重复传递
那么本次我们也将从这两方面来看看 Kafka 都做了哪些工作来提升可靠性。
二、分区副本
1. 分区副本的含义 我们之前了解到一个Kafka的主题被分为了若干个分区每个分区都是一个有序的消息队列。如上图我们就把topicA分成了1\2\3\4 四个分区实线圆柱但是我们还看到了更多的“虚线圆柱”这些其实是 1\2\3\4 分区的副本在Kafka中每个分区都有多个副本。副本是指一个分区在其他Broker上的备份。副本的作用是提高消息的可靠性和容错性一旦某个Broker宕机其他Broker上的副本就可以接替宕机的Broker继续提供服务。
我们可以看到不同的Broker之间。互相存储着对方的副本分区比如Partition1 存在Broker1 上但Partition1 的副本可以放在Broker2 同理 Partition2 的副本也可以放在Broker1 如果我们专注于一个分区那么其情况如下 为了区分我们一般把“实线圆柱”的分区称为 Leader“虚线圆柱”的分区称为Follower。Leader副本负责读写请求而Follower副本则只负责复制Leader副本的数据。
2. AR 与 ISR机制
上面我们讲了副本分区的Follower所有副本的合集统称为AR(Assigned Replicas)但是不同副本和Leader到底一致性有多高呢会不会出现有的副本同步得及时有的副本因为网络原因同步得很慢呢 这里又引出了一个重要的概念叫做ISRIn-Sync Replica机制。
ISR是一个机制也代表着一个同步合集顾名思义它包含着所有处于同步状态的副本。当一个副本和Leader副本的差距超过一定程度时这个副本就会被认为是不同步的不再被加入到ISR中。也因此Kafka中的 ISR 并不是一直不变的 那么既然ISR是动态的那哪些副本会被包含在ISR中呢 其实其主要依据就是 副本需要保证能够及时地接收并复制Leader副本的消息也就是需要保证与leader副本的消息同步延迟在一定的时间范围内默认情况下是10秒钟由参数 replica.lag.time.max.ms 控制。 换而言之因为分区与ISR机制我们的消息一旦被Kafka 接收后就会复制多份并很快落盘。这意味着即使某一台Broker节点宕机乃至硬盘损毁也不会导致数据丢失。
三、ACKS设置
如果说备份机制是保障消息不会在Kafka服务器丢失那么消息丢失的另一个重要原因就是消息在发送中丢失。
这种场景下我们就需要利用消息确认机制了此时我们也会利用到ISR比如我们在发送消息时可以通过设置ACK的值来决定同步的情况 acks0如果设置为零则生产者根本不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并被视为已发送。在这种情况下不能保证服务器已经收到记录重试配置也不会生效因为客户端通常不会知道任何故障。为每条记录返回的偏移量将始终设置为-1。 acks1这意味着领导者会将记录写入其本地日志但不会等待所有追随者的完全确认。在这种情况下如果领导者在确认记录后立即失败但在追随者复制之前记录将丢失。 acksall这意味着leader将等待所有ISR来确认记录。这保证了只要至少有一个副本处于ISR内记录就不会丢失。这是最有力的保证。这相当于acks-1的设置。
当我们设置acksall 或者 -1 的时候 broker 的 min.insync.replicas 参数起作用一个典型的场景是topic 有 3个 副本客户端设置 acks -1服务端设置 topic level 的 min.insync.replicas 2这样至少有 2 个副本写入后broker 才会返回但是如果 topic 只有 1 个副本而 acks allmin.insync.replicas 2就会报 NOT_ENOUGH_REPLICAS 错误 四、重试机制
发消息不可能万无一失当Kafka在发送或接收消息时发生错误时可以通过重试来解决这些问题提高系统的可靠性。这点不用多说那么在生产端我们可以配置哪些重试的参数呢 retries重试次数默认为0表示不启用重试机制但建议开启。 retry.backoff.ms每次重试的时间间隔默认为100ms。 retry.buffer.records每个分区的缓冲区中可以存储的最大重试消息数。 在实际应用中一些小故障是可以通过重试来解决的但是重试次数过多也会增加网络通信的负担甚至会导致消息堵塞。所以建议将其设置为1或2这样可以在第一次发送失败后进行重试从而提高消息的可靠性。但是如果网络状况很差或者需要处理重要的消息可以适当增加retries的值。
五、幂等性设计
如果你仔细观察上面的ACKS的设置相信你会发现这并不完美如果你将ACKs设置为-1all可以保证Producer到Server之间不会丢失数据即At Least Once(最少一次)但不能保证数据不重复。而如果你将ACKs级别设置为0不需要等写log则可以保证生产者每条消息只会被发送一次即At Most Once最多一次但不能保证不会发生数据丢失。
难道就没有一种既不会丢失也不会重复的方案吗其实是有的这个时候我们可以使用 ‘ack all’ 幂等性来解决而开启幂等性 即设置 enable.idempotence true 请注意启用幂等性要求 ①max.in.flight.requests.per.connection小于或等于5为任何允许的值保留消息顺序 ②retries重试次数大于0 ③acks必须为“all”。 PS我发现不少文章把开启幂等性参数写成 enable.idompotence不知是笔误还是什么原因 开启幂等性的意义在于生产者将确保在流中每条消息正好只会发送一次那么它是怎么实现的呢Kafka生产者的幂等性算法主要包括以下三个方面的实现
生产者编号 Producer在初始化的时候只有初始化的时候会随机生成PID会被分配一个PIDproducerid分区编号不同的分区有自己的paritionid即分区号序列号发往同一Partition的消息会附带Sequence Number即发送数据的编号代表着向分区发送的第几条消息这样PID, PartitionID, SeqNumber就相当于构成了一个主键。Broker端会对PID, PartitionID, SeqNumber做缓存当具有相同主键的消息提交时Broker只会持久化一条。
六、消费偏移量
前面说了很多都是生产者与Kafka的其实对于消费者Kakfa也有同步提交偏移量的设计。在 Kafka 的消费者 API 中同步提交偏移量可以通过设置 enable.auto.commit 参数为 false 然后在消费者应用中手动控制提交偏移量的时机来实现。 具体来说可以使用 commitSync() 方法提交偏移量该方法会一直阻塞直到提交偏移量成功或抛出异常。示例代码如下
Properties props new Properties();
// 设置 Kafka 集群地址等配置信息
props.put(bootstrap.servers, localhost:9092);
props.put(group.id, test-group);
props.put(enable.auto.commit, false); // 禁止自动提交偏移量
props.put(auto.offset.reset, earliest);
props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer);
props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumer(props);
consumer.subscribe(Arrays.asList(test-topic));try {while (true) {ConsumerRecordsString, String records consumer.poll(Duration.ofMillis(100));for (ConsumerRecordString, String record : records) {// 处理消息}consumer.commitSync(); // 手动提交偏移量}
} finally {consumer.close();
}
通过这种方式可以确保消息在被消费者处理后再提交偏移量从而避免了消息丢失或者重复处理的问题。同时手动提交偏移量也提高了消费者的可控性可以根据自身情况自由地设置提交偏移量的时机。
七、可靠性不足分析
尽管我们在上面讲了一些Kafka为了实现可靠性而做的设计一般情况下这种程度的可靠性足以应付了。但在实际应用过程中Kafka仍然可能会面临以下几个可靠性问题
生产者重复发送尽管开启了幂等性但不要忘记幂等性设置仅表示生产者对同一个分区的消息的写入是有序的、幂等的如果producer挂了重启之后producer会重新生成producerid此时幂等性校验就不准了。消费者重复消费如果消费者在提交偏移量前宕机了将导致Kafka认为该消息没有被消费在消费者重启后又会消费该消息导致重复消费。
这些情况Kafka自身已经无法解决。我们的解决策略只能契合在我们的业务处理上目前一个通用的方案是全局性ID生产/消费两端校验
我们可以先根据业务对消息进行去重然后使用诸如雪花算法等方案为每一条去重后的消息生产全局性的唯一ID并在发送和消费之前在redis或其他较快的存储件中进行标记这样当发生重复发送/消费时就能及时发现了。此时你可以选择放弃本次发送/消费也可以将该异常情况上报由人工来进行检查与处理
总结
本次我们对Kafka的可靠性进行了分析和优化实践。一般来说我们可以通过设置acks、开启幂等性消费端手动提交偏移量等方式来保证可靠性也足以应付大部分场景。而且实际应用过程中还可以配合全局ID等手段完善可靠场景。当然架构服务于业务需求所以最终还是需要结合具体的业务需求和场景来选择合适的部署方式和配置参数在后面我们还会继续进行Kafka的深入解析如果你对此有兴趣可以直接订阅本 kafka 专栏
