您的位置:首页 > 游戏 > 手游 > 北京建工集团有限公司官网_b2b外贸平台有哪些平台_seo怎么刷关键词排名_老铁外链工具

北京建工集团有限公司官网_b2b外贸平台有哪些平台_seo怎么刷关键词排名_老铁外链工具

2024/12/23 2:03:10 来源:https://blog.csdn.net/qq_43563660/article/details/143999601  浏览:    关键词:北京建工集团有限公司官网_b2b外贸平台有哪些平台_seo怎么刷关键词排名_老铁外链工具
北京建工集团有限公司官网_b2b外贸平台有哪些平台_seo怎么刷关键词排名_老铁外链工具

文章目录

  • 一、消费消息概览
    • 1.1、基本代码
    • 1.2、消费过程
  • 二、消费者组
    • 2.1、push & pull
    • 2.2、消费者组
  • 三、调度器Coordinator
  • 四、消费者分配策略
  • 五、偏移量offset
    • 5.1、起始偏移量
    • 5.2、指定偏移量消费
    • 5.3、偏移量提交
      • 5.3.1、自动提交
      • 5.3.2、手动提交
    • 5.4、偏移量的保存
  • 六、消费者事务

如果想了解kafka基础架构、生产者架构和kafka存储消息可以参考 kafka基础、 Kafka进阶_1.生产消息和 kafka进阶_2.存储消息。

一、消费消息概览

1.1、基本代码

public class KafkaConsumerTest {public static void main(String[] args) {// TODO 创建消费者配置参数集合Map<String, Object> paramMap = new HashMap<>();paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");// TODO 通过配置,创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);// TODO 订阅主题consumer.subscribe(Collections.singletonList("test"));// TODO 消费数据final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));// TODO 遍历数据for (ConsumerRecord<String, String> record : poll) {System.out.println( record.value() );}// TODO 关闭消费者consumer.close();}
}

主要的属性配置:

参数名参数作用类型推荐值
bootstrap.servers集群地址必须
key.deserializer对数据Key进行反序列化的类完整名称必须Kafka提供的字符串反序列化类:StringSerializer
value.deserializer对数据Value进行反序列化的类完整名称必须Kafka提供的字符串反序列化类:ValueSerializer
group.id消费者组ID,用于标识完整的消费场景,一个组中可以包含多个不同的消费者对象

1.2、消费过程

  消费者消费数据时,一般情况下,只是设定了订阅的主题名称,那是如何消费到数据的呢。我们这里说一下服务端拉取数据的基本流程:

在这里插入图片描述

  1. 服务端获取到用户拉取数据的请求:Kafka消费客户端会向Broker发送拉取数据的请求FetchRequest,服务端Broker获取到请求后根据请求标记FETCH交给应用处理接口KafkaApis进行处理。
  2. 通过副本管理器拉取数据:副本管理器需要确定当前拉取数据的分区,然后进行数据的读取操作。
  3. 判定首选副本:2.4版本前,数据读写的分区都是Leader分区,从2.4版本后,kafka支持Follower副本进行读取。主要原因就是跨机房或者说跨数据中心的场景,为了节约流量资源,可以从当前机房或数据中心的副本中获取数据。这个副本称之为首选副本。
  4. 拉取分区数据:Kafka的底层读取数据是采用日志段LogSegment对象进行操作的。
  5. 零拷贝:为了提高数据读取效率,Kafka的底层采用nio提供的FileChannel零拷贝技术,直接从操作系统内核中进行数据传输,提高数据拉取的效率。

二、消费者组

  从数据处理的角度来讲,消费者和生产者的处理逻辑都相对比较简单。Producer生产者的基本数据处理逻辑就是向Kafka发送数据,并获取Kafka的数据接收确认响应。

在这里插入图片描述

而消费者的基本数据处理逻辑就是向Kafka请求数据,并获取Kafka返回的数据。

在这里插入图片描述
  逻辑确实很简单,但是Kafka为了能够构建高吞吐,高可靠性,高并发的分布式消息传输系统,所以在很多细节上进行了扩展和改善:比如生产者可以指定分区,可以异步和同步发送数据,可以进行幂等性操作和事务处理。对应的,消费者功能和处理细节也进行了扩展和改善。

2.1、push & pull

