Kafka 消费者专题

目录

  • 消费者
    • 消费者组
    • 消费方式
    • 消费规则
    • 独立消费主题
      • 代码示例（极简）
      • 代码示例（独立消费分区）
    • offset
      • 自动提交
      • 代码示例（自动提交）
      • 手动提交
      • 代码示例（同步）
      • 代码示例（异步）
    • 其他说明

消费者

消费者组

  1. 由多个消费者组成
  2. 消费者组之间互不影响。
  3. 其他消费规则如下

消费方式

  1. push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。

  2. consumer采用pull（拉）模式从broker中读取数据。pull模式则可以根据consumer的消费能力以适当的速率消费消息。
    pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka的消费者在消费数据时会传入一个时长参数timeout，如果当前没有数据可供消费，consumer会等待一段时间之后再返回，这段时长即为timeout。

消费规则

在这里插入图片描述

  1. 一个消费者（单独消费者或消费者组中的一个）可以消费一个分区中的数据也可以消费两个或以上的分区数据
  2. 消费者组中的消费者必须访问不同的数据分区，不能访问同一个
  3. 同一个分区中的数据允许被不同的消费者访问（消费者不属于同一个组）

独立消费主题

代码示例（极简）

以下代码创建模拟一个消费者组（testCg）中的消费者，订阅来自topicA的消息
CustomTopicConsumer.java

package com.wunaiieq.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;//创建一个独立消费者，消费topicA主题下的数据
public class CustomTopicConsumer {public static void main(String[] args) {//1.创建消费者属性文件对象Properties prop = new Properties();//2.为属性对象设置相关参数//设置kafka服务器prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.16.100:9092");//设置key和value的序列化类prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//设置消费者的消费者组的名称prop.put(ConsumerConfig.GROUP_ID_CONFIG, "testCg");//3.创建消费者对象KafkaConsumer<String, String> kafkaConsumer =new KafkaConsumer<String, String>(prop);//4.注册要消费的主题ArrayList<String> topics = new ArrayList<>();topics.add("topicA");//订阅主题kafkaConsumer.subscribe(topics);//5.拉取数据并打印输出while (true) {//6.设置1s消费一批数据ConsumerRecords<String, String> consumerRecords =kafkaConsumer.poll(Duration.ofSeconds(1));//7.打印输出消费到的数据for (ConsumerRecord consumerRecord : consumerRecords) {System.out.println(consumerRecord);}}}
}

代码示例（独立消费分区）

这个消费者需要消费topicA分区下的0号分区数据

package com.wunaiieq.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;//创建一个独立消费者，消费topicA主题0号分区中的数据
public class ConsumTopicPartitionConsumer {public static void main(String[] args) {//1.创建属性对象Properties prop = new Properties();//2.设置相关参数prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.16.100:9092");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.GROUP_ID_CONFIG,"testCg2");//3.创建消费者对象KafkaConsumer<String,String> kafkaConsumer =new KafkaConsumer<String, String>(prop);//4.为消费者注册主题和分区号List<TopicPartition> topicPartitions =new ArrayList<>();topicPartitions.add(new TopicPartition("topicA",0));kafkaConsumer.assign(topicPartitions);//5.消费数据while(true){ConsumerRecords<String, String> consumerRecords =kafkaConsumer.poll(Duration.ofSeconds(1));for(ConsumerRecord consumerRecord:consumerRecords){System.out.println(consumerRecord);}}}
}

offset

表示消费者在特定主题分区中的消费进度。
一般而言，这个offset不会主动去用，除非宕机重启等等
可以手动查看offset值和状况

修改系统配置
[root@node4 ~]# cd /opt/kafka/config/ [root@node4 config]# vim
consumer.properties exclude.internal.topics=false
查询offset
kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server node2:9092 --consumer.config config/consumer.properties --formatter “kafka.coordinator.group.GroupMetadataManager$OffsetsMessageFormatter” --from-beginning

描述和作用：

