Kafka AdminClient 管理Kafka Offset代码实现

标签: kafka offset  offset管理  kafka offset 代码实现  AdminClient  kafka 获取offset

一:offset保存位置

     kafka的offset管理,可以交给zookeeper实现,经过配置offsets.storage=zookeeper;也可以直接跟kafka自己管理,在你 的配置文件设置下面的配置项;

offsets.storage=kafka 

dual.commit.enabled=true 

kafka自己管理自己的offset,并经过命令可以查看

bin/kafka-consumer-groups.sh --bootstrap-server broker1:9092 --describe --group kafkatest


 如上图可以查看到group中每个消费者对应topic的patition,当前消费的offset;每个partition中最大的offset等信息;经过测试,当没有消费者进行消费时候,若执行上面命令会出现下面的现象:消费者不见了


以上就是用kafka命令获取的结果;

下面是经过查看借鉴源码,用Java API实现的offset管理(kafka 版本为0.10)

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.10</artifactId>
    <version>0.10.2.1</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.1</version>
</dependency>

测试类代码如下:

public static void main(String[] args) throws InstantiationException, IllegalAccessException {
	  Properties props = new Properties();
	  props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
	  AdminClient adminClient = AdminClient.create(props);
	  ConsumerGroupSummary  consumerGroupSummary =  adminClient.describeConsumerGroup("kafkatest");
	  if(consumerGroupSummary.state().equals("Empty")){
		  System.out.println("niaho");
	  }
	  Option<List<ConsumerSummary>> consumerSummaryOption =  consumerGroupSummary.consumers();

	  List<ConsumerSummary> ConsumerSummarys = consumerSummaryOption.get();//获取组中的消费者
	  KafkaConsumer consumer = getNewConsumer();
	  for(int i=0;i<ConsumerSummarys.size();i++){ //循环组中的每一个消费者
		  
		  ConsumerSummary consumerSummary = ConsumerSummarys.apply(i);
		  String consumerId  = consumerSummary.consumerId();//获取消费者的id
		  scala.collection.immutable.Map<TopicPartition, Object> maps = 
				  adminClient.listGroupOffsets("kafkatest");//或者这个组消费的所有topic,partition和当前消费到的offset
		  List<TopicPartition> topicPartitions= consumerSummary.assignment();//获取这个消费者下面的所有topic和partion
		  for(int j =0;j< topicPartitions.size();j++){ //循环获取每一个topic和partion
			  TopicPartition topicPartition = topicPartitions.apply(j);
			  String CURRENToFFSET = maps.get(topicPartition).get().toString();
			  long endOffset =getLogEndOffset(topicPartition);
			  System.out.println("topic的名字为:"+topicPartition.topic()+"====分区为:"+topicPartition.partition()+"===目前消费offset为:"+CURRENToFFSET+"===,此分区最后offset为:"+endOffset);
		  }
	  }
}
引用的两个方法代码如下:

获取消费者

public static KafkaConsumer getNewConsumer(){
	     Properties props = new Properties();
	     props.put("bootstrap.servers", "localhost:9092");
	     props.put("group.id", "kafkatest");
	     props.put("enable.auto.commit", "true");
	     props.put("auto.offset.reset", "earliest");
	     props.put("auto.commit.interval.ms", "1000");
	     props.put("auto.commit.interval.ms", "1000");
	     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
	     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
	     return consumer;
  }
获取最后的offset:

public static long getLogEndOffset(TopicPartition topicPartition){
	  KafkaConsumer<String, String> consumer= getNewConsumer();
	  consumer.assign(Arrays.asList(topicPartition));
	  consumer.seekToEnd(Arrays.asList(topicPartition));
	  long endOffset = consumer.position(topicPartition);
	  return endOffset;
  }
运行后,可以获取结果如下:





版权声明:本文为lisi1129原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/lisi1129/article/details/72869194

智能推荐

springboot中实现kafka指定offset消费

