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…
阅读更多...
轻量级通信协议 JSON-RPC 2.0 详解

轻量级通信协议 JSON-RPC 2.0 详解

目录 JSON-RPC 2.0 简介 请求对象 响应对象 通知 批量请求 错误码 使用场景 文档和版本控制 社区和支持 小结 参考资料 JSON-RPC 2.0 简介 JSON-RPC (JavaScript Object Notation - Remote Procedure Call) 是一种轻量级的远程过程调用协议&#xff0c;使用 JSON&am…
阅读更多...
ffmpeg之h264格式转yuv

ffmpeg之h264格式转yuv

h264格式转yuv详细步骤 初始化FFmpeg库&#xff1a;通过av_register_all()来初始化必要的组件。打开输入文件并查找解码器&#xff1a;使用avformat_open_input和 avcodec_find_decoder 打开H.264文件&#xff0c;并查找视频流。分配并配置解码上下文&#xff1a;使用 avcodec…
阅读更多...
创建VUE脚手架

创建VUE脚手架

1.输入 npm create vuelatest2.创建完成
阅读更多...
vue3如何实现防抖？

vue3如何实现防抖？

第一 防抖就是我们设置一个调用时间&#xff0c;点击后设置时间开始倒计时&#xff0c;如果再次点击会重新倒计时 npm或yarn安装&#xff1a; npm install lodash <template><div click"debouncedInputHandler"><button>打印</button>…
阅读更多...
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…
阅读更多...
用shell把(1到100)100个整数写入到二进制文件

用shell把(1到100)100个整数写入到二进制文件

author: hjjdebug date: 2025年 01月 05日 星期日 16:14:41 CST description: 用shell把(1到100)100个整数写入到二进制文件 1. 问题分析(需求分析) 在c语言下, 整数int 是按4个字节存储的,能表示最大0xffffffff的范围 但shell并没有一条命令能够实现把一个整数写入二进制文件…
阅读更多...
SpringBoot下载文件的几种方式

SpringBoot下载文件的几种方式

小文件&#xff1a;直接将文件一次性读取到内存中&#xff0c;文件大可能会导致OOM GetMapping("/download1")public void download1(HttpServletResponse response) throws IOException {// 指定要下载的文件File file new File("C:\\Users\\syd\\Desktop\\do…
阅读更多...
Flink使用

Flink使用

Window下启动支持 下载或复制老版本的放在bin目录下即可&#xff1b; flink.bat echo off setlocalSET bin%~dp0 SET FLINK_HOME%bin%.. SET FLINK_LIB_DIR%FLINK_HOME%\lib SET FLINK_PLUGINS_DIR%FLINK_HOME%\pluginsSET JVM_ARGS-Xmx512mSET FLINK_JM_CLASSPATH%FLINK_LI…
阅读更多...
Python基础知识回顾-数据结构

Python基础知识回顾-数据结构

Tuple 在 Python 中&#xff0c;我们经常使用 Tuple 来将相关数据分组。Tuple 指的是有序且不可变的元素集合。 形式&#xff1a;通常以逗号分隔的元素写在括号"() "中。 数据类型和索引&#xff1a;包含 String、整数和浮点数&#xff0c;并使用正索引和负索引访问…
阅读更多...
8. C++ 面向对象之特性一（封装）

8. C++ 面向对象之特性一（封装）

面向对象主要包括三大类&#xff1a;封装&#xff0c;继承&#xff0c;多态 1.类和对象 c认为&#xff0c;万物皆为对象&#xff0c;对象上有其属性和行为 人可以作为对象&#xff0c;属性有姓名、年龄、身高、体重...&#xff0c;行为有走、跑、跳、吃饭、唱歌... 车也可以作…
阅读更多...
WebRtc02:WebRtc架构、目录结构、运行机制

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

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

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

快速添加对应类所需的头文件包含 一&#xff0c;简介二&#xff0c;操作步骤 一&#xff0c;简介 本文介绍一下&#xff0c;如何快速添加对应类所需要包含的头文件&#xff0c;可以提高开发效率&#xff0c;供参考。 二&#xff0c;操作步骤 以QTime类为例&#xff1a; 选中…
阅读更多...
Apache MINA 反序列化漏洞CVE-2024-52046

Apache MINA 反序列化漏洞CVE-2024-52046

漏洞描述&#xff1a; Apache MINA 是一个功能强大、灵活且高性能的网络应用框架。它通过抽象网络层的复杂性&#xff0c;提供了事件驱动架构和灵活的 Filter 链机制&#xff0c;使得开发者可以更容易地开发各种类型的网络应用。 Apache MINA 框架的 ObjectSerializationDeco…
阅读更多...
服务器docker配置过程

服务器docker配置过程

1.docker安装 参考官方文档&#xff1a;https://docker.cadn.net.cn/manuals/engine_install_ubuntu 2.docker镜像源替换 官方文档&#xff1a;https://cloud.tencent.com/document/product/1207/45596 镜像源根据你租了哪家的去找官方文档即可。
阅读更多...
RabbitMQ通过代码创建交换机和队列

RabbitMQ通过代码创建交换机和队列

常见交换机 RabbitMQ提供的交换机一共的四种&#xff0c;分别是&#xff1a; 1. Fanout&#xff1a;采用广播形式来发送消息&#xff0c;会将消息路由到所有绑定了的队列之中。 2. Direct&#xff1a;通过Binding Key与队列绑定&#xff0c;生产者在发送信息的时候会通过Routin…
阅读更多...
js es6 reduce函数， 通过规格生成sku

js es6 reduce函数， 通过规格生成sku

const specs [{ name: 颜色, values: [红色, 蓝色, 绿色] },{ name: 尺寸, values: [S, M, L] } ];function generateSKUs(specs) {return specs.reduce((acc, spec) > {const newAcc [];for (const combination of acc) {for (const value of spec.values) {newAcc.push(…
阅读更多...
WPF通过反射机制动态加载控件

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

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

Copyright @ 2022~2023