kafka的位移

这篇具有很好参考价值的文章主要介绍了kafka的位移。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

概要

本文主要总结kafka的位移是如何管理的,在broker端如何通过命令行查看到位移信息,并从代码层面总结了位移的提交方式。

消费位移

对于 Kafka 中的分区而言,它的每条消息都有唯一offset ,用来表示消息在分区中对应位置;对于消费者来说,它也有 offset 的概念,消费者使用 offse 来表示消费到分区中某个消息所在的位置。可通过命令行在查看到一个群组,在topic中两者当前的位置
bin/kafka-consumer-groups.sh --bootstrap-server node1:9092 --describe --group kafka-boot

[root@node1 kafka_2.13-3.2.1]# bin/kafka-consumer-groups.sh --bootstrap-server node1:9092 --describe  --group kafka-boot

Consumer group 'kafka-boot' has no active members.

GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
kafka-boot      test-error-topic 0          26              26              0               -               -               -
kafka-boot      normal-test      0          23              24              1               -               -               -

这里对offse 做些区分 对于消息在分区中的位置 CURRENT-OFFSET称为“偏移量” 或消息位移;对于消费者消费到的位置,LOG-END-OFFSET称为“位移 ,有时候也会更明确地称之为“消费位移“。

生产者位移跟消费者位移的关系可以用下图来说明:
kafka的位移,kafka,主题位移,提交位移的方式

总结几个需要注意的点:

  • 分区副本有两种类型
    领导者副本:生产者跟消费者的请求都只会经过领导者副本
    跟随者副本:首领之外的副本,不处理客户端请求,从领导者副本那里通过拉取的方式同步消息
  • 消费位移存储在Zookeeper或Kafka中,新消费者客户端,偏移量存储咋Kafka内部主题 __consumer_offsets
  • 消费者提交的位移是当前消费消息位移的下一个位置,即:lastConsumeedOffset+1

__consumer_offsets主题

Consumer需要向Kafka记录自己的位移数据,这个汇报过程称为提交位移(Committing Offsets)。

老版本 Consumer 的位移是提交到 ZooKeeper 中保存的。当 Consumer 重启后,它能自动从 ZooKeeper 中读取位移数据,从而在上次消费截止的地方继续消费。这种设计使得Kafka Broker 不需要保存位移数据,减少了 Broker 端需要持有的状态空间,因而有利于实现高伸缩性。

但是,ZooKeeper 其实并不适用于这种高频的写操作,Kafka 社区自 0.8.2.x 版本开始推出了全新的位移管理
机制,将 Consumer 的位移数据作为一条条普通的 Kafka 消息,提交到 __consumer_offsets 中。可以这么说,
__consumer_offsets 的主要作用是保存 Kafka 消费者的位移信息。这种方式能够满足高频的写操作。

两个相关参数:
offsets.topic.num.partitions : 设置 __consumer_offsets主题的分区数,默认是50个分区
offsets.topic.replication.factor : 设置__consumer_offsets主题的副本数,默认是3(下载安装的包中此值可能为1 )

当Kafka 集群中的第一个 Consumer 程序启动时,Kafka 会自动创建位移主题

一共有50个分区,那么消费者将位移提交到了哪个分区呢?

通过如下公式可以选出consumer消费的offset要提交到__consumer_offsets的哪个分区,这个分区leader对应的broker
就是这个consumer group的coordinator
公式:Math.abs(groupID.hashCode()) % numPartitions

Kafka 1.0.2及以后提供了kafka_consumer_groups.sh脚本供用户查看consumer信息

1. 创建一个topic,分区数设置为1,副本数设置为1

[root@node1 kafka_2.13-3.2.1]# bin/kafka-topics.sh --bootstrap-server node1:9092 --create --topic test-offset --partitions 1 --replication-factor 1
Created topic test-offset.

