关于 Topic 和 Partition:
Topic:
在 kafka 中,topic 是一个存储消息的逻辑概念,可以认为是一个消息集合。每条消息发送到 kafka 集群的消息都有一个类别。物理上来说,不同的 topic 的消息是分开存储的,每个 topic 可以有多个生产者向它发送消息,也可以有多个消费者去消费其中的消息。
Partition:
每个 topic 可以划分多个分区(每个 Topic 至少有一个分区),同一 topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时,都会被分配一个 offset(称之为偏移量),它是消息在此分区中的唯一编号,kafka 通过 offset保证消息在分区内的顺序,offset 的顺序不跨分区,即 kafka只保证在同一个分区内的消息是有序的。下图中,对于名字为 test 的 topic,做了 3 个分区,分别是p0、p1、p2.
? 每一条消息发送到 broker 时,会根据 partition 的规则选择存储到哪一个 partition。如果 partition 规则设置合理,那么所有的消息会均匀的分布在不同的partition中,这样就有点类似数据库的分库分表的概念,把数据做了分片处理。
Topic&Partition 的存储:
Partition 是以文件的形式存储在文件系统中,比如创建一个名为 firstTopic 的 topic,其中有 3 个 partition,那么在kafka 的数据目录(/tmp/kafka-log)中就有 3 个目录,firstTopic-0~3,命名规则是<topic_name>-<partition_id>,创建3个分区的topic:
sh kafka-topics.sh --create --zookeeper 192.168.254.135:2181 --replication-factor 1 --partitions 3 --topic firstTopic
kafka 消息分发策略:
消息是 kafka 中最基本的数据单元,在 kafka 中,一条消息由 key、value 两部分构成,在发送一条消息时,我们可以指定这个 key,那么 producer 会根据 key 和 partition 机制来判断当前这条消息应该发送并存储到哪个 partition 中。我们可以根据需要进行扩展 producer 的 partition 机制。
我们可以通过如下代码来实现自己的分片策略:
public class MyPartition implements Partitioner {//实现Partitioner接口 private Random random=new Random(); @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { //获得分区列表 List<PartitionInfo> partitionInfos=cluster.partitionsForTopic(topic); int partitionNum=0; if(key==null){ partitionNum=random.nextInt(partitionInfos.size()); //随机分区 }else{ partitionNum=Math.abs((key.hashCode())%partitionInfos.size()); } System.out.println("key ->"+key+"->value->"+value+"->"+partitionNum); return partitionNum; //指定发送的分区值 } @Override public void close() { } @Override public void configure(Map<String, ?> configs) { } }
然后基于之前的代码在producer上需要在消息发送端增加配置:指定自己的partiton策略
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.gupaoedu.kafka.MyPartition");
消息默认的分发机制:
默认情况下,kafka 采用的是 hash 取模的分区算法。如果Key 为 null,则会随机分配一个分区。这个随机是在这个参数”metadata.max.age.ms”的时间范围内随机选择一个。对于这个时间段内,如果 key 为 null,则只会发送到唯一的分区。这个值值哦默认情况下是 10 分钟更新一次。关 于 Metadata ,简单理解就是Topic/Partition 和 broker 的映射关系,每一个 topic 的每一个 partition,需要知道对应的 broker 列表是什么,leader是谁、follower 是谁。这些信息都是存储在 Metadata 这个类里面。
消费端如何消费指定的分区:
通过下面的代码,就可以消费指定该 topic 下的 0 号分区。其他分区的数据就无法接收。
//消费指定分区的时候,不需要再订阅 //kafkaConsumer.subscribe(Collections.singletonList(topic)); //消费指定的分区 TopicPartition topicPartition=new TopicPartition(topic,0); kafkaConsumer.assign(Arrays.asList(topicPartition));
消息的消费原理:
在实际生产过程中,每个 topic 都会有多个 partitions,多个 partitions 的好处在于,一方面能够对 broker 上的数据进行分片有效减少了消息的容量从而提升 io 性能。另外一方面,为了提高消费端的消费能力,一般会通过多个consumer 去消费同一个 topic ,也就是消费端的负载均衡机制,也就是我们接下来要了解的,在多个 partition 以及多个 consumer 的情况下,消费者是如何消费消息的?kafka 存在 consumer group的 概 念 , 也 就是 group.id 一 样 的 consumer ,这些consumer 属于一个 consumer group,组内的所有消费者协调在一起来消费订阅主题的所有分区。当然每一个分区只能由同一个消费组内的 consumer 来消费,那么同一个consumer group 里面的 consumer 是怎么去分配该消费哪个分区里的数据的呢?举个简单的例子就是如果存在的分区输,即partiton的数量于comsumer数量一致的时候,每个comsumer对应一个分区,如果comsumer数量多于分区,那么多出来的数量的comsumer将不工作,相反则是其中将会有comsumer消费多个分区。
分区分配策略:
在 kafka 中,存在两种分区分配策略,一种是 Range(默认)、另 一 种 另 一 种 还 是 RoundRobin ( 轮 询 )。 通 过comsumer的配置partition.assignment.strategy 这个参数来设置。
Range strategy(范围分区):
Range 策略是对每个主题而言的,首先对同一个主题里面的分区按照序号进行排序,并对消费者按照字母顺序进行排序。假设我们有 10 个分区,3 个消费者,排完序的分区将会是 0, 1, 2, 3, 4, 5, 6, 7, 8, 9;消费者线程排完序将会是C1-0, C2-0, C3-0。然后将 partitions 的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽,那么前面几个消费者线程将会多消费一个分区。比如我们有 10 个分区,3 个消费者线程, 10 / 3 = 3,而且除不尽,那么消费者线程 C1-0 将会多消费一个分区,所以最后分区分配的结果看起来是这样的:
C1-0 将消费 0, 1, 2, 3 分区
C2-0 将消费 4, 5, 6 分区
C3-0 将消费 7, 8, 9 分区
假如我们有 11 个分区,那么最后分区分配的结果看起来是这样的:
C1-0 将消费 0, 1, 2, 3 分区
C2-0 将消费 4, 5, 6, 7 分区
C3-0 将消费 8, 9, 10 分区
假如我们有 2 个主题(T1 和 T2),分别有 10 个分区,那么最后分区分配的结果看起来是这样的:
C1-0 将消费 T1 主题的 0, 1, 2, 3 分区以及 T2 主题的 0, 1, 2, 3 分区
C2-0 将消费 T1 主题的 4, 5, 6 分区以及 T2 主题的 4, 5, 6 分区
C3-0 将消费 T1 主题的 7, 8, 9 分区以及 T2 主题的 7, 8, 9 分区
可以看出,C1-0 消费者线程比其他消费者线程多消费了 2 个分区,这就是 Range strategy 的一个很明显的弊端.
RoundRobin strategy(轮询分区):
轮询分区策略是把所有 partition 和所有 consumer 线程都列出来,然后按照 hashcode 进行排序。最后通过轮询算法分配 partition 给消费线程。如果所有 consumer 实例的订阅是相同的,那么 partition 会均匀分布。假如按照 hashCode 排序完的 topic&partitions 组依次为 T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9,我们的消费者线程排序为 C1-0, C1-1, C2-0, C2-1,最后分区分配的结果为:
C1-0 将消费 T1-5, T1-2, T1-6 分区;
C1-1 将消费 T1-3, T1-1, T1-9 分区;
C2-0 将消费 T1-0, T1-4 分区;
C2-1 将消费 T1-8, T1-7 分区;
使用轮询分区策略必须满足两个条件
1. 每个主题的消费者实例具有相同数量的流
2. 每个消费者订阅的主题必须是相同的
什么时候会触发这个策略呢?当出现以下几种情况时,kafka 会进行一次分区分配操作,也就是 kafka consumer 的 rebalance。
1. 同一个 consumer group 内新增了消费者。
2. 消费者离开当前所属的 consumer group,比如主动停机或者宕机。
3. topic 新增了分区(也就是分区数量发生了变化)。
4.消费者主动取消订阅topic
kafka consuemr 的 rebalance 机制规定了一个 consumer group 下的所有 consumer 如何达成一致来分配订阅 topic的每个分区。而具体如何执行分区策略,就是前面提到过的两种内置的分区策略。而 kafka 对于分配策略这块,提供了可插拔的实现方式, 也就是说,除了这两种之外,我们还可以创建自己的分配机制。可以通过继承 AbstractPartitionAssignor 抽象类实现 assign 来做到。
谁来执行 Rebalance 以及管理 consumer 的 group 呢?
Kafka 提供了一个角色:coordinator(协调员) 来执行对于 consumer group 的管理,当 consumer group 的第一个 consumer 启动的时候,它会去和 kafka server(broker) 确定谁是它们组的 coordinator。之后该 group 内的所有成员都会和该 coordinator 进行协调通信。consumer group 如何确定自己的 coordinator 是谁呢? 消费 者 向 kafka 集 群 中 的 任 意 一 个 broker 发 送 一 个GroupCoordinatorRequest 请求,服务端会返回一个负载最 小 的 broker 节 点 的 id , 并 将 该 broker 设 置 为coordinator。在 rebalance 之前,需要保证 coordinator 是已经确定好了的,整个 rebalance 的过程分为两个步骤 ,一个是JoinGroup 的过程,在这个过程之后会进入一个Synchronizing Group State 阶段。那么这两个阶段都做了什么呢?
JoinGroup 的过程:
表示加入到 consumer group 中,在这一步中,所有的成员都会向 coordinator 发送 joinGroup 的请求。一旦所有成员都发送了 joinGroup 请求,那么 coordinator 会选择一个 consumer 担任 leader 角色,并把组成员信息和订阅信息发送消费者。下图就是描述了这么一个过程,并且请求与响应中携带的一些重要的信息。
protocol_metadata: 序列化后的消费者的订阅信息
leader_id: 消费组中的消费者,coordinator 会选择一个座位 leader,对应的就是 member_id
member_metadata 对应消费者的订阅信息
members:consumer group 中全部的消费者的订阅信息
generation_id:年代信息,类似于 zookeeper 的时候的 epoch 是一样的,对于每一轮 rebalance ,generation_id 都会递增。主要用来保护 consumer group。隔离无效的 offset 提交。也就是上一轮的 consumer 成员无法提交 offset 到新的 consumer group 中。
Synchronizing Group State 阶段:
进入了 Synchronizing Group State阶段,主要逻辑是向 GroupCoordinator 发 送SyncGroupRequest 请求,并且处理 SyncGroupResponse响应,简单来说,就是 leader 将消费者对应的 partition 分配方案同步给 consumer group 中的所有 consumer,每个消费者都会向 coordinator 发送 syncgroup 请求,不过只有 leader 节点会发送分配方案,其他消费者只是打打酱油而已。当 leader 把方案发给 coordinator 以后,coordinator 会把结果设置到 SyncGroupResponse 中。这样所有成员都知道自己应该消费哪个分区。
member_assignment :在syncGroup发送请求的时候,只有leader角色的comsumer才会去发送这个信息,而其他消费端是空的。然后会通过coordinator去分发给各个comsumer。
? consumer group 的分区分配方案是在客户端执行的!Kafka 将这个权利下放给客户端主要是因为这样做可以有更好的灵活性
offset :
每个 topic可以划分多个分区(每个 Topic 至少有一个分区),同一topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时,都会被分配一个 offset(称之为偏移量),它是消息在此分区中的唯一编号,kafka 通过 offset 保证消息在分区内的顺序,offset 的顺序不跨分区,即 kafka 只保证在同一个分区内的消息是有序的; 对于应用层的消费来说,每次消费一个消息并且提交以后,会保存当前消费到的最近的一个 offset。那么 offset 保存在哪里?
offset 在哪里维护?
在 kafka 中,提供了一个__consumer_offsets_* 的一个topic , 把 offset 信 息 写 入 到 这 个 topic 中 。__consumer_offsets——按保存了每个 consumer group某一时刻提交的 offset 信息。__consumer_offsets 默认有50 个分区。可以在 /tmp/kafka-logs/ 下查看。那么如何查看对应的 consumer_group 保存在哪个分区中呢?
通过Math.abs(“groupid”.hashCode())%groupMetadataTopicPartitionCount ; 由 于 默 认 情 况 下groupMetadataTopicPartitionCount 有 50 个分区,计算得到的结果为:4, 意味着当前的 consumer_group 的位移信息保存在__consumer_offsets 的第 4个分区,执行如下命令,可以查看当前 consumer_goup 中的offset 位移信息
sh kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 4 --broker-list 192.168.254.135:9092,192.168.254.136:9092,192.168.254.137:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"
执行语句可以看到如下结果:
这个意思就是 KafkaConsumerDemo1 消费组在 testTopic 中现在的 offsets 现在是 111.
消息的存储:
首先我们需要了解的是,kafka 是使用日志文件的方式来保存生产者和发送者的消息,每条消息都有一个 offset 值来表示它在分区中的偏移量。Kafka 中存储的一般都是海量的消息数据,为了避免日志文件过大,Log 并不是直接对应在一个磁盘上的日志文件,而是对应磁盘上的一个目录,这个目录的明明规则是<topic_name>_<partition_id>比如创建一个名为 firstTopic 的 topic,其中有 3 个 partition,那么在 kafka 的数据目录(/tmp/kafka-log,这里可以通过server.properties中的log.dirs=/tmp/kafka-logs去修改)中就有 3 个目录,firstTopic-0~3多个分区在集群中的分配 如果我们对于一个 topic,在集群中创建多个 partition,那么 partition 是如何分布的呢?
1.将所有 N Broker 和待分配的 i 个 Partition 排序
2.将第 i 个 Partition 分配到第(i mod n)个 Broker 上
结合前面讲的消息分发策略,就应该能明白消息发送到 broker 上,消息会保存到哪个分区中,并且消费端应该消费哪些分区的数据了。
原文地址:https://www.cnblogs.com/wuzhenzhao/p/10137490.html