kafka新版API

Kafka 2022-09-04
原文链接：https://blog.csdn.net/Simon_09010817/article/details/83750115

接上篇 Kafka新版消费者API示例（一）https://blog.csdn.net/Simon_09010817/article/details/83748974

kafka手动提交策略提供了更加灵活的管理方式，在某些场景我们需要对消费偏移量有更精准的管理。以保证消息不被重复消费以及消息不丢失。

Kafka提供两种手动提交方式：

1.异步提交(commitAsync)：

   异步模式下，提交失败也不会尝试提交。消费者线程不会被阻塞，因为异步操作，可能在提交偏移量操作结果未返回时就开始下一次拉取操作。

2.同步提交(CommitSync)：

    同步模式下，提交失败时一直尝试提交，直到遇到无法重试才结束。同步方式下，消费者线程在拉取消息时会被阻塞，直到偏移量提交操作成功或者在提交过程中发生错误。

实现手动提交前需要在创建消费者时关闭自动提交，设置enable.auto.commit=false。

由于异步提交不会等消费偏移量提交成功后再拉取下一次消息，因此异步提交提供了一个偏移量提交回调方法commitAsync(OffsetCommitCallback callback)。提交偏移量完成之后会回调OffsetCommitCallback接口的onComplete()方法

示例代码：

  1. package com.simon.kafka.consumer.newconsumer;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5.  
  6. import java.util.*;
  7.  
  8. /**
  9.  * Created by Simon on 2018/11/5.
  10.  */
  11. public class KafkaConsumerAsync {
  12.  
  13.     public static void main(String[] args) throws InterruptedException {
  14.  
  15.         // 1、准备配置文件
  16.         String kafkas = "192.168.1.100:9092,192.168.1.100:9093,192.168.1.100:9094";
  17.         Properties props = new Properties();
  18.         //kafka连接信息
  19.         props.put("bootstrap.servers",kafkas);
  20.         //消费者组id
  21.         props.put("group.id", "test_group");
  22.         //是否自动提交offset
  23.         props.put("enable.auto.commit", "false");
  24.         //在没有offset的情况下采取的拉取策略
  25.         props.put("auto.offset.reset", "none");
  26.         //自动提交时间间隔
  27.         props.put("auto.commit.interval.ms", "1000");
  28.         //设置一次fetch请求取得的数据最大为1k
  29.         props.put("fetch.max.bytes", "1024");
  30.         //key反序列化
  31.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  32.         //value反序列化
  33.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  34.  
  35.         String topic = "test";
  36.         // 2、创建KafkaConsumer
  37.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  38.         // 3、订阅数据，不给定监听器
  39.         consumer.subscribe(Collections.singleton(topic));
  40.  
  41.         try{
  42.             //最少处理100条
  43.             int minCommitSize = 100;
  44.             //定义计数器
  45.             int icount = 0;
  46.             // 4、获取数据
  47.             while (true) {
  48.                 ConsumerRecords<String, String> records = consumer.poll(100);
  49.                 for (ConsumerRecord<String, String> record : records) {
  50.                     System.out.printf("topic = %s,partition = %d,offset = %d, key = %s, value = %s%n", record.topic(), record.partition(),record.offset(), record.key(), record.value());
  51.                     icount++;
  52.                 }
  53.                 Thread.sleep(5000);
  54.             
  55.                 //在业务逻辑处理成功后提交offset
  56.                 if(icount >= minCommitSize){
  57.                     //满足最少消费100条,再进行异步提交
  58.                     consumer.commitAsync(new OffsetCommitCallback() {
  59.                         @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
  60.                             if(exception == null){
  61.                                 System.out.println("commit success");
  62.                             }else {
  63.                                 //提交失败，对应处理
  64.                                 System.out.println("commit Failed");
  65.                             }
  66.                         }
  67.                     });
  68.                     
  69.                     //计数器归零
  70.                     icount = 0 ;
  71.                 }
  72.             }
  73.         }catch (Exception e){
  74.             e.printstacktrace();
  75.         }finally {
  76.             //关闭连接
  77.             consumer.close();
  78.         }
  79.     }
  80. }

以时间戳查询消息：

    Kafka在0.10.1.1版本上增加了时间戳索引文件。Kafka消费者API提供了一个offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch)方法，参数为一个map对象，key为待查询的分区，value为待查询的时间戳。会返回一个大于等于该事件戳的第一条消息对应的偏移量和时间戳。若待查询分区不存在，会一直阻塞。

 示例：

   将kafka-client的maven依赖改为1.0.0 。在0.10.0.1中无法引入OffsetAndTimestamp类

  1.  <!--引入kafka-clients-->
  2.     <!--<dependency>
  3.       <groupId>org.apache.kafka</groupId>
  4.       <artifactId>kafka-clients</artifactId>
  5.       <version>0.10.0.1</version>
  6.     </dependency>-->
  7.  
  8.     <dependency>
  9.       <groupId>org.apache.kafka</groupId>
  10.       <artifactId>kafka-clients</artifactId>
  11.       <version>1.0.0</version>
  12.     </dependency>