  Kafka的主题如果就一个分区的话,那么在硬件配置相同的情况下,消费者Consumer消费主题数据的方式没有什么太大的差别。

在这里插入图片描述
  不过,Kafka为了能够构建高吞吐,高可靠性,高并发的分布式消息传输系统,它的主题是允许多个分区的,那么就会发现不同的消费数据的方式区别还是很大的。

1、如果数据由Kafka进行推送(push),那么多个分区的数据同时推送给消费者进行处理,明显一个消费者的消费能力是有限的,那么消费者无法快速处理数据,就会导致数据的积压,从而导致网络、存储等资源造成极大的压力,影响吞吐量和数据传输效率。

在这里插入图片描述
2.如果kafka的分区数据在内部可以存储的时间更长一些,再由消费者根据自己的消费能力向kafka申请(拉取)数据,那么整个数据处理的通道就会更顺畅一些。Kafka的Consumer就采用的这种拉取数据的方式。

在这里插入图片描述

2.2、消费者组

  消费者可以根据自身的消费能力主动拉取Kafka的数据,但是毕竟自身的消费能力有限,如果主题分区的数据过多,那么消费的时间就会很长。对于kafka来讲,数据就需要长时间的进行存储,那么对Kafka集群资源的压力就非常大。如果希望提高消费者的消费能力,并且减少kafka集群的存储资源压力。所以有必要对消费者进行横向伸缩,从而提高消息消费速率。

在这里插入图片描述
  不过这么做有一个问题,就是每一个消费者是独立,那么一个消费者就不能消费主题中的全部数据,简单来讲,就是对于某一个消费者个体来讲,主题中的部分数据是没有消费到的,也就会认为数据丢了,这个该如何解决呢?那如果我们将这多个消费者当成一个整体,是不是就可以了呢?这就是所谓的消费者组 Consumer Group。在kafka中,每个消费者都对应一个消费组,消费者可以是一个线程,一个进程,一个服务实例,如果kafka想要消费消息,那么需要指定消费哪个topic的消息以及自己的消费组id(groupId)。

在这里插入图片描述

三、调度器Coordinator

  消费者想要拉取数据,首先必须要加入到一个组中,成为消费组中的一员,同样道理,如果消费者出现了问题,也应该从消费者组中剥离。而这种加入组和退出组的处理,都应该由专门的管理组件进行处理,这个组件在kafka中,我们称之为消费者组调度器(Group Coordinator)。Group Coordinator是Broker上的一个组件,用于管理和调度消费者组的成员、状态、分区分配、偏移量等信息。每个Broker都有一个Group Coordinator对象,负责管理多个消费者组,但每个消费者组只有一个Group Coordinator。

在这里插入图片描述

四、消费者分配策略

五、偏移量offset

  偏移量offset是消费者消费数据的一个非常重要的属性。默认情况下,消费者如果不指定消费主题数据的偏移量,那么消费者启动消费时,无论当前主题之前存储了多少历史数据,消费者只能从连接成功后当前主题最新的数据偏移位置读取,而无法读取之前的任何数据,如果想要获取之前的数据,就需要设定配置参数或指定数据偏移量。

5.1、起始偏移量

在消费者的配置中,我们可以增加偏移量相关参数auto.offset.reset,用于从最开始获取主题数据:

public class KafkaConsumerTest {public static void main(String[] args) {// TODO 创建消费者配置参数集合Map<String, Object> paramMap = new HashMap<>();paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");paramMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");// TODO 通过配置,创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);// TODO 订阅主题consumer.subscribe(Arrays.asList("test"));while ( true ) {// TODO 消费数据final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));// TODO 遍历数据for (ConsumerRecord<String, String> record : poll) {System.out.println( record );}}}
}

auto.offset.reset有三种取值:

1、earliest
对于同一个消费者组,从头开始消费。就是说如果这个topic有历史消息存在,现在新启动了一个消费者组,且auto.offset.reset=earliest,那将会从头开始消费(未提交偏移量的场合)。

在这里插入图片描述
2、latest
对于同一个消费者组,消费者只能消费到连接topic后,新产生的数据(未提交偏移量的场合)。

在这里插入图片描述
3、none:生产环境不使用。

5.2、指定偏移量消费

除了上一节所讲的从最开始的偏移量或最后的偏移量读取数据以外,Kafka还支持从指定偏移量的位置开始消费数据:

public class KafkaConsumerOffsetTest {public static void main(String[] args) {Map<String, Object> paramMap = new HashMap<>();paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");KafkaConsumer<String, String> c = new KafkaConsumer<String, String>(paramMap);// TODO 订阅主题c.subscribe(Collections.singletonList("test"));// TODO 拉取数据,获取基本集群信息c.poll(Duration.ofMillis(100));// TODO 根据集群的基本信息配置需要消费的主题及偏移量final Set<TopicPartition> assignment = c.assignment();for (TopicPartition topicPartition : assignment) {if ( topicPartition.topic().equals("test") ) {c.seek(topicPartition, 0);}}// TODO 拉取数据while (true) {final ConsumerRecords<String, String> poll = c.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : poll) {System.out.println( record.value() );}}}
}

5.3、偏移量提交

