网赌赢了钱被网站黑了需要怎么做,怎么用html做个人的网页,建设礼品网站的策划书,杭州 做网站目录 5.1 Kafka的消费方式5.2 Kafka 消费者工作流程1、总体流程2、消费者组原理3、消费者组初始化流程4、消费者组详细消费流程 5.3 消费者API1 独立消费者案例#xff08;订阅主题#xff09;2 独立消费者案例#xff08;订阅分区#xff09;3 消费者组案例 5.4 生产经验—… 目录 5.1 Kafka的消费方式5.2 Kafka 消费者工作流程1、总体流程2、消费者组原理3、消费者组初始化流程4、消费者组详细消费流程 5.3 消费者API1 独立消费者案例订阅主题2 独立消费者案例订阅分区3 消费者组案例 5.4 生产经验——分区的分配以及再平衡1、 Range 以及再平衡1Range 分区策略原理2Range 分区分配策略demo演示3Range 分区分配再平衡案例 2 RoundRobin 以及再平衡1RoundRobin 分区策略原理2RoundRobin 分区分配策略demo3RoundRobin 分区分配再平衡案例 3 Sticky 以及再平衡1) 定义2 Sticky 分区策略demo演示3Sticky 分区分配再平衡 5.5 offset位移1、offset 的默认维护位置1__consumer_offsets 查看 2、自动提交 offset1消费者自动提交 offset 3、手动提交 offset1同步提交 offset2异步提交 offset 4、指定Offset进行消费5、指定时间进行消费6 、漏消费和重复消费7 生产经验——数据积压 5.1 Kafka的消费方式
pull拉模 式consumer采用从broker中主动拉取数据。Kafka采用这种方式。
缺点 pull模式不足之处是如 果Kafka没有数据消费者可能会陷入循环中一直返回空数据
push推模式Kafka没有采用这种方式因为由broker决定消息发送速率很难适应所有消费者的消费速率 5.2 Kafka 消费者工作流程
1、总体流程
【注意】
消费者只能从主分区上拉取数据从节点起到同步和冗余数据的作用每个分区的数据只能由消费者组中一个消费者消费一个消费者可以消费多个分区数据每个消费者的offset由消费者提交到系统主题保存
2、消费者组原理
Consumer GroupCG消费者组由多个consumer组成。形成一个消费者组的条件是所有消费者的groupid相同。
消费者组内每个消费者负责消费不同分区的数据一个分区只能由一个组内消费者消费。消费者组之间互不影响。所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者
3、消费者组初始化流程
4、消费者组详细消费流程 5.3 消费者API
1 独立消费者案例订阅主题
public class CustomConsumer {public static void main(String[] args) {// 0 配置Properties properties new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.239.11:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,test5);// 设置分区分配策略// 1 创建一个消费者 , helloKafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题 firstArrayListString topics new ArrayList();topics.add(first);kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}kafkaConsumer.commitAsync();}}
}2 独立消费者案例订阅分区
public class CustomConsumerPartition {public static void main(String[] args) {// 0 配置Properties properties new Properties();// 连接properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,hadoop102:9092,hadoop103:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,test);// 1 创建一个消费者KafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题对应的分区ArrayListTopicPartition topicPartitions new ArrayList();topicPartitions.add(new TopicPartition(first,0));kafkaConsumer.assign(topicPartitions);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}
3 消费者组案例
1需求测试同一个主题的分区数据只能由一个消费者组中的一个消费 5.4 生产经验——分区的分配以及再平衡
1、 Range 以及再平衡
1Range 分区策略原理 【缺点】 容易产生数据倾斜 2Range 分区分配策略demo演示
①、创建7个分区的topic ②、启动 CustomProducer 生产者发送7条消息到 0 - 6号分区
public class CustomProducerCallback {public static void main(String[] args) throws InterruptedException {// 0 配置Properties properties new Properties();// 连接集群 bootstrap.serversproperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 192.168.239.11:9092);// 指定对应的key和value的序列化类型 key.serializer
// properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringSerializer);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 1 创建kafka生产者对象KafkaProducerString, String kafkaProducer new KafkaProducer(properties);// 2 发送数据for (int i 0; i 7; i) {kafkaProducer.send(new ProducerRecord(test, i, i , houchen i), new Callback() {Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception null) {System.out.println(主题 metadata.topic() 分区 metadata.partition());}}});Thread.sleep(2);}// 3 关闭资源kafkaProducer.close();}
}③、启动三个消费者组成一个消费者组查看各个消费者的消费情况
由下述结果确实可以看到 Kafka 默认的分区分配策略就是 Range
public class CustomConsumer {public static void main(String[] args) {// 0 配置Properties properties new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.239.11:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,mygroup);// 1 创建一个消费者 KafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题 firstArrayListString topics new ArrayList();topics.add(test);kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}kafkaConsumer.commitAsync();}}
}3Range 分区分配再平衡案例
1停止掉 0 号消费者快速重新发送消息观看结果45s 以内越快越好。 1 号消费者消费到 3、4 号分区数据。 2 号消费者消费到 5、6 号分区数据。
0 号消费者的任务会整体被分配到 1 号消费者或者 2 号消费者。 说明0 号消费者挂掉后消费者组需要按照超时时间 45s 来判断它是否退出所以需要等待时间到了 45s 后判断它真的退出就会把任务分配给其他 broker 执行。
2再次重新发送消息观看结果45s 以后。 1 号消费者消费到 0、1、2、3 号分区数据。 2 号消费者消费到 4、5、6 号分区数据。 说明消费者 0 已经被踢出消费者组所以重新按照 range 方式分配。 2 RoundRobin 以及再平衡
1RoundRobin 分区策略原理 2RoundRobin 分区分配策略demo
①、依次在 CustomConsumer、CustomConsumer1、CustomConsumer2 三个消费者代 码中修改分区分配策略为 RoundRobin //RoundRobin 分区分配策略properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,org.apache.kafka.clients.consumer.RoundRobinAssignor);②、重启 3 个消费者重复发送消息的步骤观看分区结果 3RoundRobin 分区分配再平衡案例
停止掉 0 号消费者快速重新发送消息观看结果45s 以内越快越好。 1 号消费者消费到 2、5 号分区数据 2 号消费者消费到 4、1 号分区数据 0 号消费者的任务会按照 RoundRobin 的方式把数据轮询分成 0 和6 、 3 号分区数据分别由 1 号消费者或者 2 号消费者消费。
说明0 号消费者挂掉后消费者组需要按照超时时间 45s 来判断它是否退出所以需要等待时间到了 45s 后判断它真的退出就会把任务分配给其他 broker 执行 2再次重新发送消息观看结果45s 以后。 1 号消费者消费到 0、2、4、6 号分区数据 2 号消费者消费到 1、3、5 号分区数据 说明消费者 0 已经被踢出消费者组所以重新按照 RoundRobin 方式分配。 3 Sticky 以及再平衡
1) 定义
粘性分区定义可以理解为分配的结果带有“粘性的”。即在执行一次新的分配之前考虑上一次分配的结果尽量少的调整分配的变动可以节省大量的开销
2 Sticky 分区策略demo演示
3Sticky 分区分配再平衡 5.5 offset位移
1、offset 的默认维护位置 __consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.idtopic分区号value 就是当前 offset 的值。每隔一段时间kafka 内部会对这个 topic 进行compact也就是每个 group.idtopic分区号就保留最新数据 1__consumer_offsets 查看
2、自动提交 offset
为了使我们能够专注于自己的业务逻辑Kafka提供了自动提交offset的功能。
自动提交offset的相关参数
enable.auto.commit是否开启自动提交offset功能默认是trueauto.commit.interval.ms自动提交offset的时间间隔默认是5s 1消费者自动提交 offset
// 自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);3、手动提交 offset
虽然自动提交offset十分简单便利但由于其是基于时间提交的开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API
手动提交offset的方法有两种分别是commitSync同步提交和commitAsync异步提交。 两者的相同点是都会将本次提交的一批数据最高的偏移量提交 不同点是同步提交阻塞当前线程一直到提交成功并且会自动失败重试由不可控因素导致也会出现提交失败而异步提交则没有失败重试机制故有可能提交失败。
commitSync同步提交必须等待offset提交完毕再去消费下一批数据。commitAsync异步提交 发送完提交offset请求后就开始消费下一批数据了。 1同步提交 offset
由于同步提交 offset 有失败重试机制故更加可靠但是由于一直等待提交结果提交的效率比较低。以下为同步提交 offset 的示例。
public class CustomConsumerByHandSync {public static void main(String[] args) {// 0 配置Properties properties new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.239.11:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,test);// 手动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 1 创建一个消费者 , helloKafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题 firstArrayListString topics new ArrayList();topics.add(first);kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}// 手动提交offsetkafkaConsumer.commitSync();}}
} 2异步提交 offset
虽然同步提交 offset 更可靠一些但是由于其会阻塞当前线程直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下会选用异步提交 offset 的方式。
public class CustomConsumerByHandSync {public static void main(String[] args) {// 0 配置Properties properties new Properties();// 连接 bootstrap.serversproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.239.11:9092);// 反序列化properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());// 配置消费者组idproperties.put(ConsumerConfig.GROUP_ID_CONFIG,test);// 手动提交properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);// 1 创建一个消费者 , helloKafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题 firstArrayListString topics new ArrayList();topics.add(first);kafkaConsumer.subscribe(topics);// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}// 手动提交offsetkafkaConsumer.commitAsync();}}
} 4、指定Offset进行消费
public class CustomConsumerSeek {public static void main(String[] args) {// 0 配置信息Properties properties new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.239.11:9092);properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());properties.put(ConsumerConfig.GROUP_ID_CONFIG,test3);// 1 创建消费者KafkaConsumerString, String kafkaConsumer new KafkaConsumer(properties);// 2 订阅主题ArrayListString topics new ArrayList();topics.add(second);kafkaConsumer.subscribe(topics);// 指定位置进行消费SetTopicPartition assignment kafkaConsumer.assignment();// 保证分区分配方案已经制定完毕while (assignment.size() 0){kafkaConsumer.poll(Duration.ofSeconds(1));assignment kafkaConsumer.assignment();}// 指定消费的offsetfor (TopicPartition topicPartition : assignment) {kafkaConsumer.seek(topicPartition,100);}// 3 消费数据while (true){ConsumerRecordsString, String consumerRecords kafkaConsumer.poll(Duration.ofSeconds(1));for (ConsumerRecordString, String consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}5、指定时间进行消费
需求在生产环境中会遇到最近消费的几个小时数据异常想重新按照时间消费。 例如要求按照时间消费前一天的数据怎么处理
6 、漏消费和重复消费
7 生产经验——数据积压