[root@node1 kafka_2.13-3.2.1]# bin/kafka-topics.sh --bootstrap-server node1:9092 --describe --topic test-offset
Topic: test-offset      TopicId: in6gxQ5OQS6x9R8V3oJ7AQ PartitionCount: 1       ReplicationFactor: 1    Configs: segment.bytes=1073741824
        Topic: test-offset      Partition: 0    Leader: 0       Replicas: 0     Isr: 0

2. 向主题test-offset中发送消息

[root@node1 kafka_2.13-3.2.1]# bin/kafka-console-producer.sh --broker-list node1:9092 --topic test-offset
>hello

3. 创建一个消费组,并从头开始消费

[root@node1 kafka_2.13-3.2.1]# bin/kafka-console-consumer.sh --bootstrap-server node1:9092  --from-beginning --consumer-property group.id=testOffsetGroup   --topic test-offset
hello

4. 用代码根据上面的公式计算消费组testOffsetGroup提交位移的分区数

@Test
void getCommitOffsetPartitionTest() {
    String groupId = "testOffsetGroup";
    // 运行结果为16
    System.out.println(Math.abs(groupId.hashCode() % 50));
}
  1. 将kafka配置文件consumer.properties中设置exclude.internal.topics=false,并重启服务
    6. 查看主题__consumer_offsets第16分区上的信息,可以看到消费组testOffsetGroup提交的位移确实保存在了16分区上

[root@node1 kafka_2.13-3.2.1]# bin/kafka-console-consumer.sh --topic __consumer_offsets --partition 16 --bootstrap-server node1:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896116191, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896121189, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896126188, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896131188, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896133573, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896162124, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896167124, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896172123, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896177124, expireTimestamp=None)
[testOffsetGroup,test-offset,0]::OffsetAndMetadata(offset=1, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1691896178781, expireTimestamp=None)

从上面也可看出__consumer_offsets topic的每一日志项的格式都是:
[Group, Topic, Partition]::[OffsetMetadata[Offset, Metadata], CommitTime, ExpirationTime]

客户端提交消费位移是使用OffsetCommitRequest 请求实现的,其结构如下
kafka的位移,kafka,主题位移,提交位移的方式

__consumer_offsets这个主题中的消息格式为KV对,key为[Group, Topic, Partition],value可以简单理解为记录了偏移量;这样的记录方式,使得broker端不需要关系group下有多少个消费者,新增消费者或者减少消费者发生重平衡时,都能准确地定位到对应地分区应该从哪个位置开始消费。

位移提交

鉴于位移提交甚至是位移管理对 Consumer 端的巨大影响,Kafka,特别是KafkaConsumer API,提供了多种提交位移的方法。从用户的角度来说,位移提交分为自动提交和手动提交;从 Consumer 端的角度来说,位移提交分为同步提交和异步提交。

自动提交

自动提交,就是指 Kafka Consumer 在后台默默地为你提交位移
两个重要的参数

  • enable.auto.commit设置是否自动提交位移,默认是true
  • auto.commit.interval.ms:设置自动提交为true时,该参数生效,标识多久提交一次位移,默认5s,
public static void main(String[] args) {
      Map<String, Object> configs = new HashMap<>();
      configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
      configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
      configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer1");
      configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "con1");
      // 设置偏移量自动提交。自动提交是默认值。这里做示例。
      configs.put("enable.auto.commit", "true");
      // 偏移量自动提交的时间间隔
      configs.put("auto.commit.interval.ms", "2000");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(configs);
      consumer.subscribe(Collections.singleton("tp_demo_01"));

      while (true) {
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
          for (ConsumerRecord<String, String> record : records) {
              System.out.println(record.topic()
                      + "\t" + record.partition()
                      + "\t" + record.offset()
                      + "\t" + record.key()
                      + "\t" + record.value());
          }
      }
 }

