可以做专利聚类分析的免费网站,阜阳网站建设价格,常州住房和城乡建设部网站,即墨网络有限公司消息发送过程
使用Kafka发送消息时#xff0c;一般有两种方式分别是#xff1a;
同步发送异步发送
同步发送时#xff0c;可以在发送消息后#xff0c;通过get方法等待消息结果#xff0c;这种情况能够准确的拿到消息最终的发送结果#xff0c;要么是成功、要么是失败…消息发送过程
使用Kafka发送消息时一般有两种方式分别是
同步发送异步发送
同步发送时可以在发送消息后通过get方法等待消息结果这种情况能够准确的拿到消息最终的发送结果要么是成功、要么是失败。
而异步发送是采用了callback的方式进行回调的可以大大的的提升消息的吞吐量也可以根据回调来判断消息是否发送成功。
不管是同步发送还是异步发送最终都需要在Producer端把消息发送到Broker中那么这个过程大致如下: Kafka的Producer在发送消息时通常涉及两个线程主线程(Main)和发送线程(Sender)和一个消息累加器(RecordAccumulator) Main线程是Producer的入口负责初始化Producer的配置、创建KafkaProducer实例并执行发送逻辑。它会按照用户定义的发送方式(同步或异步)发送消息,然后等待消息发送完成。一条消息的发送在调用send方法后会经过拦截器、序列化器及分区器。拦截器主要用于在消息发送之前和之后对消息进行定制化的处理如对消息进行修改、记录日志、统计信息等。序列化器负责将消息的键和值对象转换为字节数组以便在网络上传输。分区器决定了一条消息被发送到哪个Partition中。它根据消息的键(如果有)或者特定的分区策略选择出一个目标Partition。 RecordAccumulator在Kafka Producer中起到了消息积累和批量发送的作用当Producer发送消息时不会立即将每条消息发送到Broker而是将消息添加到RecordAccumulator维护的内部缓冲区中RecordAccumulator会根据配置的条件(如batch.size、linger.ms)对待发送的消息进行批量处理。当满足指定条件时RecordAccumulator将缓冲区中的消息组织成一个批次(batch)然后一次性发送给Broker。如果发送失败或发生错RecordAccumulator可以从将消息重新分配到新的批次中进行重试。这样可以确保消息不会丢失,同时提高消息的可靠性。 Send线程是负责实际的消息发送和处理的。发送线程会定期从待发送队列中取出消息并将其发送到对应的Partition的 Leader Broker上。它主要负责网络通信操作,并处理发送请求的结果,包括确认的接收、错误处理等。 NetworkClient和Selector是两个重要的组件分别负责网络通信和1/0多路复用。
发送线程会把消息发送到Kafka集群中对应的Partition的Parrtition LeaderPartition Leader接收到消息后会对消息进行一系列的处理。它会将消息写入本地的日志文件(Log)存储为segment文件因为是顺序写segment文件也是顺序截断为了保证数据的可靠性和高可用性Kafka使用了消息复制机制。Leader Broker接收到消息后会将消息复制到其他副本(Partition Follower)。副本是通过网络复制数据的它们门会定期从LeaderBroker同步消息。
每一个Partition Follower在写入本地log之后会向Leader发送一个TACK但是我们的Producer其实也是需要依赖ACK才能知道消息有没有投递成功的而这个ACK是何时发送的,Producer又要不要关心呢? 这就涉及到了kafka的ack机制生产者会根据设置的request.required.acks参数不同选择等待或或直接发送下一条消息:
request.required.acks 0 表示Producer不等待来自Leader的ACK确认直接发送送下一条消息。在这种情况下如果Leader分片所在服务器发生宕机那么这些已经发送的数据会丢失。request.required.acks 1 表示Producer等待来自Leader的ACK确认当收到确认人后才发送下一条消息。在这种情况下消息一定会被写入到 Leader服务器但并不保证Follow节点已已经同步完成。所以如果在消息已经被写入Leader分片但是还未同步到Follower节点此时Leade分片所在服务器宕机了那么这条消息也就丢失了无法被消费到。request.required.acks -1 Leader会把消息复制到集群中的所有ISR(In-Sync Replicas同步副本)要等待所有ISR的ACK确认后再向Producer发送ACK消息然后Producer再继续发下下一条消息。
ISR机制
Kafka 中的 ISRIn-Sync Replicas机制是一种用于确保数据可靠性和一致性的重要机制。ISR 是一组副本它包括分区的领导者Leader和追随者Follower副本这些副本与领导者保持数据同步。以下是关于 Kafka 的 ISR 机制的详细介绍
意义
ISR 机制动态维护了一个与 Leader 副本保持同步的副本集合只有在 ISR 集合中的副本才有资格参与 Leader 的选举。通过 ISR 机制可以确保在 Leader 副本出现故障时能够快速从 ISR 集合中选举出新的 Leader从而避免数据丢失和服务中断。
用途
保证数据可靠性ISR 机制通过副本冗余机制提供了 Kafka 消息的高可靠性。实现故障转移ISR 机制可以做到故障转移保障服务的可用性。当 Leader 副本出现故障时Kafka 会从 ISR 集合中选举出新的 Leader从而保证服务的连续性。平衡复制方案ISR 机制平衡了主从架构下复制方案的选择同步/异步/少数服从多数让使用者根据参数自行选择。
实现方式
数据同步Leader Replica 接收到 Producer 发送的消息后将其写入本地日志并通过 Pull 模式等待 Follower Replica 主动拉取。Follower Replica 从 Leader Replica 拉取数据并写入本地日志后将拉取偏移量fetch offset返回给 Leader。同步状态监测Leader Replica 持续监控每个 Follower Replica 的拉取偏移量将其与自身的最新消息偏移量log end offset进行比较。若 Follower Replica 的拉取偏移量与 Leader 相差不超过一定阈值由 replica.lag.time.max.ms 参数控制则认为该 Follower 处于同步状态将其纳入 ISR。ISR 调整当 Follower Replica 因网络延迟、Broker 故障等原因导致拉取偏移量落后过多超出阈值时Leader Replica 会将其从 ISR 中移除。当 Follower Replica 恢复同步后再次将其加入 ISR。
详细过程
当消息被写入Kafka的分区时它首先会被写入Leader然后LLeader将消息复制给ISR中的所有副本。只有当ISR中的所有副本都成功地接收到并确认了消息后主副本才会认为消息已成功提交。这种机制确保了数据的可靠性和一致性。
在Kafka中ISR(In-Sync Replicas)列表的维护是通过副本状态和配置参数来进行的。具体的ISR列表维护机制在不同的Kafka版本中有所变化。
在0.9.x之前的版本Kafka有一个核心的参数:replica.lag.max.messages表示如果Follower落后 Leader的消息数量超过了这个参数值就认为Follower就会从ISR列表里移除。
但是基于replica.lag.max.messages 这种实现在瞬间高并发访问的情况下会有问题:比如Leader瞬间接收到几万条消息,然后所有Follower还没来得及同步过去,此时所有follower都会被踢出ISR列表。
Kafka从0.9.x版本开始引入了 replica.lag.max.ms参数表示如果果某个Follower的LEO (latest end offset)一直落后Leader超过了10秒那么才会被从ISR列表里移除。
这样的话,即使出现瞬间流量,导致Follower落后很多数据,但是只要在限定的时间内尽快追上来就行了。
总之通过 ISR 机制Kafka 可以保证在 Leader 副本出现故障时能够快速从 ISR 集合中选举出新的 Leader从而避免数据丢失和服务中断。同时ISR 机制也可以提高 Kafka 系统的可靠性和可用性。
