电商网站产品模块,网站的超链接怎么做,网络营销与直播电商专业介绍,网站备案需要什么流程这里写目录标题1.生产者1.1 生产者发送原理1.2 分区有什么好处?1.3 生产消息时, 是如何决定消息落盘到哪个分区的?1.4 生产者如何提高吞吐量1.5 如何保证生产的消息不丢失(能成功落盘)1.6 ack为-1, 就肯定不会丢失数据吗?1.7 生产者重复发送消息的场景1.8 生产者如何保证数据…
这里写目录标题1.生产者1.1 生产者发送原理1.2 分区有什么好处?1.3 生产消息时, 是如何决定消息落盘到哪个分区的?1.4 生产者如何提高吞吐量1.5 如何保证生产的消息不丢失(能成功落盘)1.6 ack为-1, 就肯定不会丢失数据吗?1.7 生产者重复发送消息的场景1.8 生产者如何保证数据不重复发送1.9 什么是幂等性1.10 生产者事务2.消费者2.1 消息队列的两种模式2.2 Kafka怎么实现这两种消费模式2.3 消费者组的初始化流程2.4 如果消费者组中的消费者挂了, 那么组中消费者所消费的分区是否会调整?2.5 消费组形成的条件2.6 offset3.Broker3.1 Broker Controller是什么?3.2 Broker Controller如何选举分区Leader3.3 分区副本的作用4.最常见问题汇总4.1 Kafka中怎么保证消息不会丢失和不重复消费4.1.1 生产者---保证数据不丢失4.1.2 生产者---保证数据不重新消费4.1.3 消费者---保证数据不丢失1.生产者
1.1 生产者发送原理
KafkaProducer将消息封装成ProducerRecord, 然后通过拦截链, 根据指定序列化方式进行序列化, 其次在分区器中根据设置的分区策略进行数据分区,封装成TopicPartition, TopicPartition中就包含了目标partition的信息其后分区器将消息写入RecordAccumulator进行缓冲, RecordAccumulator是一个双端队列, RecordAccumulator中维护了一个ConcurrentMapTopicPartition, Deque 类型的集合, 其中的Key是TopicPartition,它用来标识目标partition(消息的最终存储位置), Value是Deque 队列,用来缓冲发往目标 partition 的消息。当Deque达到一定阈值后,就会唤醒sender线程将消息发送到kafka集群 1.2 分区有什么好处?
便于合理使用存储资源, 可以将一个完整的数据,切割成多个快(Partition), 分布存储在多个Broker上,. 合理控制分区的任务, 可以实现负载均衡的效果提高并行能力, 生产者可以以分区为单位发送数据, 消费者可以以分区为单位消费数据 1.3 生产消息时, 是如何决定消息落盘到哪个分区的?
指明partition的情况, 写入指定分区没有指明partition, 但有key, 通过key的hash值与topic的分区数进行取余操作, 得到partition的值既没有partition, 也没有key, kafka采用黏性分区器, 会随机选择一个分区, 并尽可能的一直使用该分区, 待该分区的ProducerBatch已满或者已完成, kafka再随机一个分区进行使用(和上一次的分区不同) 例如第一次随机选择了0号分区, 等0号分区当前批次满了(16K)或者linger.ms设置的时间到了, kafka再随机一个分区进行使用 1.4 生产者如何提高吞吐量
Deque达到一定阈值后,就会唤醒sender线程将消息发送到kafka集群, 这个阈值受两个参数影响
batch.size:批次大小, 默认16K 当Deque中的积压的消息达到16K后, 就会唤醒sender线程将消息打包发送到同一个分区linger.ms:等待时间,默认为0 如果Deque中的消息一直没有达到16K, 此时会根据linger.ms设置的时间,比如设置了1秒, 那么到了这个时间(上一个批次的消息发送完成后开始计时),即使数据没有达到16K, 也会唤醒sender线程发送消息
因为linger.ms默认为0, 所以来一个消息就会唤醒sender来发送消息, 这样的效率并不高(会频繁开启线程发送消息), 为了提高拉取速度的能力, 我们希望一次能发送很多消息
所以在生产环境中, 我们一般会修改linger.ms的值, 改为5~100ms, 而batch.size使用默认值即可 注意点:不能将batch.size和linger.ms设置的很大, 这样每批次消息的发送时间间隔就会很大(延迟过大)
// batch.size批次大小默认 16K
properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);// linger.ms等待时间默认 0
properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);// RecordAccumulator缓冲区大小默认 32Mbuffer.memory
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);// compression.type压缩默认 none可配置值 gzip、snappy、lz4 和 zstd
properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,snappy);1.5 如何保证生产的消息不丢失(能成功落盘)
生产消息可靠 消息能成功落盘(落到分区的所有副本中)
kafka为消费者提供了消息确认机制:
一、ack为0(数据会丢失) 生产者只管发送消息, 不用等待任何来自服务器的响应(不关系消息是否最终落盘) 不会重复发送消息
数据丢失场景:如果当中出现问题导致服务器没有收到消息, 没有落盘到partition生产者无从得知会造成消息丢失
二、ack为1(数据会丢失) 生产者发送消息后, 等待分区的leader落盘消息后应答 如果leader没收到应答,或是收到失败应答, 则会重新发送消息
数据丢失场景:如果leader落盘成功了, 向producer也收到了成功响应, 但是还没来得及将消息同步副本(follower), 此时leader挂了, 此时服务器会从follower中推选新的leader, 新的leader并没有同步消息, 而producer也不会再发了, 此时消息就丢失了
三、ack为-1(数据不会丢失) 生产者发送消息后, 等待分区的所有ISR副本(包括leader以及其他存活的副本)落盘消息后应答 producer只有收到分区中所有ISR副本的成功落盘消息后才认为是推送成功, 反之重新发送消息 1.6 ack为-1, 就肯定不会丢失数据吗?
如果分区的副本设置为1(即只有leader没有follower), 或者ISR中应答的最小副本数量(min.insync.replicas 默认为1)设置为1, 这种情况下就和ack1时效果是一样的, 存在数据丢失问题(leader:0, isr:0)
分区副本大于等于2(除了leader以外, 存在至少一个follow副本)ACK级别设置为-1(保证ISR中所有节点都存入消息)min.insync.replicas 2 ISR里应答的最小副本数量大于等于2(ISR中的数量至少有两个, 否则broker不处理这条消息, 并直接给生产者报错)
min.insync.replicas n代表的语义是如果生产者acksall而在发送消息时Broker的ISR数量没有达到nBroker不能处理这条消息需要直接给生产者报错。 1.7 生产者重复发送消息的场景
消息重复存在几个场景
生产端:服务器响应失败后, 基本的解决措施就是重发消息消费端: poll 一批数据处理完毕还没提交 offset 机子宕机重启了又会 poll 上批数据再度消费就造成了消息重复。
生产端重复发送消息的场景 Leader收到数据后, 将数据落盘, 并将数据同步到follower, 此时在给Producer应答时Leader宕机了, 此时Producer就会收到服务器传来的响应失败, 重新发送消息, 服务器会重新挑选一个follower成为leader, 而这个新的leader其实已经落盘了消息 1.8 生产者如何保证数据不重复发送
数据发送有三种情况
至少一次(At Least Once)最多一次(At Most Once)精确一次(Exactly Once)
一、至少一次 什么是至少一次:生产者发送到kafka集群, 至少kafka集群能收到一次数据 如何保证至少一次:ACK级别设置为-1或者all 分区副本大于等于2 ISR里应答的最小副本数量大于等于2 至少一次存在的问题:kafka集群重复收到数据的问题, 即可以保证数据不丢失但是不能保证数据不重复
二、最多一次 什么是最多一次:生产者发送到kafka集群, 不论成功与否, 只会发送一次 如何保证最多一次:ACK级别设置为0 最多一次会产生的问题:无法保证数据是否落盘, 即可以保证数据不重复但是不能保证数据不丢失
三、精确一次 什么是精确一次:数据既不会丢失, 也不会重复发送 如何保证精确一次:Kafka 0.11版本以后引入了一项重大特性幂等性和事务, 通过这两点来保证严格一次 1.9 什么是幂等性
精确一次 幂等性 至少一次
幂等性就是指Producer不论向Broker发送多少次重复数据Broker端都只会持久化一条保证了不重复。
如何保证幂等性? 一个消息会被封装成TopicPartition, TopicPartition中记录了以下几个信息:PID、Partition、SeqNumber; 重复数据的判断标准具有PID, Partition, SeqNumber相同主键的消息提交时Broker只会持久化一条。
PID是每个Producer在初始化时分配的一个唯一ID, 对于一个PID来说, Sequence Number是从0开始自增Partition 表示分区的标识Sequence Number是Producer在发送消息时, 会给每一条消息标识Sequence Number, 同一条消息被重复发送时, Sequence Number是不会递增
幂等性的条件
只能保证Producer在单个会话内不丢不重, 如果producer出现意外挂掉了再重启是无法保证幂等性, 因为PID已经改变了(单会话)幂等性无法跨域多个topic-partition, 只能保证单个partition内的幂等性(单分区)
所以幂等性只能保证的是在单分区单会话内不重复。
如何使用幂等性 enable.idempotence被设置成true后, Producer自动升级成幂等性Producer其他所有的代码逻辑都不需要改变。(enable.idempotence默认为true, 不需要手动开启)
properties.put(“enable.idempotence”, ture)
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG true)。1.10 生产者事务
开启事务, 必须开启幂等性(即enable.idempotence设置为true), 不需要保证精确一次(幂等性 至少一次) 2.消费者
2.1 消息队列的两种模式
一对一模式发布/订阅模式 2.2 Kafka怎么实现这两种消费模式
一、一对一模式 保证只有一个consumerGroup(消费者组)订阅topic(主题), 即使这个consumerGroup(消费者组)中存在多个consumer(消费者), 但是一个partitioner(分区)只能被一个consumer(消费者)消费, 这样就能保证消息只被一个消费者消费(一对一) 在同一个consumerGroup中一个partitioner(分区)只能被一个consumer(消费者)消费,一对一关系 一个consumer(消费者)可以消费多个partitioner(分区),一对多关系 二、发布/订阅模式 多个consumerGroup(消费者组)一起订阅topic(主题), consumerGroup(消费者组)在消费消息时, 都会在partitioner(分区)中创建对应的_consumer_offsets(消费者偏移量), 这个偏移量使用key-value的存储机构, key为consumerGroup.id topic 分区号, value是当前offset的值, 每过5毫秒(默认值), consumerGroup都会提交offset, kafka根据最新的offset来更新roup.idtopic分区号对应的value 2.3 消费者组的初始化流程
kafka集群根据consumer组的groupId计算出选择哪个broker的coordinator作为群组协调器组中所有消费者都会主动向coordinator发送joinGroup请求, 第一个加入的消费者称为leader消费者leader消费者负责制定消费策略, 并将消费策略发送给coordinatorcoordinator把消费方案下发给所有consumer每个消费者都会和coordinator保持心跳默认3s, 一旦超时session.timeout.ms45s该消费者会被移除并触发再平衡或者消费者处理消息的时间过长max.poll.interval.ms5分钟也会触发再平衡 2.4 如果消费者组中的消费者挂了, 那么组中消费者所消费的分区是否会调整?
当消费者群组里的消费者发生变化或者主题里的分区发生了变化都会导致再均衡现象的发生
有4种再平衡策略
RangeRoundRobinStickyCooperativeSticky 2.5 消费组形成的条件
形成一个消费者组的条件是所有消费者的groupid相同。 2.6 offset
形成一个消费者组的条件是所有消费者的groupid相同。 3.Broker
3.1 Broker Controller是什么?
在Kafka集群中某个Broker将被选举出来担任一种特殊的角色其用于管理和协调Kafka集群即管理集群中的所有分区的状态并执行相应的管理操作。每个Kafka集群任意时刻都只能有一个Controller。当集群启动时所有Broker都参与Controller的竞选最终有一个胜出一旦Controller在某个时刻崩溃集群中的其他的Broker会收到通知然后开启新一轮的Controller选举新选举出来的Controller将承担起之前Controller的所有工作。
controller的作用
维护每台Broker上的分区副本信息维护每个分区的Leader副本信息 3.2 Broker Controller如何选举分区Leader
选举规则:在isr中存活为前提, 按照AR中排在前面的优先, 例如AR[1, 0, 2], ISR[1, 2], 那么leader会按照1,2的顺序轮巡
对于topicA的partition0这个分区,它选举出broker1作为leader, 而broker0、broker2作为follower, controller会把这个信息告诉zookeeper(将节点信息上传到zookeeper),这是为了防止controller挂了后, 新的controller不知道主副本信息 3.3 分区副本的作用
kafka副本的作用:提高数据的可靠性kafka默认副本1个, 生产环境一般配置2个, 保证数据可靠性; 太多副本会增加磁盘存储空间, 增加网络上传数据传输, 降低效率kafka中副本分为:Leader和Follower, kafka生产者只会把数据发往Leader, 然后follower自己找leader进行数据同步kafka分区中所有的副本统称为AR(Assigned Repllicas), AR ISR OSR ISR: 表示和 Leader 保持同步的 Follower 集合。 如果 Follower 长时间未向 Leader 发送通信请求或同步数据则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定默认 30s。Leader 发生故障之后就会从 ISR 中选举新的 Leader。ISR: OSR: 表示 Follower 与 Leader 副本同步时延迟过多的副本。 4.最常见问题汇总
4.1 Kafka中怎么保证消息不会丢失和不重复消费
至少一次:
分区副本大于等于2(除了leader以外, 存在至少一个follow副本)ACK级别设置为-1(保证ISR中所有节点都存入消息)min.insync.replicas 2
ISR里应答的最小副本数量大于等于2(ISR中的数量至少有两个, 否则broker不处理这条消息, 并直接给生产者报错) 4.1.1 生产者—保证数据不丢失
分区副本大于等于2(除了leader以外, 存在至少一个follow副本)ACK级别设置为-1(保证ISR中所有节点都存入消息)min.insync.replicas 2
ISR里应答的最小副本数量大于等于2(ISR中的数量至少有两个, 否则broker不处理这条消息, 并直接给生产者报错) 4.1.2 生产者—保证数据不重新消费
精确一次 幂等性 至少一次
幂等性默认为开启, 底层使用Producer.id、Partition、SeqNumber作为重复数据的判断标准 4.1.3 消费者—保证数据不丢失
消费者为什么会出现消息丢失的情况, 我们得先了解offset提交机制
自动提交(默认) 消息poll下来后(还没消费), 直接提交offset, 速度很快, 但是可能会因为消费失败, 造成数据丢失手动提交 在消息消费时/ 消费后再提交offset
如果在消息处理完成前就提交了offset那么就有可能造成数据的丢失。为了避免数据丢失可以采用手动提交offset
消费者为什么会重复消费消息? kafka重复消费的根本原因就是“数据消费了但是offset没更新
