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。

相关文章

Debezium 与 Apache Kafka 的集成方式步骤详解

Debezium 与 Apache Kafka 的集成方式步骤详解

《Debezium与ApacheKafka的集成方式步骤详解》本文详细介绍了如何将Debezium与ApacheKafka集成,包括集成概述、步骤、注意事项等,通过KafkaConnect,D... 目录一、集成概述二、集成步骤1. 准备 Kafka 环境2. 配置 Kafka Connect3. 安装 D
阅读更多...
Java调用DeepSeek API的最佳实践及详细代码示例

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》：本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.
阅读更多...
Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

《Android悬浮窗开发示例((动态权限请求|前台服务和通知|悬浮窗创建)》本文介绍了Android悬浮窗的实现效果,包括动态权限请求、前台服务和通知的使用,悬浮窗权限需要动态申请并引导... 目录一、悬浮窗 动态权限请求1、动态请求权限2、悬浮窗权限说明3、检查动态权限4、申请动态权限5、权限设置完毕后
阅读更多...
在 Spring Boot 中使用 @Autowired和 @Bean注解的示例详解

在 Spring Boot 中使用 @Autowired和 @Bean注解的示例详解

《在SpringBoot中使用@Autowired和@Bean注解的示例详解》本文通过一个示例演示了如何在SpringBoot中使用@Autowired和@Bean注解进行依赖注入和Bean... 目录在 Spring Boot 中使用 @Autowired 和 @Bean 注解示例背景1. 定义 Stud
阅读更多...
oracle DBMS_SQL.PARSE的使用方法和示例

oracle DBMS_SQL.PARSE的使用方法和示例

《oracleDBMS_SQL.PARSE的使用方法和示例》DBMS_SQL是Oracle数据库中的一个强大包,用于动态构建和执行SQL语句,DBMS_SQL.PARSE过程解析SQL语句或PL/S... 目录语法示例注意事项DBMS_SQL 是 oracle 数据库中的一个强大包，它允许动态地构建和执行
阅读更多...
Python中顺序结构和循环结构示例代码

Python中顺序结构和循环结构示例代码

《Python中顺序结构和循环结构示例代码》：本文主要介绍Python中的条件语句和循环语句,条件语句用于根据条件执行不同的代码块,循环语句用于重复执行一段代码,文章还详细说明了range函数的使... 目录一、条件语句（1）条件语句的定义（2）条件语句的语法（a）单分支 if（b）双分支 if-else（
阅读更多...
Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

Deepseek R1模型本地化部署+API接口调用详细教程(释放AI生产力)

《DeepseekR1模型本地化部署+API接口调用详细教程(释放AI生产力)》本文介绍了本地部署DeepSeekR1模型和通过API调用将其集成到VSCode中的过程,作者详细步骤展示了如何下载和... 目录前言一、deepseek R1模型与chatGPT o1系列模型对比二、本地部署步骤1.安装oll
阅读更多...
浅析如何使用Swagger生成带权限控制的API文档

浅析如何使用Swagger生成带权限控制的API文档

《浅析如何使用Swagger生成带权限控制的API文档》当涉及到权限控制时,如何生成既安全又详细的API文档就成了一个关键问题,所以这篇文章小编就来和大家好好聊聊如何用Swagger来生成带有... 目录准备工作配置 Swagger权限控制给 API 加上权限注解查看文档注意事项在咱们的开发工作里，API
阅读更多...
Python中Markdown库的使用示例详解

Python中Markdown库的使用示例详解

《Python中Markdown库的使用示例详解》Markdown库是一个用于处理Markdown文本的Python工具,这篇文章主要为大家详细介绍了Markdown库的具体使用,感兴趣的... 目录一、背景二、什么是 Markdown 库三、如何安装这个库四、库函数使用方法1. markdown.mark
阅读更多...
一分钟带你上手Python调用DeepSeek的API

一分钟带你上手Python调用DeepSeek的API

《一分钟带你上手Python调用DeepSeek的API》最近DeepSeek非常火,作为一枚对前言技术非常关注的程序员来说,自然都想对接DeepSeek的API来体验一把,下面小编就来为大家介绍一下... 目录前言免费体验API-Key申请首次调用API基本概念最小单元推理模型智能体自定义界面总结前言最
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2