wordpress将公网ip改为域名,潍坊seo管理,网页设计与制作教程第二版考试,徐州泰安抖音代运营优质博文#xff1a;IT-BLOG-CN
一、消费者Rebalance机制
在Apache Kafka中#xff0c;消费者组 Consumer Group会在以下几种情况下发生重新平衡Rebalance#xff1a; 【1】消费者加入或离开消费者组#xff1a; 当一个新的消费者加入消费者组或一个现有的消费者离开消费…优质博文IT-BLOG-CN
一、消费者Rebalance机制
在Apache Kafka中消费者组 Consumer Group会在以下几种情况下发生重新平衡Rebalance 【1】消费者加入或离开消费者组 当一个新的消费者加入消费者组或一个现有的消费者离开消费者组时Kafka会触发重新平衡以重新分配分区给消费者。 【2】消费者崩溃或失去连接 如果Kafka检测到某个消费者崩溃或失去连接例如由于网络问题或消费者进程被终止它会触发重新平衡。 【3】主题的分区数量发生变化 如果一个主题的分区数量增加或减少Kafka会触发重新平衡以确保新的分区被分配给消费者组中的消费者。 【4】消费者组协调器变更 消费者组协调器是负责管理消费者组的一个Kafka Broker。如果消费者组协调器发生变更例如协调器所在的Broker崩溃也会触发重新平衡。 【5】消费者组成员发送心跳失败 消费者需要定期向消费者组协调器发送心跳heartbeat以表明它们仍然活跃。如果心跳失败协调器会认为该消费者已经失去连接从而触发重新平衡。
rebalance只针对subscribe这种不指定分区消费的情况如果通过assign这种消费方式指定了分区kafka不会进行rebanlance。
Kafka在高峰期重平衡rebalancing会导致消费者组的停顿影响系统的性能和稳定性。为了避免在高峰期发生重平衡可以采取以下几种策略 【1】优化分区分配策略 使用RangeAssignor或StickyAssignor等分区分配策略来减少重平衡的频率和影响。
RangeAssignor 是Kafka默认的分区分配策略之一它将分区按范围分配给消费者。
我们通过一个具体的例子来说明RangeAssignor如何分配分区。
假设我们有一个Kafka主题my-topic它有6个分区P0, P1, P2, P3, P4, P5并且我们有3个消费者C1, C2, C3在一个消费者组中。
初始分配假设初始分配如下
C1: P0, P1
C2: P2, P3
C3: P4, P5消费者组成员变化现在假设C2离开了消费者组那么RangeAssignor会重新分配分区以确保分区尽量按顺序和均匀地分配给剩余的消费者。新的分配可能如下
C1: P0, P1, P2
C3: P3, P4, P5在这个过程中RangeAssignor将分区按顺序重新分配给剩余的消费者确保每个消费者分配到的分区尽量连续。
新消费者加入现在假设有一个新消费者C4加入了消费者组RangeAssignor会再次按顺序和均匀地分配分区。新的分配可能如下
C1: P0, P1
C3: P2, P3
C4: P4, P5在这个过程中RangeAssignor将分区重新分配以确保每个消费者分配到的分区尽量连续和均匀。
通过这个例子我们可以看到RangeAssignor的分配策略 1、将分区按顺序分配给消费者。 2、当消费者组成员变化时重新分配分区以确保分区尽量按顺序和均匀地分配给所有消费者。 3、分区分配尽量保持连续性。 这种策略的好处是分区分配简单且稳定减少了分区在消费者组成员变化时的重新分配范围从而减少了重平衡的频率和影响。
以下是配置RangeAssignor的代码示例
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Properties;public class RangeAssignorExample {public static void main(String[] args) {Properties props new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);props.put(ConsumerConfig.GROUP_ID_CONFIG, example-group);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);// 设置分区分配策略为 RangeAssignorprops.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.RangeAssignor);KafkaConsumerString, String consumer new KafkaConsumer(props);// 订阅主题consumer.subscribe(List.of(example-topic));// 消费消息的逻辑// ...}
}StickyAssignor 是Kafka 2.4及以上版本引入的一种分区分配策略它的目标是尽量保持分区分配的稳定性减少重平衡的频率。
我们通过一个具体的例子来说明StickyAssignor如何分配分区。
假设我们有一个Kafka主题my-topic它有6个分区P0, P1, P2, P3, P4, P5并且我们有3个消费者C1, C2, C3在一个消费者组中。
初始分配假设初始分配如下
C1: P0, P1
C2: P2, P3
C3: P4, P5消费者组成员变化现在假设C2离开了消费者组那么StickyAssignor会尽量保持现有的分区分配不变并重新分配C2的分区。新的分配可能如下
C1: P0, P1, P2
C3: P3, P4, P5在这个过程中StickyAssignor尽量保持C1和C3的分区分配不变只是将C2的分区重新分配给其他消费者。
新消费者加入现在假设有一个新消费者C4加入了消费者组StickyAssignor会尝试保持现有的分区分配不变并将分区尽量均匀地分配给所有消费者。新的分配可能如下
C1: P0, P1
C3: P4, P5
C4: P2, P3在这个过程中StickyAssignor保持了C1和C3的分区不变并将C2的分区重新分配给C4。
通过这个例子我们可以看到StickyAssignor的分配策略 1、尽量保持现有的分区分配不变。 2、当消费者组成员变化时尽量最小化分区在消费者之间的移动。 3、尽量保持分区分配的平衡性。 这种策略的好处是减少了重平衡带来的影响提高了分区分配的稳定性减少了因分区移动带来的数据重新加载和处理的开销。
以下是配置StickyAssignor的代码示例
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Properties;public class StickyAssignorExample {public static void main(String[] args) {Properties props new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);props.put(ConsumerConfig.GROUP_ID_CONFIG, example-group);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);// 设置分区分配策略为 StickyAssignorprops.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, org.apache.kafka.clients.consumer.StickyAssignor);KafkaConsumerString, String consumer new KafkaConsumer(props);// 订阅主题consumer.subscribe(List.of(example-topic));// 消费消息的逻辑// ...}
}或者在配置中进行指定
group.idmy-consumer-group
partition.assignment.strategyorg.apache.kafka.clients.consumer.StickyAssignor【2】增加session.timeout.ms和heartbeat.interval.ms增加session.timeout.ms和heartbeat.interval.ms的值这样可以减少消费者因为心跳超时而被认为失效从而触发重平衡。
1、session.timeout.ms是消费者与Kafka broker之间的会话超时时间。如果在这个时间内Kafka broker没有收到某个消费者的心跳broker就会认为该消费者已经失效并触发重平衡。 2、heartbeat.interval.ms是消费者发送心跳给Kafka broker的时间间隔。心跳是消费者向broker表示自己仍然活跃的方式。
session.timeout.ms30000
heartbeat.interval.ms30003、heartbeat.interval.ms的值通常要远小于session.timeout.ms的值。这样可以确保在会话超时之前消费者有多次机会发送心跳。一般建议session.timeout.ms至少是heartbeat.interval.ms的10倍以确保有足够的时间进行多次心跳尝试。
【3】合理配置消费者组确保消费者组中的消费者数量稳定避免频繁地增加或减少消费者。尽量在低峰期进行消费者的添加或移除操作。
【4】优化消费者性能提高消费者的处理能力确保消费者能够及时处理消息避免因为处理延迟导致的重平衡。使用异步处理或批量处理来提高消费者的吞吐量。
【5】监控和报警实时监控Kafka集群和消费者组的状态设置报警机制当检测到重平衡风险时及时采取措施。
【6】使用静态成员Static MembershipKafka 2.3及以上版本支持静态成员功能可以通过配置group.instance.id来减少重平衡的频率。 group.instance.id是Kafka 2.4.0引入的一个配置项用于为每个消费者实例指定一个唯一的标识符。当消费者组中的消费者具有唯一的group.instance.id时Kafka可以更智能地处理消费者组成员的变化从而减少不必要的重平衡。 静态成员通过配置group.instance.id消费者实例变成了“静态成员”即使它们暂时断开连接Kafka也会保留它们的成员身份。这与传统的动态成员没有group.instance.id不同动态成员在断开连接后会被移除从而触发重平衡。
group.idmy-consumer-group
group.instance.idconsumer-instance-1【7】调整rebalance.timeout.ms增加rebalance.timeout.ms的值确保消费者有足够的时间完成重平衡过程避免因超时导致的频繁重平衡。
消费者Rebalance分区分配策略
主要包含四种relalance策略RangeAssignor范围分配策略RoundRobinAssignor轮询分配策略StickyAssignor粘性分配策略CooperativeStickyAssignor协作粘性分配策略之前已经讲过两个这里聊聊剩下的两个
RoundRobinAssignor轮询分配策略
RoundRobinAssignor采用轮询的方式将分区分配给消费者。它会将所有分区和消费者按照字典顺序排序然后依次将每个分区分配给下一个消费者直到所有分区都被分配完毕。
CooperativeStickyAssignor协作粘性分配策略
CooperativeStickyAssignor是StickyAssignor的改进版本它引入了协作重平衡的概念使得重平衡过程更加平滑减少了重平衡期间的停顿时间。
二、Rebalance 过程 第一阶段选择组协调器 组协调器GroupCoordinator每个consumer group都会选择一个broker作为自己的组协调器coordinator负责监控这个消费组里的所有消费者的心跳以及判断是否宕机然后开启消费者rebalance。
consumer group中的每个consumer启动时会向kafka集群中的某个节点发送FindCoordinatorRequest请求来查找对应的组协调器GroupCoordinator并跟其建立网络连接。
组协调器选择方式consumer消费的offset要提交到__consumer_offsets的哪个分区这个分区leader对应的broker就是这个consumer group的coordinator
第二阶段加入消费组JOIN GROUP 在成功找到消费组所对应的GroupCoordinator之后就进入加入消费组的阶段在此阶段的消费者会向GroupCoordinator发送JoinGroupRequest请求并处理响应。然后GroupCoordinator从一个consumer group中选择第一个加入group的consumer作为leader(消费组协调器)把consumer group情况发送给这个leader接着这个leader会负责制定分区方案。
第三阶段SYNC GROUP consumer leader通过给GroupCoordinator发送SyncGroupRequest接着GroupCoordinator就把分区方案下发给各个consumer他们会根据指定分区的leader broker进行网络连接以及消息消费。
