Kafka旧版消费者API示例（低级消费者）

2023-12-17 08:32
文章标签 api 示例 kafka 消费者 低级 旧版

本文主要是介绍Kafka旧版消费者API示例（低级消费者），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Kafka旧版高级消费者API强依赖ZK，目前高版本kafka已将offset移交内部主题，若高版本可选用新版API。

实现低级API变成实现的主要步骤：

  • 获取指定主题相应分区对应的元数据信息

  • 找出指定分区的Leader副本节点，创建SimpleConsumer，建立与Leader副本的连接

  • 构造消费请求

  • 获取数据并处理

  • 对偏移量进行处理

  • 当代理发生变化时进行相应处理，保证消息被正常消费

1.创建一个类并定义部分常量：

public class LowConsumerAPI {/*** broker list*/private static final String BROKER_LIST = "192.168.1.100:9092,192.168.1.100:9093,192.168.1.100:9094";/*** 连接超时时间：1min*/private static final int TIME_OUT = 60 * 1000;/*** 读取消息缓存区大小：1M*/private static final int BUFFER_SIZE = 1024 * 1024;/*** 每次获取消息的条数*/private static final int FETCH_SIZE = 100000;/*** 发生错误时重试的次数*///private static final int RETRIES_TIME = 3;/*** 允许发生错误的最大次数*/private static final int MAX_ERROR_NUM = 3;

2.定义获取主题相应分区元数据信息的方法：

/*** 获取指定主题指定分区的元数据*/private PartitionMetadata fetchPartitionMetadata(List<String> brokerList, String topic, int partitionId) {SimpleConsumer consumer = null;TopicMetadataRequest metadataRequest = null;TopicMetadataResponse metadataResponse = null;List<TopicMetadata> topicMetadatas = null;try{/** 循环是因为不确定传入的partition的leader节点是哪个*/for(String host : brokerList) {// 1. 构建一个消费者SimpleConsumer，它是获取元数据的执行者String[] hostsAndPorts = host.split(":");// 最后一个参数是 clientIdconsumer = new SimpleConsumer(hostsAndPorts[0], Integer.parseInt(hostsAndPorts[1]),TIME_OUT, BUFFER_SIZE, topic + "-" + "0" + "-" + "client");// 2. 构造请求主题元数据信息的请求 TopicMetadateRequestmetadataRequest = new TopicMetadataRequest(Arrays.asList(topic));// 3. 通过send()正式与代理通信，发送TopicMetadateRequest请求获取元数据try {metadataResponse = consumer.send(metadataRequest);} catch (Exception e) {//有可能与代理失去连接System.out.println("get TopicMetadataResponse failed!");e.printStackTrace();continue;}// 4. 获取主题元数据TopicMetadata列表，每个主题的每个分区的元数据信息对应一个TopicMetadata对象topicMetadatas = metadataResponse.topicsMetadata();// 5. 遍历主题元数据信息列表for(TopicMetadata topicMetadata : topicMetadatas) {//6. 获取当前分区对应元数据信息PartitionMetadata    for(PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) {if(partitionMetadata.partitionId() != partitionId) {continue;} else {return partitionMetadata;}}}}} catch (Exception e) {System.out.println("Fetch PartitionMetadata failed!");e.printStackTrace();} finally {if(consumer != null) {consumer.close();}}return null;}

3.根据分区元数据信息找出指定分区的Leader节点：

   /*** 根据分区的元数据信息获取它的leader节点*/private String getLeader(PartitionMetadata metadata) {if(metadata.leader() == null) {System.out.println("can not find partition" + metadata.partitionId() + "'s leader!");return null;}return metadata.leader().host()+"_"+metadata.leader().port();}

4.对偏移量进行管理：

/*** 获取指定主题指定分区的消费偏移量*/private long getOffset(SimpleConsumer consumer, String topic, int partition, long beginTime, String clientName) {TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<>();/** PartitionOffsetRequestInfo(beginTime, 1)用于配置获取offset的策略* beginTime有两个值可以取*     kafka.api.OffsetRequest.EarliestTime()，获取最开始的消费偏移量，不一定是0，因为segment会删除*     kafka.api.OffsetRequest.LatestTime()，获取最新的消费偏移量*/requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(beginTime, 1));// 构造获取offset的请求OffsetRequest request = new OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);OffsetResponse response = consumer.getOffsetsBefore(request);if(response.hasError()) {System.out.println("get offset failed!" + response.errorCode(topic, partition));return -1;}long[] offsets = response.offsets(topic, partition);if(offsets == null || offsets.length == 0) {System.out.println("get offset failed! offsets is null");return -1;}return offsets[0];}

5.当代理发生变化时，做出相应变化：

 /*** 重新寻找partition的leader节点的方法*/private String findNewLeader(List<String> brokerList, String oldLeader, String topic, int partition) throws Exception {for (int i = 0; i < 3; i++) {boolean goToSleep = false;PartitionMetadata metadata = fetchPartitionMetadata(brokerList, topic, partition);if (metadata == null) {goToSleep = true;} else if (metadata.leader() == null) {goToSleep = true;} else if (oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {// 这里考虑到 zookeeper 还没有来得及重新选举 leader 或者在故障转移之前挂掉的 leader 又重新连接的情况goToSleep = true;} else {return metadata.leader().host();}if (goToSleep) {Thread.sleep(1000);}}System.out.println("Unable to find new leader after Broker failure!");throw new Exception("Unable to find new leader after Broker failure!");}

6.定义consume方法，获取数据并处理：

/*** 处理数据的方法*/public void consume(List<String> brokerList, String topic, int partition) {SimpleConsumer consumer = null;try {// 1. 获取分区元数据信息PartitionMetadata metadata = fetchPartitionMetadata(brokerList,topic, partition);if(metadata == null) {System.out.println("can not find metadata!");return;}// 2. 找到分区的leader节点String leaderBrokerAndPort = getLeader(metadata);String[] brokersAndPorts = leaderBrokerAndPort.split("_");String leaderBroker = brokersAndPorts[0];int  port = Integer.parseInt(brokersAndPorts[1]);String clientId = topic + "-" + partition + "-" + "client";// 3. 创建一个消费者用于消费消息consumer = new SimpleConsumer(leaderBroker ,port ,TIME_OUT, BUFFER_SIZE, clientId);// 4. 配置获取offset的策略为，获取分区最开始的消费偏移量long offset = getOffset(consumer, topic, partition, kafka.api.OffsetRequest.EarliestTime(), clientId);int errorCount = 0;kafka.api.FetchRequest request = null;kafka.javaapi.FetchResponse response = null;while(offset > -1) {// 运行过程中，可能因为处理错误，把consumer置为 null，所以这里需要再实例化if(consumer == null) {consumer = new SimpleConsumer(leaderBroker ,port , TIME_OUT, BUFFER_SIZE, clientId);}// 5. 构建获取消息的requestrequest = new FetchRequestBuilder().clientId(clientId).addFetch(topic, partition, offset, FETCH_SIZE).build();// 6. 获取响应并处理response = consumer.fetch(request);if(response.hasError()) {errorCount ++;if(errorCount > MAX_ERROR_NUM) {break;}short errorCode = response.errorCode(topic, partition);if(ErrorMapping.OffsetOutOfRangeCode() == errorCode) {// 如果是因为获取到的偏移量无效，那么应该重新获取// 这里简单处理，改为获取最新的消费偏移量offset = getOffset(consumer, topic, partition, kafka.api.OffsetRequest.LatestTime(), clientId);continue;} else if (ErrorMapping.OffsetsLoadInProgressCode() == errorCode) {Thread.sleep(300000);continue;} else {consumer.close();consumer = null;// 更新leader brokerleaderBroker = findNewLeader(brokerList, leaderBroker, topic, partition);continue;}// 如果没有错误} else {// 清空错误记录errorCount = 0;long fetchCount = 0;// 处理消息for(MessageAndOffset messageAndOffset : response.messageSet(topic, partition)) {long currentOffset = messageAndOffset.offset();if(currentOffset < offset) {System.out.println("get an old offset[" + currentOffset + "], excepted offset is offset[" + offset + "]");continue;}offset = messageAndOffset.nextOffset();ByteBuffer payload = messageAndOffset.message().payload();byte[] bytes = new byte[payload.limit()];payload.get(bytes);// 把消息打印到控制台System.out.println("message: " + new String(bytes, "UTF-8") + ", offset: " + messageAndOffset.offset());fetchCount++;}if (fetchCount == 0) {Thread.sleep(1000);}}}} catch (Exception e) {System.out.println("exception occurs when consume message");e.printStackTrace();} finally {if (consumer != null) {consumer.close();}}}

7.定义主函数启动：

public static void main(String[] args) {LowConsumerAPI lowConsumerAPI = new LowConsumerAPI();lowConsumerAPI.consume(Arrays.asList(BROKER_LIST.split(",")), "test", 0);}

 

这篇关于Kafka旧版消费者API示例（低级消费者）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/503745。

相关文章

搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
zeroclipboard 粘贴板的应用示例， 兼容 Chrome、IE等多浏览器

zeroclipboard 粘贴板的应用示例， 兼容 Chrome、IE等多浏览器

zeroclipboard单个复制按钮和多个复制按钮的实现方法 最近网站改版想让复制代码功能在多个浏览器上都可以实现，最近看网上不少说我们的代码复制功能不好用的，我们最近将会增加代码高亮等功能，希望大家多多支持我们 zeroclipboard是一个跨浏览器的库类 它利用 Flash 进行复制，所以只要浏览器装有 Flash 就可以运行，而且比 IE 的
阅读更多...
【LabVIEW学习篇 - 21】：DLL与API的调用

【LabVIEW学习篇 - 21】：DLL与API的调用

文章目录 DLL与API调用DLLAPIDLL的调用 DLL与API调用 LabVIEW虽然已经足够强大，但不同的语言在不同领域都有着自己的优势，为了强强联合，LabVIEW提供了强大的外部程序接口能力，包括DLL、CIN(C语言接口)、ActiveX、.NET、MATLAB等等。通过DLL可以使用户很方便地调用C、C++、C#、VB等编程语言写的程序以及windows自带的大
阅读更多...
如何更优雅地对接第三方API

如何更优雅地对接第三方API

如何更优雅地对接第三方API 本文所有示例完整代码地址：https://github.com/yu-linfeng/BlogRepositories/tree/master/repositories/third 我们在日常开发过程中，有不少场景会对接第三方的API，例如第三方账号登录，第三方服务等等。第三方服务会提供API或者SDK，我依稀记得早些年Maven还没那么广泛使用，通常要对接第三方
阅读更多...
基于SpringBoot的宠物服务系统+uniapp小程序+LW参考示例

基于SpringBoot的宠物服务系统+uniapp小程序+LW参考示例

系列文章目录 1.基于SSM的洗衣房管理系统+原生微信小程序+LW参考示例 2.基于SpringBoot的宠物摄影网站管理系统+LW参考示例 3.基于SpringBoot+Vue的企业人事管理系统+LW参考示例 4.基于SSM的高校实验室管理系统+LW参考示例 5.基于SpringBoot的二手数码回收系统+原生微信小程序+LW参考示例 6.基于SSM的民宿预订管理系统+LW参考示例 7.基于
阅读更多...
Spring Roo 实站（ 一 ）部署安装 第一个示例程序

Spring Roo 实站（ 一 ）部署安装 第一个示例程序

转自：http://blog.csdn.net/jun55xiu/article/details/9380213 一：安装 注：可以参与官网spring-roo： static.springsource.org/spring-roo/reference/html/intro.html#intro-exploring-sampleROO_OPTS http://stati
阅读更多...
java线程深度解析（五）——并发模型（生产者-消费者）

java线程深度解析（五）——并发模型（生产者-消费者）

http://blog.csdn.net/Daybreak1209/article/details/51378055 三、生产者-消费者模式     在经典的多线程模式中，生产者-消费者为多线程间协作提供了良好的解决方案。基本原理是两类线程，即若干个生产者和若干个消费者，生产者负责提交用户请求任务（到内存缓冲区），消费者线程负责处理任务（从内存缓冲区中取任务进行处理），两类线程之
阅读更多...
Java基础回顾系列-第五天-高级编程之API类库

Java基础回顾系列-第五天-高级编程之API类库

Java基础回顾系列-第五天-高级编程之API类库 Java基础类库StringBufferStringBuilderStringCharSequence接口AutoCloseable接口RuntimeSystemCleaner对象克隆 数字操作类Math数学计算类Random随机数生成类BigInteger/BigDecimal大数字操作类 日期操作类DateSimpleDateForma
阅读更多...
Restful API 原理以及实现

Restful API 原理以及实现

先说说API 再说啥是RESRFUL API之前，咱先说说啥是API吧。API大家应该都知道吧，简称接口嘛。随着现在移动互联网的火爆，手机软件，也就是APP几乎快爆棚了。几乎任何一个网站或者应用都会出一款iOS或者Android APP，相比网页版的体验，APP确实各方面性能要好很多。 那么现在问题来了。比如QQ空间网站，如果我想获取一个用户发的说说列表。 QQ空间网站里面需要这个功能。
阅读更多...
京东物流查询｜开发者调用API接口实现

京东物流查询｜开发者调用API接口实现

快递聚合查询的优势 1、高效整合多种快递信息。2、实时动态更新。3、自动化管理流程。 聚合国内外1500家快递公司的物流信息查询服务，使用API接口查询京东物流的便捷步骤，首先选择专业的数据平台的快递API接口：物流快递查询API接口-单号查询API - 探数数据 以下示例是参考的示例代码： import requestsurl = "http://api.tanshuapi.com/a
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2