设置了 enable.auto.commit 为 true,Kafka 会保证在开始调用 poll 方法时,提交上次 poll 返回的所有消息。从顺序上来说,poll 方法的逻辑是先提交上一批消息的位移,再处理下一批消息,因此它能保证不出现消费丢失的情况。但是会出现消息重复消费

在默认情况下,Consumer 每 5 秒自动提交一次位移。现在,我们假设提交位移之后的 3秒发生了 Rebalance 操作。在 Rebalance 之后,所有 Consumer 从上一次提交的位移处继续消费,但该位移已经是 3 秒前的位移数据了,故在Rebalance 发生前 3 秒消费的所有数据都要重新再消费一次。虽然你能够通过减少 auto.commit.interval.ms 的值来提高提交频率,但这么做只能缩小重复消费的时间窗口,不可能完全消除它。这是自动提交机制的一个缺陷。

手动同步提交

开启手动提交位移的方法就是设置enable.auto.commit 为 false。但是,仅仅设置它为 false 还不够,因为你只是告诉
Kafka Consumer 不要自动提交位移而已,你还需要调用相应的 API 手动提交位移。

public static void main(String[] args) {
    Map<String, Object> configs = new HashMap<>();
    configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
    configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
    configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer1");
    configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "con1");

   configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(configs);
    consumer.subscribe(Collections.singleton("tp_demo_01"));
    while (true) {
        ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofSeconds(1));
        process(records); // 处理消息
        try {
            consumer.commitSync();
        } catch (CommitFailedException e) {
           handle(e); // 处理提交失败异常
        }
    }
}

调用 commitSync() 时,Consumer 程序会处于阻塞状态,直到远端的 Broker 返回提交结果,这个状态才会结束,这样就会影响TPS。

鉴于此问题,还有另外一个提交方式

手动异步提交

public static void main(String[] args) {
   Map<String, Object> configs = new HashMap<>();
   configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
   configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

   configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
   configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer1");
   configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
   configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "con1");
   configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
   KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(configs);
   while (true) {
       ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
       process(records); // 处理消息
       consumer.commitAsync((offsets, exception) -> {
           if (exception != null) {
               handle(exception);
           }
       });
   }
}

commitAsync 是否能够替代 commitSync 呢?答案是不能。commitAsync 的问题在于,出现问题时它不会自动重试。因为它是异步操作,倘若提交失败后自动重试,那么它重试时提交的位移值可能早已经“过期”或不是最新值了。因此,异步提交的重试其实没有意义,所以 commitAsync 是不会重试的。

是手动提交,需要将 commitSync 和 commitAsync 组合使用才能到达最理想的效果,原因有两个:文章来源地址https://www.toymoban.com/news/detail-651841.html

  1. 利用 commitSync 的自动重试来规避那些瞬时错误,比如网络的瞬时抖动,Broker 端 GC 等。这些问题都是短暂的,自动重试通常都会成功。
  2. 不希望程序总处于阻塞状态,影响 TPS。
public static void main(String[] args) {
   Map<String, Object> configs = new HashMap<>();
   configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
   configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

   configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
   configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer1");
   configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
   configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "con1");
   configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
   KafkaConsumer<String, String> consumer = new KafkaConsumer<String,
           String>(configs);
   consumer.subscribe(Collections.singleton("tp_demo_01"));
   try {
       while (true) {
           ConsumerRecords<String, String> records =
                   consumer.poll(Duration.ofSeconds(1));
           consumer.commitAsync();
           process(records); // 处理消息
           consumer.commitAsync(); // 异步提交
       }
   } catch (Exception e) {
       handle(e); // 处理异常
   } finally {
       try {
           consumer.commitSync();// 最后一次提交使用同步阻塞式提交
       } finally {
           consumer.close();
       }

   }
}

