zoukankan      html  css  js  c++  java

  • 理解 Kafka 中的 Topic 和 Partition

    首先要注意的是，Kafka 中的 Topic 和 ActiveMQ 中的 Topic 是不一样的。

    在 Kafka 中，Topic 是一个存储消息的逻辑概念，可以认为是一个消息集合。每条消息发送到 Kafka 集群的消息都有一个类别。物理上来说，不同的 Topic 的消息是分开存储的，每个 Topic 可以有多个生产者向它发送消息，也可以有多个消费者去消费其中的消息。

    每个 Topic 可以划分多个分区（每个 Topic 至少有一个分区），同一 Topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个 offset，它是消息在此分区中的唯一编号，Kafka 通过 offset 保证消息在分区内的顺序，offset 的顺序不跨分区，即 Kafka 只保证在同一个分区内的消息是有序的。

    消息是每次追加到对应的 Partition 的后面：


    Topic & Partition 的存储

    Topic 是一个逻辑上的概念，具体的存储还是基于 Partition 来的。

    创建一个 test2 Topic（注意这里的 partitions 参数为 3）：

    可以进入 /tmp/kafka-logs 目录下进行查看（当前机器 IP 是 192.168.220.135）：

    在 135 机器：

    在另外一台 136 机器上： 

    可以发现在 135 机器上有 test2-0 和 test2-2，在 136 机器上有 test2-1。接下来再结合 Kafka 的消息分发策略来看。

    消息分发

    Kafka 中最基本的数据单元就是消息，而一条消息其实是由 Key + Value 组成的（Key 是可选项，可传空值，Value 也可以传空值），这也是与 ActiveMQ 不同的一个地方。在发送一条消息时，我们可以指定这个 Key，那么 Producer 会根据 Key 和 partition 机制来判断当前这条消息应该发送并存储到哪个 partition 中（这个就跟分片机制类似）。我们可以根据需要进行扩展 Producer 的 partition 机制（默认算法是 hash 取 %）。

    扩展自己的 partition：

    1. package dongguabai.kafka.partition;
    2.  
    3. import org.apache.kafka.clients.producer.Partitioner;
    4. import org.apache.kafka.common.Cluster;
    5. import org.apache.kafka.common.PartitionInfo;
    6.  
    7. import java.util.List;
    8. import java.util.Map;
    9. import java.util.Random;
    10.  
    11. /**
    12. * 消息发送后会调用自定义的策略
    13. *
    14. * @author Dongguabai
    15. * @date 2019/1/18 15:40
    16. */
    17. public class MyPartitioner implements Partitioner {
    18.  
    19. @Override
    20. public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    21. //获取当前 topic 有多少个分区（分区列表）
    22. List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    23. int partitionNum = 0;
    24. if (key == null) { //之前介绍过 Key 是可以传空值的
    25. partitionNum = new Random().nextInt(partitions.size()); //随机
    26. } else {
    27. //取 %
    28. partitionNum = Math.abs((key.hashCode()) % partitions.size());
    29. }
    30. System.out.println("key：" + key + "，value：" + value + "，partitionNum：" + partitionNum);
    31. //发送到指定分区
    32. return partitionNum;
    33. }
    34.  
    35. @Override
    36. public void close() {
    37.  
    38. }
    39.  
    40. @Override
    41. public void configure(Map<String, ?> configs) {
    42.  
    43. }
    44. }

    改造 Kafka Demo 中的 Producer 的代码：

    1. package dongguabai.kafka;
    2.  
    3. import org.apache.kafka.clients.producer.*;
    4.  
    5. import java.util.Properties;
    6. import java.util.concurrent.ExecutionException;
    7.  
    8. /**
    9. * @author Dongguabai
    10. * @date 2019/1/17 11:26
    11. */
    12. public class KafkaProducerDemo extends Thread {
    13. /**
    14. * 消息发送者
    15. */
    16. private final KafkaProducer<Integer, String> producer;
    17.  
    18. /**
    19. * topic
    20. */
    21. private final String topic;
    22.  
    23. private final Boolean isAsync;
    24.  
    25. public KafkaProducerDemo(String topic, Boolean isAsync) {
    26. this.isAsync = isAsync;
    27. //构建相关属性
    28. //@see ProducerConfig
    29. Properties properties = new Properties();
    30. //Kafka 地址
    31. properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.220.135:9092,192.168.220.136:9092");
    32. //kafka 客户端 Demo
    33. properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerDemo");
    34. //The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent.
    35. /**发送端消息确认模式：
    36. * 0：消息发送给broker后，不需要确认（性能较高，但是会出现数据丢失，而且风险最大，因为当 server 宕机时，数据将会丢失）
    37. * 1：只需要获得集群中的 leader节点的确认即可返回
    38. * -1/all:需要 ISR 中的所有的 Replica进行确认（集群中的所有节点确认），最安全的，也有可能出现数据丢失（因为 ISR 可能会缩小到仅包含一个 Replica）
    39. */
    40. properties.put(ProducerConfig.ACKS_CONFIG, "-1");
    41.  
    42. /**【调优】
    43. * batch.size 参数（默认 16kb）
    44. * public static final String BATCH_SIZE_CONFIG = "batch.size";
    45. *
    46. * producer对于同一个 分区 来说，会按照 batch.size 的大小进行统一收集进行批量发送，相当于消息并不会立即发送，而是会收集整理大小至 16kb.若将该值设为0，则不会进行批处理
    47. */
    48.  
    49. /**【调优】
    50. * linger.ms 参数
    51. * public static final String LINGER_MS_CONFIG = "linger.ms";
    52. * 一个毫秒值。Kafka 默认会把两次请求的时间间隔之内的消息进行搜集。相当于会有一个 delay 操作。比如定义的是1000（1s），消息一秒钟发送5条，那么这 5条消息不会立马发送，而是会有一个 delay操作进行聚合，
    53. * delay以后再次批量发送到 broker。默认是 0，就是不延迟（同 TCP Nagle算法），那么 batch.size 也就不生效了
    54. */
    55. //linger.ms 参数和batch.size 参数只要满足其中一个都会发送
    56.  
    57. /**【调优】
    58. * max.request.size 参数（默认是1M） 设置请求最大字节数
    59. * public static final String MAX_REQUEST_SIZE_CONFIG = "max.request.size";
    60. * 如果设置的过大，发送的性能会受到影响，同时写入接收的性能也会受到影响。
    61. */
    62.  
    63. //设置 key的序列化，key 是 Integer类型，使用 IntegerSerializer
    64. //org.apache.kafka.common.serialization
    65. properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
    66. //设置 value 的序列化
    67. properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    68.  
    69. //指定分区策略
    70. properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"dongguabai.kafka.partition.MyPartitioner");
    71.  
    72. //构建 kafka Producer，这里 key 是 Integer 类型，Value 是 String 类型
    73. producer = new KafkaProducer<Integer, String>(properties);
    74. this.topic = topic;
    75. }
    76.  
    77. public static void main(String[] args) {
    78. new KafkaProducerDemo("test2",true).start();
    79. }
    80.  
    81. @Override
    82. public void run() {
    83. int num = 0;
    84. while (num < 100) {
    85. String message = "message--->" + num;
    86. System.out.println("start to send message 【 " + message + " 】");
    87. if (isAsync) { //如果是异步发送
    88. producer.send(new ProducerRecord<Integer, String>(topic, message), new Callback() {
    89. @Override
    90. public void onCompletion(RecordMetadata metadata, Exception exception) {
    91. if (metadata!=null){
    92. System.out.println("async-offset："+metadata.offset()+"-> partition"+metadata.partition());
    93. }
    94. }
    95. });
    96. } else { //同步发送
    97. try {
    98. RecordMetadata metadata = producer.send(new ProducerRecord<Integer, String>(topic, message)).get();
    99. System.out.println("sync-offset："+metadata.offset()+"-> partition"+metadata.partition());
    100. } catch (InterruptedException | ExecutionException e) {
    101. e.printStackTrace();
    102. }
    103. }
    104. num++;
    105. try {
    106. Thread.sleep(1000);
    107. } catch (InterruptedException e) {
    108. e.printStackTrace();
    109. }
    110. }
    111. }
    112. }

    在 Consumer 中接收消息的时候输出分区：

    1. package dongguabai.kafka;
    2.  
    3. import org.apache.kafka.clients.consumer.ConsumerConfig;
    4. import org.apache.kafka.clients.consumer.ConsumerRecord;
    5. import org.apache.kafka.clients.consumer.ConsumerRecords;
    6. import org.apache.kafka.clients.consumer.KafkaConsumer;
    7.  
    8. import java.util.Collections;
    9. import java.util.Properties;
    10.  
    11. /**
    12. * @author Dongguabai
    13. * @date 2019/1/17 11:55
    14. */
    15. public class KafkaConsumerDemo extends Thread {
    16.  
    17. private final KafkaConsumer<Integer, String> kafkaConsumer;
    18.  
    19. public KafkaConsumerDemo(String topic) {
    20. //构建相关属性
    21. //@see ConsumerConfig
    22. Properties properties = new Properties();
    23. properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.220.135:9092,192.168.220.136:9092");
    24. //消费组
    25. /**
    26. * consumer group是kafka提供的可扩展且具有容错性的消费者机制。既然是
    27. 一个组，那么组内必然可以有多个消费者或消费者实例((consumer instance),
    28. 它们共享一个公共的ID，即group ID。组内的所有消费者协调在一起来消费订
    29. 阅主题(subscribed topics)的所有分区(partition)。当然，每个分区只能由同一
    30. 个消费组内的一个consumer来消费.后面会进一步介绍。
    31. */
    32. properties.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaConsumerDemo");
    33.  
    34. /** auto.offset.reset 参数 从什么时候开始消费
    35. * public static final String AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset";
    36. *
    37. * 这个参数是针对新的groupid中的消费者而言的，当有新groupid的消费者来消费指定的topic时，对于该参数的配置，会有不同的语义
    38. * auto.offset.reset=latest情况下，新的消费者将会从其他消费者最后消费的offset处开始消费topic下的消息
    39. * auto.offset.reset= earliest情况下，新的消费者会从该topic最早的消息开始消费
    40. auto.offset.reset=none情况下，新的消费组加入以后，由于之前不存在 offset，则会直接抛出异常。说白了，新的消费组不要设置这个值
    41. */
    42.  
    43. //enable.auto.commit
    44. //消费者消费消息以后自动提交，只有当消息提交以后，该消息才不会被再次接收到（如果没有 commit，消息可以重复消费，也没有 offset），还可以配合auto.commit.interval.ms控制自动提交的频率。
    45. //当然，我们也可以通过consumer.commitSync()的方式实现手动提交
    46. properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    47.  
    48. /**max.poll.records
    49. *此参数设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔
    50. 要处理的最大值。通过调整此值，可以减少poll间隔
    51. */
    52.  
    53. //间隔时间
    54. properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    55. //反序列化 key
    56. properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
    57. //反序列化 value
    58. properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    59. //构建 KafkaConsumer
    60. kafkaConsumer = new KafkaConsumer<>(properties);
    61. //设置 topic
    62. kafkaConsumer.subscribe(Collections.singletonList(topic));
    63. }
    64.  
    65.  
    66. /**
    67. * 接收消息
    68. */
    69. @Override
    70. public void run() {
    71. while (true) {
    72. //拉取消息
    73. ConsumerRecords<Integer, String> consumerRecord = kafkaConsumer.poll(100000000);
    74. for (ConsumerRecord<Integer, String> record : consumerRecord) {
    75. //record.partition() 获取当前分区
    76. System.out.println(record.partition()+"】】 message receive 【" + record.value() + "】");
    77. }
    78. }
    79. }
    80.  
    81. public static void main(String[] args) {
    82. new KafkaConsumerDemo("test2").start();
    83. }
    84.  
    85. }

    首先启动 Consumer，再启动 Producer：

    可以看到是能够对的上的。

    默认情况下，Kafka 采用的是 hash 取 % 的分区算法。如果 Key 为 null，则会随机分配一个分区。这个随机是在这个参数“metadata.max.age.ms“的时间范围内随机选择一个。对于这个时间段内，如果 Key 为 null，则只会发送到唯一的分区。这个值默认情况下是 10 分钟更新一次（因为 partition 状态可能会发生变化）。

    关于 Metadata

    Metadata 包含 Topic 和 Partition 和 broker 的映射关系，每一个 Topic 的每一个 partition，需要知道对应的 broker 列表是什么，Leader 是谁，Follower 是谁。这些信息都是存储在 Metadata 这个类中。

    消费端如何消费指定分区

    Consumer 可以指定具体消费的分区。

    再重新启动 Consumer 和 Producer：

    可以看到 Consumer 只消费了分区为 1 的消息。

     以上是单个 Consumer 消费（指定）分区的情况。一般每个 Topic 都会有多个 partition（主要是用于数据分片，减少消息的容量，从而提升 I/O 性能）。当然也可以使用多个 Consumer 从而提高消费能力，有一个消费组的概念（具体可参看 https://blog.csdn.net/Dongguabai/article/details/86520617）。

    如果 Consumer1、Consumer2 和 Consumer3 都属于 group.id 为 1 的消费组。那么 Consumer1 就会消费 p0，Consumer2 就会消费 p1，Consumer3 就会消费 p2。

    可以先测试一下。创建三个 Consumer。要注意的是这里不能使用指定分区的方式：

    ‘

    而且它们都是同一个消费组：

    同时启动三个 Consumer，和 Producer：

    可以看到三个 Consumer 分别消费三个 Partition，很均匀。对同一个 Group 来说，其中的 Consumer 可以消费指定分区也可以消费自动分配的分区（这里是 Consumer 数量和 partition 数量一致，均匀分配）。那么如果 Consumer 数量大于 partition 数量呢，如果 Consumer 数量小于 partition 数量呢，测试也很简单，这里就不多做测试了。

    要注意的是如果 Consumer 数量比 partition 数量多，会有的 Consumer 闲置无法消费，这样是一个浪费。如果 Consumer 数量小于 partition 数量会有一个 Consumer 消费多个 partition。Kafka 在 partition 上是不允许并发的。Consuemr 数量建议最好是 partition 的整数倍。 还有一点，如果 Consumer 从多个 partiton 上读取数据，是不保证顺序性的，Kafka 只保证一个 partition 的顺序性，跨 partition 是不保证顺序性的。增减 Consumer、broker、partition 会导致 Rebalance。

    Kafka 分区分配策略

    在 Kafka 中，同一个 Group 中的消费者对于一个 Topic 中的多个 partition 存在一定的分区分配策略。

    在 Kafka 中，存在两种分区分配策略，一种是 Range（默认）、另一种是 RoundRobin（轮询）。通过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 的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。

    假如在 Topic1 中有 10 个分区，3 个消费者线程，10/3 = 3，而且除不尽，那么消费者线程 C1-0 将会多消费一个分区，所以最后分区分配的结果是这样的：

    C1-0 将消费 0,1,2,3 分区
    C2-0 将消费 4,5,6 分区
    C3-0 将消费 7,8,9 分区

    假如在 Topic1 中有 11 个分区，那么最后分区分配的结果看起来是这样的：

    C1-0 将消费 0,1,2,3 分区
    C2-0 将消费 4, 5, 6, 7 分区
    C3-0 将消费 8,9,10 分区

    假如有两个 Topic：Topic1 和 Topic2，都有 10 个分区，那么最后分区分配的结果看起来是这样的：

    C1-0 将消费 Topic1 的 0,1,2,3 分区和 Topic1 的 0,1,2,3 分区
    C2-0 将消费 Topic1 的 4,5,6 分区和Topic2 的 4,5,6 分区 
    C3-0 将消费 Topic1 的 7,8,9 分区和Topic2 的 7,8,9 分区

    其实这样就会有一个问题，C1-0 就会多消费两个分区，这就是一个很明显的弊端。

    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 新增了分区（也就是分区数量发生了变化）

    Kafka Consuemr 的 Rebalance 机制规定了一个 Consumer group 下的所有 Consumer 如何达成一致来分配订阅 Topic 的每个分区。而具体如何执行分区策略，就是前面提到过的两种内置的分区策略。而 Kafka 对于分配策略这块，提供了可插拔的实现方式，也就是说，除了这两种之外，我们还可以创建自己的分配机制。

    谁来执行 Rebalance 以及管理 Consumer 的 group 呢?

    Consumer group 如何确定自己的 coordinator 是谁呢，消费者向 Kafka 集群中的任意一个 broker 发送一个 GroupCoord inatorRequest 请求，服务端会返回一个负载最小的 broker 节点的 id，并将该 broker 设置为 coordinator。

    JoinGroup 的过程

    在 Rebalance 之前，需要保证 coordinator 是已经确定好了的，整个 Rebalance 的过程分为两个步骤，Join和Syncjoin：表示加入到 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 中。这样所有成员都知道自己应该消费哪个分区。

    Consumer group 的分区分配方案是在客户端执行的！Kafka 将这个权利下放给客户端主要是因为这样做可以有更好的灵活性。
     

    参考资料：

    https://blog.csdn.net/Dongguabai/article/details/86520617
  • 相关阅读:
    python学习笔记（unittest）
    python学习笔记（session）
    python学习笔记（requests）
    jmeter的学习（配置环境）
    Codeforces 576D. Flights for Regular Customers 题解
    Codeforces 1316F. Battalion Strength 题解
    2020年第十一届蓝桥杯省赛J-网络分析(带权并查集)
    第十一届蓝桥杯b组省赛 C.合并检测(内附详细的样例)
    蓝桥杯2020.7月真题走方格(到达终点的不同方案数)(记忆化搜索+DP)
    Codeforces Global Round 11 A. Avoiding Zero(思维构造)
  • 原文地址：https://www.cnblogs.com/lidabo/p/13671522.html
Copyright © 2011-2022 走看看