代码：

  1. package com.simon.kafka.consumer.newconsumer;
  2.  
  3. import org.apache.kafka.clients.consumer.*;
  4. import org.apache.kafka.common.TopicPartition;
  5.  
  6. import java.util.Collections;
  7. import java.util.HashMap;
  8. import java.util.Map;
  9. import java.util.Properties;
  10.  
  11. /**
  12.  * Created by Simon on 2018/11/5.
  13.  */
  14. public class KafkaConsumerTimestamps {
  15.  
  16.     public static void main(String[] args) throws InterruptedException {
  17.  
  18.         // 1、准备配置文件
  19.         String kafkas = "192.168.1.100:9092,192.168.1.100:9093,192.168.1.100:9094";
  20.         Properties props = new Properties();
  21.         //kafka连接信息
  22.         props.put("bootstrap.servers",kafkas);
  23.         //消费者组id
  24.         props.put("group.id", "test_group");
  25.         //客户端id
  26.         props.put("client.id", "test_group");
  27.         //是否自动提交offset
  28.         props.put("enable.auto.commit", "true");
  29.         //在没有offset的情况下采取的拉取策略
  30.         props.put("auto.offset.reset", "none");
  31.         //自动提交时间间隔
  32.         props.put("auto.commit.interval.ms", "1000");
  33.         //设置一次fetch请求取得的数据最大为1k
  34.         props.put("fetch.max.bytes", "1024");
  35.         //key反序列化
  36.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  37.         //value反序列化
  38.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  39.  
  40.         String topic = "test";
  41.         // 2、创建KafkaConsumer
  42.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  43.         // 3、订阅主题
  44.         TopicPartition topicPartition = new TopicPartition(topic,0);
  45.         consumer.assign(Collections.singleton(topicPartition));
  46.         try{
  47.             Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
  48.  
  49.             // 设置查询12 小时之前消息的偏移量
  50.             timestampsToSearch.put(topicPartition, (System.currentTimeMillis() - 12 * 3600 * 1000));
  51.  
  52.             // 会返回时间大于等于查找时间的第一个偏移量
  53.             Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(timestampsToSearch);
  54.             OffsetAndTimestamp offsetTimestamp = null;
  55.             // 用for 轮询，当然由于本例是查询的一个分区，因此也可以用if 处理
  56.             for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetMap.entrySet()) {
  57.                 // 若查询时间大于时间戳索引文件中最大记录索引时间，
  58.                 // 此时value 为空,即待查询时间点之后没有新消息生成
  59.                 offsetTimestamp = entry.getValue();
  60.                 if (null != offsetTimestamp) {
  61.                     // 重置消费起始偏移量
  62.                     consumer.seek(topicPartition, entry.getValue().offset());
  63.                 }
  64.             }
  65.             while (true) {
  66.                 //4.轮询拉取消息
  67.                 ConsumerRecords<String, String> records = consumer.poll(100);
  68.                 for (ConsumerRecord<String, String> record : records) {
  69.                     System.out.printf("topic = %s,partition = %d,offset = %d, key = %s, value = %s%n", record.topic(), record.partition(),record.offset(), record.key(), record.value());
  70.  
  71.                 }
  72.             }
  73.         }catch (Exception e){
  74.             e.printstacktrace();
  75.         }finally {
  76.             //关闭连接
  77.             consumer.close();
  78.         }
  79.     }
  80. }

由于集群环境已选型为kafka0.10.0.1，本次无法按指定时间戳拉取，报错信息为不支持当前broker版本。

速度控制：

   应用场景中我们可能需要暂停某些分区消费，先消费其他分区，当达到某个条件再恢复该分区消费。

Kafka提供两种方法用于速度控制的方法：

     1.pause(Collection<TopicPartition> partitions)：暂停某些分区在拉取操作时返回数据给客户端

  1. //无返回值
  2. consumer.pause(Collections.singleton(topicPartition));

     2.resume(Collection<TopicPartition> partitions)：恢复某些分区向客户端返回数据

  1. //无返回值
  2. consumer.resume(Collections.singleton(topicPartition));

 

相关文章

kafka分区分配策略
# 前言 现有主流消息中间件都是生产者-消费者模型，主要角色...
Kafka报错：Exception in thread “main“ joptsimple.UnrecognizedOptionException: zookeeper is not a recogn
错误的根源是：kafka版本过高所致，2.2+=的版本，已经不需要...
湖仓一体电商项目十九：业务实现之编写写入DWS层业务代码
DWS层主要是存放大宽表数据，此业务中主要是针对Kafka topic...
Springboot 配置使用 Kafka
不多BB讲原理，只教你怎么用，看了全网没有比我更详细的了，...
【Kafka从成神到升仙系列 五】面试官问我 Kafka 生产者的网络架构，我直接开始从源码背起.......
终于写完了，其实最开始学kafka的时候是今年2月份，那时候还...
Greenplum GPKafka【实践 01】使用GPKafka实现Kafka数据导入Greenplum数据库踩坑问题记录不断更新ing
使用GPKafka实现Kafka数据导入Greenplum数据库踩坑问题记录（...