  • Offset是Kafka中标识消息在分区内位置的一个唯一标识符。每个消息都有一个对应的Offset值，用于表示消息在分区中的相对位置。Offset是从0开始递增的，每当有新的消息写入分区时，Offset就会加1。Offset是不可变的，即使消息被删除或过期，Offset也不会改变或重用。
  • 定位消息：通过指定Offset，消费者可以准确地找到分区中的某条消息，或者从某个位置开始消费消息。
  • 记录消费进度：消费者在消费完一条消息后，需要提交Offset来告诉Kafka Broker自己消费到哪里了。这样，如果消费者发生故障或重启，它可以根据保存的Offset来恢复消费状态。
  • __consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.id+topic名称+分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 进行compact，也就是每个 group.id+topic名称+分区号就保留最新数据。

自动提交

自动提交主要是根据时间设置，每隔一段时间提交

代码示例（自动提交）

设置offset自动提交，每xxx秒提交一次（默认是5秒）

package com.wunaiieq.offset;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class ConsumerAutoOffset {public static void main(String[] args) {//1.创建属性对象Properties prop = new Properties();//2.设置属性参数prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.16.100:9092");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//配置消费者组prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cgauto");//是否自动提交offset： true表示自动提交，false表示非自动提交prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);//提交offset的时间周期1000ms,默认是5000msprop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);//3.创建消费者对象KafkaConsumer<String,String> kafkaConsumer =new KafkaConsumer<String, String>(prop);//4.设置消费主题kafkaConsumer.subscribe(Arrays.asList("topicA"));//5.消费消息while(true){//6.读取消息ConsumerRecords<String, String> consumerRecords =kafkaConsumer.poll(Duration.ofSeconds(1));//7.循环输出消息for(ConsumerRecord cr:consumerRecords){System.out.println(cr.value());}}}
}

手动提交

手动提交offset的方法有两种方式：

  1. commitSync同步提交:必须等待offset提交完毕，再去消费下一批数据。

  2. commitAsync异步提交:发送完提交offset请求后，就开始消费下一批数据了。
    两者的区别：

相同点是，都会将本次消费的一批数据的最高的偏移量提交；
不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而异步提交则没有失败重试机制，故有可能提交失败。

代码示例（同步）

等待offset提交完毕，再去消费下一批数据。

package com.wunaiieq.offset;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class ConsumerHandSyncCommit {public static void main(String[] args) {//1.创建属性对象Properties prop = new Properties();//2.设置相关参数prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//配置消费者组prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cghandSyncCommit");//设置为非自动提交prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//3.创建消费者对象KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(prop);//4.注册消费主题consumer.subscribe(Arrays.asList("topicA"));//5.消费数据while(true){ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));for(ConsumerRecord record:records){System.out.println(record.value());}//6.同步提交offsetconsumer.commitSync();}}
}

代码示例（异步）

代码上的区别很小，提交方式由consumer.commitSync();改为consumer.commitAsync();

package com.wunaiieq.offset;import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class ConsumerHandASyncCommit {public static void main(String[] args) {//1.创建属性对象Properties prop = new Properties();//2.设置相关参数prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.16.100:9092,192.168.16.101:9092,192.168.16.102:9092");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//配置消费者组prop.put(ConsumerConfig.GROUP_ID_CONFIG,"cghandAsyncCommit");//设置为非自动提交prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//3.创建消费者对象KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(prop);//4.注册消费主题consumer.subscribe(Arrays.asList("topicA"));//5.消费数据while(true){ConsumerRecords<String, String> records =consumer.poll(Duration.ofSeconds(1));for(ConsumerRecord record:records){System.out.println(record.value());}//6.异步提交offsetconsumer.commitAsync();}}
}

其他说明

  1. 一个消费者允许消费多个主题

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/web/65167.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【游戏设计原理】47 - 超游戏思维

【游戏设计原理】47 - 超游戏思维

对于这条原理&#xff0c;我首先想到的是开放世界&#xff0c;或者探索性游戏&#xff0c;这是最能包容各类玩家的游戏类型。这类游戏定义了基本规则&#xff0c;玩家的可操作性很强。就像上图里的沙池一样&#xff0c;里面有滑梯&#xff0c;是规则性比较明确的&#xff0c;而…
阅读更多...
奥迪TT MK1(初代奥迪TT、第一代奥迪TT)仪表盘故障/不精准/水温/剩余油量不准，如何修复、测试、复位？

奥迪TT MK1(初代奥迪TT、第一代奥迪TT)仪表盘故障/不精准/水温/剩余油量不准，如何修复、测试、复位？

故障现象 水温不准&#xff0c;冷启动就130℃汽油加满&#xff0c;指针依然在中间偏左的位置 如下图&#xff1a; 诊断过程 通过VAG KKL 409 USB OBD接口读取水温和油位数值正常&#xff0c;故判断是仪表指针马达损坏或需要重置指针位置 维修步骤 推荐选择CH340&#xff08;老…
阅读更多...
Nginx——服务器端集群搭建与扩展模块（五/五）

Nginx——服务器端集群搭建与扩展模块（五/五）

目录 1.Nginx实现服务器端集群搭建1.1.Nginx 与 Tomcat 部署1.1.1.环境准备 (Tomcat)1.1.2.环境准备 (Nginx) 1.2.Nginx实现动静分离1.2.1.概述1.2.2.需求分析1.2.3.动静分离实现步骤 1.3.Nginx 实现 Tomcat 集群搭建1.4.Nginx 高可用解决方案1.4.1.概述1.4.2.Keepalived 介绍1…
阅读更多...
创建VUE脚手架

创建VUE脚手架

1.输入 npm create vuelatest2.创建完成
阅读更多...
JavaWeb开发（六）XML介绍

JavaWeb开发（六）XML介绍

1. XML介绍 1.1. 什么是XML &#xff08;1&#xff09;XML 指可扩展标记语言(EXtensible Markup Language)XML 是一种很像HTML的标记语言。   &#xff08;2&#xff09;XML 的设计宗旨是传输数据(目前主要是作为配置文件)&#xff0c;而不是显示数据。   &#xff08;3&a…
阅读更多...
WebRtc02:WebRtc架构、目录结构、运行机制

WebRtc02:WebRtc架构、目录结构、运行机制

整体架构 WebRtc主要分为三层&#xff1a; CAPI层&#xff1a;外层调用Session管理核心层&#xff1a;包括视频引擎、音频引擎、网络传输 可由使用者重写视频引擎&#xff1a;编解码器、视频缓存、视频增强音频引擎&#xff1a;编解码器、音频缓存、回音消除、降噪传输&#x…
阅读更多...
【Qt】快速添加对应类所需的头文件包含

【Qt】快速添加对应类所需的头文件包含

快速添加对应类所需的头文件包含 一&#xff0c;简介二&#xff0c;操作步骤 一&#xff0c;简介 本文介绍一下&#xff0c;如何快速添加对应类所需要包含的头文件&#xff0c;可以提高开发效率&#xff0c;供参考。 二&#xff0c;操作步骤 以QTime类为例&#xff1a; 选中…
阅读更多...
WPF通过反射机制动态加载控件

WPF通过反射机制动态加载控件

Activator.CreateInstance 是 .NET 提供的一个静态方法&#xff0c;它属于 System 命名空间。此方法通过反射机制根据提供的类型信息。 写一个小demo演示一下 要求&#xff1a;在用户反馈界面点击建议或者评分按钮 弹出相应界面 编写MainWindow.xmal 主窗体 <Window x:C…
阅读更多...
宽带、光猫、路由器、WiFi、光纤之间的关系

宽带、光猫、路由器、WiFi、光纤之间的关系

1、宽带&#xff08;Broadband&#xff09; 1.1 宽带的定义宽带指的是一种高速互联网接入技术&#xff0c;通常包括ADSL、光纤、4G/5G等不同类型的接入方式。宽带的关键特点是能够提供较高的数据传输速率&#xff0c;使得用户可以享受到稳定的上网体验。 1.2 宽带的作用宽带是…
阅读更多...
Pytest钩子函数，测试框架动态切换测试环境

Pytest钩子函数，测试框架动态切换测试环境

在软件测试中&#xff0c;测试环境的切换是个令人头疼的问题。不同环境的配置不同&#xff0c;如何高效切换测试环境成为许多测试开发人员关注的重点。你是否希望在运行测试用例时&#xff0c;能够动态选择测试环境&#xff0c;而不是繁琐地手动修改配置&#xff1f; Pytest 测…
阅读更多...
印象笔记07——试一试PDF标注

印象笔记07——试一试PDF标注

印象笔记07——试一试PDF标注 [!CAUTION] 根据第六期&#xff0c;我再次查询了资料&#xff0c;印象笔记还是有一些可圈可点的功能的&#xff08;当然部分有平替&#xff09;&#xff0c;针对会员作用&#xff0c;开发使用场景虽然是逆向的&#xff0c;但我坚信这是一部分人的现…
阅读更多...
【Vue】分享一个快速入门的前端框架以及如何搭建

【Vue】分享一个快速入门的前端框架以及如何搭建

先上效果图: 登录 菜单: 下载地址: 链接&#xff1a;https://pan.baidu.com/s/1m-ZlBARWU6_2n8jZil_RAQ 提取码&#xff1a;ui20 … 主要是可以自定义设置token,更改后端请求地址较为方便。 应用设置: 登录与token设置: 在这里设置不用登录,可以请求的接口: request.js i…
阅读更多...
通过串口通信控制led灯的亮灭

通过串口通信控制led灯的亮灭

初始化led灯的gpio接口控制灯的亮灭 初始化uart1串口 将gpio9和gpio10设置为复用模式进行串口通信 通过串口的输入输出函数实现串口通信控制led灯的亮灭
阅读更多...
计算机xinput1_4.dll丢失怎么修复？

计算机xinput1_4.dll丢失怎么修复？

电脑运行时常见问题及修复指南 作为软件开发从业者&#xff0c;深知电脑在日常使用中难免会遇到各种问题&#xff0c;如文件丢失、文件损坏和系统报错等。这些问题不仅影响工作效率&#xff0c;还可能带来数据丢失的风险。本文将详细介绍一些常见问题及其解决办法&#xff0c;…
阅读更多...
DeepSeek V3“报错家门”：我是ChatGPT

DeepSeek V3“报错家门”：我是ChatGPT

搜 &#xff1a;海讯无双Ai 要说这两天大模型圈的顶流话题&#xff0c;那绝对是非DeepSeek V3莫属了。 不过在网友们纷纷测试之际&#xff0c;有个bug也成了热议的焦点—— 只是少了一个问号&#xff0c;DeepSeek V3竟然称自己是ChatGPT。 甚至让它讲个笑话&#xff0c;生成…
阅读更多...
C++：范围for

C++：范围for

范围for&#xff08;range-based for&#xff09;是C的一种循环结构&#xff0c; 是在 C11 这个标准中引入的&#xff0c;这种类型的for循环使得遍历数组、容器中的元素更加简便和直观。 一、范围for语法 for ( 类型 变量名 : 数组名 ) 语句 //多条语句需要加⼤括号 示例&#…
阅读更多...
C++基础概念复习

C++基础概念复习

前言 本篇文章作基础复习用&#xff0c;主要是在C学习中遇到的概念总结&#xff0c;后续会继续补充。如有不足&#xff0c;请前辈指出&#xff0c;万分感谢。 1、什么是封装&#xff0c;有何优点&#xff0c;在C中如何体现封装这一特性&#xff1f; 封装是面向对象编程&…
阅读更多...
前端工程化之手搓webpack5 --【elpis全栈项目】

前端工程化之手搓webpack5 --【elpis全栈项目】

前端工程化之手搓webpack5 --【elpis全栈项目】 导读 基本流程&#xff1a;输入 – 编译 – 输出 #mermaid-svg-V8Gi7RFNikCuEhax {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-V8Gi7RFNikCuEhax .error-icon{fil…
阅读更多...
S7-200可用的modbus RTU完成位轮询

S7-200可用的modbus RTU完成位轮询

网上的信息比较散&#xff0c;官方说明也不充分&#xff0c;尤其是涉及主站按需写入的部分没有见到现成案例。 以下记录完成位轮询读取&#xff0c;同时按需写入的程序。 初始化主站&#xff0c;初始化块的完成位M9.3通过上升沿触发一个M9.4&#xff0c;用于后面启动轮询。 第…
阅读更多...
特征点检测与匹配——MATLAB R2022b

特征点检测与匹配——MATLAB R2022b

特征点检测与匹配在计算机视觉中的作用至关重要，它为图像处理、物体识别、增强现实等领域提供了坚实的基础。 目录 Harris角点检测 SIFT（尺度不变特征变换） SURF（加速稳健特征） ORB（Oriented FAST and Rotated BRIEF） 总结 特征点检测与匹配是计算机视觉中的一项基…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023