  生产环境中,消费者可能因为某些原因或故障重新启动消费,那么如果不知道之前消费数据的位置,重启后再消费,就可能重复消费(earliest)或漏消费(latest)。所以Kafka提供了保存消费者偏移量的功能,而这个功能需要由消费者进行提交操作。这样消费者重启后就可以根据之前提交的偏移量进行消费了。

注意:一旦消费者提交了偏移量,那么kafka会优先使用提交的偏移量进行消费。此时,auto.offset.reset参数是不起作用的。

5.3.1、自动提交

  所谓的自动提交就是消费者消费完数据后,无需告知kafka当前消费数据的偏移量,而是由消费者客户端API周期性地将消费的偏移量提交到Kafka中。这个周期默认为5000ms,可以通过配置进行修改。

public class KafkaConsumerCommitAutoTest {public static void main(String[] args) {// TODO 创建消费者配置参数集合Map<String, Object> paramMap = new HashMap<>();paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// TODO 启用自动提交消费偏移量,默认取值为trueparamMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);// TODO 设置自动提交offset的时间周期为1000ms,默认5000msparamMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");// TODO 通过配置,创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);// TODO 订阅主题consumer.subscribe(Arrays.asList("test"));while ( true ) {// TODO 消费数据final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));// TODO 遍历数据for (ConsumerRecord<String, String> record : poll) {System.out.println( record );}}}
}

5.3.2、手动提交

  基于时间周期的偏移量提交,是我们无法控制的,一旦参数设置的不合理,或单位时间内数据量消费的很多,却没有来及的自动提交,那么数据就会重复消费。所以Kafka也支持消费偏移量的手动提交,也就是说当消费者消费完数据后,自行通过API进行提交。不过为了考虑效率和安全,kafka同时提供了异步提交和同步提交两种方式供我们选择。注意:需要禁用自动提交ENABLE_AUTO_COMMIT_CONFIG=false,才能开启手动提交:

1、异步提交

  向Kafka发送偏移量offset提交请求后,就可以直接消费下一批数据,因为无需等待kafka的提交确认,所以无法知道当前的偏移量一定提交成功,安全性比较低,但相对,消费性能会提高

public class KafkaConsumerCommitASyncTest {public static void main(String[] args) {// TODO 创建消费者配置参数集合Map<String, Object> paramMap = new HashMap<>();paramMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// TODO 禁用自动提交消费偏移量,默认取值为trueparamMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);paramMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());paramMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test");// TODO 通过配置,创建消费者对象KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(paramMap);// TODO 订阅主题consumer.subscribe(Arrays.asList("test"));while ( true ) {// TODO 消费数据final ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(100));// TODO 遍历处理数据for (ConsumerRecord<String, String> record : poll) {System.out.println( record );}// TODO 异步提交偏移量//     此处需要注意,需要在拉取数据完成处理后再提交//     否则提前提交了,但数据处理失败,下一次消费数据就拉取不到了consumer.commitAsync();}}
}

2、同步提交

  必须等待Kafka完成offset提交请求的响应后,才可以消费下一批数据,一旦提交失败,会进行重试处理,尽可能保证偏移量提交成功,但是依然可能因为意外情况导致提交请求失败。此种方式消费效率比较低,但是安全性高。

5.4、偏移量的保存

  由于消费者在消费消息的时候可能会由于各种原因而断开消费,当重新启动消费者时我们需要让它接着上次消费的位置offset继续消费,因此消费者需要实时的记录自己以及消费的位置。0.90版本之前,这个信息是记录在zookeeper内的,在0.90之后的版本,offset保存在_consumer_offsets这个topic内。

  每个consumer会定期将自己消费分区的offset提交给_consumer_offsets,提交过去的时候,key是consumerGroupId+topic+分区号

在这里插入图片描述

六、消费者事务

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com