kafka消费过程难免会遇到需要重新消费的场景,例如我们消费到kafka数据之后需要进行存库操作,若某一时刻数据库down了,导致kafka消费的数据无法入库,为了弥补数据库down期间的数据损失,有一种做法我们可以指定kafka消费者的offset到之前某一时间的数值,然后重新进行消费。 首先创建kafka消费服务 kafka消费模型建立 kafka server中每个主题存在多个分区(part...

kafka管理offset方式之使用外部存储保存offset

1、Kafka Offset 管理–Checkpoint 启用Spark Streaming的checkpoint是存储偏移量最简单的方法。 流式checkpoint专门用于保存应用程序的状态, 比如保存在HDFS上, 在故障时能恢复。 Spark Streaming的checkpoint无法跨越应用程序进行恢复。 Spark 升级也将导致无法恢复。 在关键生产应用, 不建议使用spa...

Spark Streaming消费Kafka并手动使用Redis管理Kafka Offset

1 Spark Streaming读取Kafka的两种模式 Spark Streaming消费Kafka的数据有两种模式:Receiver和Direct模式,学习时候重点关注下Direct即可,因为在最新读取方式中已经不支持Receiver。 1.1 Receiver模式 在Spark 1.3之前,Spark Streaming消费Kafka中的数据采用基于Kafka高级消费API实现的Recei...

Kafka+Spark Streaming管理offset的几种方法

来源:大数据技术与架构 作者:王知无 大数据技术与架构 点击右侧关注,大数据开发领域最强公众号! 暴走大数据 点击右侧关注,暴走大数据! By 大数据技术与架构 场景描述:Kafka配合Spark Streaming是大数据领域常见的黄金搭档之一,主要是用于数据实时入库或分析。为了应对可能出现的引起Streaming程序崩溃的异常情况,我们一般都需要手动管理好Kafka的offset,而...

【kafka】Kafka管理与监控

1.美图 2. JVM参数配置优化 如果使用的CMS GC算法,建议JVM Heap不要太大,在4GB以内就可以。JVM太大,导致Major GC或者Full GC产生的“stop the world”时间过长,导致broker和zk之间的session超时,比如重新选举controller节点和提升follow replica为leader replica。 JVM也不能...

猜你喜欢

冒泡排序,及改进方式,性能优化400%>>>附图解加源码

首先源码附上,源码中带有注释,看不懂没关系,源码后面附带图解,最后附上代码效率提升图 源码如下: 方案一:其实实现很简单,两层循环,每次内层迭代出最大的一个值,将其放入数组最后一个位置,外层循环的末端便往前移一位。其原理如下图 方案一代码块: 方案二:优化改进 仔细观察上面的图,我们不难发现当迭代到图下这样的情况时,其实已经全排序好了,但是我们还是需要对它进行1,2,3,4,5,6的迭代,这些情况...

css3常用选择器

先介绍一下基本选择器,如下几种 *:通配符。选择页面的所有元素 E:元素选择器。对页面的一些元素进行选择,如p,div,li等 .class:类选择器 #id:id选择器。一个Id在一个页面中只能选择,这是和类选择器的区别 E F:后代选择器。如ul li。选中ul下的li E>F:子元素选择器。如下,可以用 h1 > strong {color:red;}使h1的子元素strong变...

python is not set from command line or npm configuration

问题 猜测原因  python版本冲突 解决方案  去官网找到最新版本python 点击下载,注意记住你下载的目录,下载好以后,打开这个目录下的文件夹,Python310, 整个文件夹复制一下 在项目里运行npm config list --json 找到python的路径   我的是 C:\\Python27\\python.exe 说明项目里用的还是py...

Java8 Stream流操作

前言 我们常常需要将一个容器转化成另一个容器,或是对这个容器中的数据进行批量处理,这时使用Stream流可以大大减少我们的工作量。 1 Stream概述 Java 8 是一个非常成功的版本,这个版本新增的Stream,配合同版本出现的 Lambda ,给我们操作集合(Collection)提供了极大的便利。 那么什么是Stream? Stream将要处理的元素集合看作一种流,在流的过程中,借助St...

带你玩转Visual Studio——带你跳出坑爹的Runtime Library坑

在Windows下进行C++的开发,不可避免的要与Windows的底层库进行交互,然而VS下的一项设置MT、MTd、MD和MDd却经常让人搞迷糊,相信不少人都被他坑过,特别是你工程使用了很多第三库的时候,及容易出现各种链接问题。看一下下面这个错误提示:  LIBCMT.lib(_file.obj) : error LNK2005: ___initstdio already defined...