Kafka 原理详解之消费者机制（二）

落日映苍穹つ 2022-05-18 10:09 196阅读 0赞

**Kafka中Topic与Partition关系**

Topic是一个消息存储概念，也可以认为是一个消息集合；不同的topic存储着不同的消息，一个topic包含多个分区Partition（至少包含一个），它允许多个producer往它发送消息，也允许多个consumer 消费topic上的消息；

![70][]

Partition是一个消息分区，是topic中真正存储消息的地方，不同的Partition存储不同的消息；  
Partition存储在哪里呢？  
    我们通过命令行去创一个一个topic时，他会在log.dir=/tmp/kafka-logs下创建Partition分区，Partition默认就存储在/tmp/kafka-logs目录下，这个地址可以配置；tmp目录有保存期限超过期限会被清除掉，使用生产环境下需要指定存储目录；  
当producer往broker发送消息，消息在添加到Partition中时，会给给消息分配一个offset，offset是消息在分区中的唯一编号，kafka根据offset值保证了分区内消息的顺序性；offset是不跨区的，也就是说kafka保证同一个分区消息的有序性；

![70 1][]

producer往broker发送消息时，kafka会根据Partition的消息分区机制，将消息均匀的分配到每一个Partition上，有利于提升消息的存储于读取性能；

kafka的消息分发策略是什么？

1.  消息是kafka中最基本的数据单元，一条消息由key,value组成，producer往broker中的指定topic中发送一条消息，producer会根据这条消息的key的hashcode值%分区数取模，来确定这个消息分配到那个Partition分区；
    
    1.  如：testtopic主题下有三个分区(parition),我们发送一条key=哈哈哈的消息，“哈哈哈”.hashCode()%3=0,这条消息就保存在第一个分区partition\_0中;
2.  如果发布消息时key设置为null，在配置参数metadata.max.age.ms配置的时间范围之内随机一个分区new Random().nextInt(metadata.max.age.ms)进行存储，这个时间参数10分钟更新一次；

消费者是如何消费消息的呢？

在我们的项目，每一个topic都有多个partition，多个partition的好处在于，他对broker进行的分片，每个分片存储合理的消息数，提升了io性能；一个topic一般会由多个consumer进行消费，这就是消费的一个负载均衡机制，那么多个消费者如何去消费多个分区呢？

1.  消费端可以通过设置TopicPartition来指定消费的分区；TopicPartition topicPartition = new TopicPartition(topic,0);0：代表分区数；  
    consumer.assign(Arrays.asList(topicPartition));
2.  消费端存在consumer group消费者组，一个消费者组中存在多个消费者，但是他们的groupID是一致，当一个消费者组订阅了一个topic主题是，组内的消费协调一起消费topic中的所有分区，且一个分区只能由一个消费者消费，也就是分区不支持并发；此时存在一定分区分配策略：
    
    1.  Range strategy（范围分区）：
    2.  范围分区是针对同一个topic主题而言的：把分区按照序号排序，消费者按照字母排序，分区数/取整/消费者数，得到每个消费需要消费的分区数。如：testTopic主题有10个分区，p0,p1,,,p9 三个消费者c1,c2,c3此时10/3得到每个消费者消费3了分区，多出一个分区，c\_1多消费一个分区；
    3.  范围分区的弊端：如果这个消费者组订阅了10个topic，每个topic都是10个分区，此时c1就比c2和c3多消费10个分区，造成c1过多消费负担；
    4.  RoundRobin strategy（轮询分区）：
    5.  轮询分区策略是把所有 partition 和所有 consumer 线程都 列出来，然后按照 hashcode 进行排序。最后通过轮询算 法分配 partition 给消费线程。如果所有 consumer 实例的 订阅是相同的，那么 partition 会均匀分布。

消费端的触发重新负载均衡rebalance？

什么情况下会发生消费者的重新负载均衡呢？

1.  当consumer group 新增消费者；
2.  consumer group有消费者退出时，比如主机停机；
3.  topic新增分区时，分区的数量发生变化时； kafka consuemr 的 rebalance 机制规定了一个 consumer group 下的所有 consumer 如何达成一致来分配订阅 topic 的每个分区。实现重新负载均衡机制：范围分区和轮询分区；

谁来执行rebalance以及管理consumer group呢？

Kafka 提供了一个角色:coordinator 来执行对于consumer group 的管理；  
consumer group如何确定自己的coordinate?  
消费者向kafka集群中的任意一个broker发送一个GroupCoordinatorRequest请求，服务端会返回一个负载最小的broker节点的id，并将改broker设置为Coordinate；之后改group中的所有consumer成员会和Coordinate进行通信；

Rebalance的执行分为两个过程

在执行rebalance之前，需要保证Coordinate已经确定好了；

1.  JoinGroup 的过程  
    join:consumer加入到consumer group时，在这一步中，所有的consumer会向Coordinate发送joinGroup请求，Coordinate收到所有的consumer发来的joinGroup请求，会从中选择一个consumer为leader角色（班长），并给组成员返回一个response信息；  
    response信息中有leaderid,  
    members：组成员信息,consumer group 中全部的消费者的订阅信息，只有发给给leader的members才有消息，其他消费者的members为null；  
    membermetadate:对应消费者的订阅信息；  
    generationid:年代信息，对于每一轮 rebalance， generationid 都会递增。  
    总体来说：joingroup的过程就是每个消费者向Coordinate发送一个请求，来确定消息者中谁是他们的班长（leader）,班长会拿到班级名单和班级订阅的资料书（订阅的消息）；
2.  Synchronizing Group State 阶段  
    每个消费者都会向 coordinator 发送 syncgroup 请求，不过只有 leader 节点会发送消费分区分配方案，其他消费者的请求没起什么作用。当 leader 把方案发给 coordinator 以后， coordinator 会把结果设置到 SyncGroupResponse 中返回给每一个消费者。这 样所有成员都知道自己应该消费哪个分区。
    
     *  consumer group 的分区分配方案是在客户端执行的!Kafka 将这个权利下放给客户端主要是因为这样做可以有更好的灵活性

offset存储在哪里？

前面讲到了offset是消息在一个分区中的唯一编号，kafka根据offset保证了同一个分区内消息的顺序性；这是对于消息保存在partition中而言的；  
offset对于应用层的消息者来说，它相当于一个指针，消费者每次消费了消息并提交以后，会保存当前消费到的一个offset指针位置，等消费者下次继续消费消息时从offset位置开始；  
消费者每次消费消息提交的offset保存在哪里呢？  
在kafka中，默认提供一个**consumer\_offsets的topic主题，offset信息写入到了这个topic分区中。分为50个分区，分区中保存了每一个consumer group 提交的offset值。  
那么问题来了，一个消费者组中offset应该保存在**consumer\_offsets中那个分区上呢？

例如：一个叫groupID为“hahaha”的消费组，里面的消费者提交的offset保存的分区=“hahaha”.hashCOde()%分区数（这里默认是50）=22，也就是保存在**consumeroffsets22这个分区里；这50分区的命名**consumeroffsets\[0-49\];

查看offset信息：h kafka-simple-consumer-shell.sh --topic \_consumeroffsets --partition 35 --broker-list 这里是kafka集群的主机ip地址：9092用逗号隔开（如192...1:9092,192....2:9092等等） --formatter

[70]: /images/20220518/f06784af0c4643639bf80f017196eec5.png
[70 1]: /images/20220518/20f251ebeb2d4fdd93f5a2e51126ab74.png