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。

相关文章

PostgreSQL中rank()窗口函数实用指南与示例

PostgreSQL中rank()窗口函数实用指南与示例

《PostgreSQL中rank()窗口函数实用指南与示例》在数据分析和数据库管理中,经常需要对数据进行排名操作,PostgreSQL提供了强大的窗口函数rank(),可以方便地对结果集中的行进行排名... 目录一、rank()函数简介二、基础示例：部门内员工薪资排名示例数据排名查询三、高级应用示例1. 每
阅读更多...
使用Python删除Excel中的行列和单元格示例详解

使用Python删除Excel中的行列和单元格示例详解

《使用Python删除Excel中的行列和单元格示例详解》在处理Excel数据时,删除不需要的行、列或单元格是一项常见且必要的操作,本文将使用Python脚本实现对Excel表格的高效自动化处理,感兴... 目录开发环境准备使用 python 删除 Excphpel 表格中的行删除特定行删除空白行删除含指定
阅读更多...
SpringBoot线程池配置使用示例详解

SpringBoot线程池配置使用示例详解

《SpringBoot线程池配置使用示例详解》SpringBoot集成@Async注解,支持线程池参数配置（核心数、队列容量、拒绝策略等）及生命周期管理,结合监控与任务装饰器,提升异步处理效率与系统... 目录一、核心特性二、添加依赖三、参数详解四、配置线程池五、应用实践代码说明拒绝策略（Rejected
阅读更多...
SQL中如何添加数据(常见方法及示例)

SQL中如何添加数据(常见方法及示例)

《SQL中如何添加数据(常见方法及示例)》SQL全称为StructuredQueryLanguage,是一种用于管理关系数据库的标准编程语言,下面给大家介绍SQL中如何添加数据,感兴趣的朋友一起看看吧... 目录在mysql中，有多种方法可以添加数据。以下是一些常见的方法及其示例。1. 使用INSERT I
阅读更多...
Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

Knife4j+Axios+Redis前后端分离架构下的 API 管理与会话方案(最新推荐)

《Knife4j+Axios+Redis前后端分离架构下的API管理与会话方案(最新推荐)》本文主要介绍了Swagger与Knife4j的配置要点、前后端对接方法以及分布式Session实现原理,... 目录一、Swagger 与 Knife4j 的深度理解及配置要点Knife4j 配置关键要点1.Spri
阅读更多...
SpringBoot中SM2公钥加密、私钥解密的实现示例详解

SpringBoot中SM2公钥加密、私钥解密的实现示例详解

《SpringBoot中SM2公钥加密、私钥解密的实现示例详解》本文介绍了如何在SpringBoot项目中实现SM2公钥加密和私钥解密的功能,通过使用Hutool库和BouncyCastle依赖,简化... 目录一、前言1、加密信息(示例)2、加密结果(示例)二、实现代码1、yml文件配置2、创建SM2工具
阅读更多...
MySQL 定时新增分区的实现示例

MySQL 定时新增分区的实现示例

《MySQL定时新增分区的实现示例》本文主要介绍了通过存储过程和定时任务实现MySQL分区的自动创建,解决大数据量下手动维护的繁琐问题,具有一定的参考价值,感兴趣的可以了解一下... mysql创建好分区之后，有时候会需要自动创建分区。比如，一些表数据量非常大，有些数据是热点数据，按照日期分区MululbU
阅读更多...
Python函数作用域示例详解

Python函数作用域示例详解

《Python函数作用域示例详解》本文介绍了Python中的LEGB作用域规则,详细解析了变量查找的四个层级,通过具体代码示例,展示了各层级的变量访问规则和特性,对python函数作用域相关知识感兴趣... 目录一、LEGB 规则二、作用域实例2.1 局部作用域（Local）2.2 闭包作用域（Enclos
阅读更多...
C++20管道运算符的实现示例

C++20管道运算符的实现示例

《C++20管道运算符的实现示例》本文简要介绍C++20管道运算符的使用与实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录标准库的管道运算符使用自己实现类似的管道运算符我们不打算介绍太多，因为它实际属于c++20最为重要的
阅读更多...
Java中调用数据库存储过程的示例代码

Java中调用数据库存储过程的示例代码

《Java中调用数据库存储过程的示例代码》本文介绍Java通过JDBC调用数据库存储过程的方法,涵盖参数类型、执行步骤及数据库差异,需注意异常处理与资源管理,以优化性能并实现复杂业务逻辑,感兴趣的朋友... 目录一、存储过程概述二、Java调用存储过程的基本javascript步骤三、Java调用存储过程示
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2