到了这里,关于kafka的位移的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(46)
  • Kafka - 3.x offset位移不完全指北

    由于consumer在消费过程中可能会出现断电宕机等故障,consumer恢复后,需要从故障前的位置的继续消费,所以consumer需要 实时记录自己消费到了哪个offset ,以便故障恢复后继续消费。 Kafka 0.9版本之前,consumer默认将offset保存在Zookeeper中,从0.9版本开始,consumer默认将offset保存在

    2024年02月06日
    浏览(42)
  • Kafka 入门到起飞系列 - 消费者组管理、位移管理

    消费者组 - Consumer Group 上文我们已经讲过消费者组了,我们知道消费组的存在可以保证一个主题下一个分区的消息只会被组内一个消费者消费,从而避免了消息的重复消费 消费者组是Kafka 提供的可扩展且具有容错性的消费者机制 消费组有一个或多个消费者,消费者可以是一

    2024年02月15日
    浏览(42)
  • Kafka:自动创建主题

    如果 broker 端配置参数 auto.create.topics.enable 设置为 true(默认值是 true) ,那么当生 产者向一个未创建的主题发送消息时, 会自动创建一个分区数为 num.partitions(默认值为 1)、副本因子为 default.replication.factor(默认值为 1)的主题。 除此之外,当一个消费者 开始从未知主题

    2024年02月07日
    浏览(37)
  • kafka操作命令-主题管理

    目录 1.创建主题 2.查看主题 3.修改主题 4.删除主题 1.1 创建名为:test-topic的主题,命令如下: 执行结果如下:  登录ZooKeeper客户端查看所创建的主题元数据信息,“test-topic”元数据信息如下: 可以看到,该主题有5个分区、1个副本  ⭐️⭐️⭐️⭐️ 1.2 在创建主题时,我

    2024年02月08日
    浏览(36)
  • kafka主题支持路由功能

    我们知道rabbitmq是支持消息路由的功能的,但是当我们统一消息中间件到kafka后,有一些旧的应用依然想要使用消息路由的功能时,我们可以怎么让kafka也支持消息路由的功能呢? 为了不影响kafka服务端,我们选择改造kafka的客户端,主要的想法是通过消息过滤的方式来达到消

    2024年02月13日
    浏览(28)
  • 【基础】Kafka -- 主题与分区

    主题管理包括创建主题、查看主题消息、修改主题以及删除主题等操作,Kafka 提供的 kafka-topics.sh 脚本来执行这些操作,脚本位于 $KAFKA_HOME/bin/ 目录下,该脚本实际上是调用了 kafka.admin.TopicCommand 类来执行主题管理的操作。 简单创建与查看 若 broker 端的配置参数 auto.create.top

    2023年04月22日
    浏览(36)
  • 05、Kafka ------ 各个功能的作用解释(主题和分区 详解,用命令行和图形界面创建主题和查看主题)

    Kafka 主题虽然也叫 topic,但它和 Pub-Sub 消息模型中 topic 主题及 AMQP 的 topic 都不同(AMQP 的 topic 只是 Exchange 的类型)。 Kafka 的主题只是盛装消息的逻辑容器(注意是逻辑容器),主题之下会分为若干个分区,分区才是盛装消息的物理容器。 ▲ 消息组织方式实际上是三级结构

    2024年02月03日
    浏览(42)
  • kafka如何动态消费新增topic主题

    一、解决痛点 使用spring-kafka客户端,每次新增topic主题,都需要硬编码客户端并重新发布服务,操作麻烦耗时长。kafkaListener虽可以支持通配符消费topic,缺点是并发数需要手动改并且重启服务 。对于业务逻辑相似场景,创建新主题动态监听可以用kafka-batch-starter组件 二、组件

    2023年04月21日
    浏览(39)
  • flink如何监听kafka主题配置变更

    从前一篇文章我们知道flink消费kafka主题时是采用的手动assign指定分区的方式,这种消费方式是不处理主题的rebalance操作的,也就是消费者组中即使有消费者退出或者进入也是不会触发消费者所消费的分区的,那么疑问就来了,那是否比如kafka主题分区变多,或者新增了满足

    2024年02月14日
    浏览(37)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包