Apache Kafka - 重识消费者

文章目录

  • 概述
    • Kafka消费者的工作原理
    • Kafka消费者的配置
    • Kafka消费者的实现
      • 高级API
      • 低级API
  • 导图
    • 总结

在这里插入图片描述


概述

Kafka是一个分布式的消息队列系统,它的出现解决了传统消息队列系统的吞吐量瓶颈问题。

Kafka的高吞吐量、低延迟和可扩展性使得它成为了很多公司的首选消息队列系统。

在Kafka中,消息被分成了不同的主题(Topic),每个主题又被分成了不同的分区(Partition)。

生产者(Producer)将消息发送到指定的主题中,而消费者(Consumer)则从指定的主题中读取消息。

接下来我们将介绍Kafka消费者相关的知识。

Kafka消费者的工作原理

Kafka消费者从指定的主题中读取消息,消费者组(Consumer Group)则是一组消费者的集合,它们共同消费一个或多个主题。在一个消费者组中,每个消费者都会独立地读取主题中的消息。当一个主题有多个分区时,每个消费者会读取其中的一个或多个分区。消费者组中的消费者可以动态地加入或退出,这样就可以实现消费者的动态扩展。

Kafka消费者通过轮询(Polling)方式从Kafka Broker中读取消息。当一个消费者从Broker中读取到一条消息后,它会将该消息的偏移量(Offset)保存在Zookeeper或Kafka内部主题中。消费者组中的消费者会协调并平衡分区的分配,保证每个消费者读取的分区数量尽可能均衡。

Kafka消费者的配置

  1. bootstrap.servers

    该参数用于指定Kafka集群中的broker地址,多个地址以逗号分隔。消费者会从这些broker中获取到集群的元数据信息,以便进行后续的操作。

  2. group.id

    该参数用于指定消费者所属的消费组,同一消费组内的消费者共同消费一个主题的消息。如果不指定该参数,则会自动生成一个随机的group.id。

  3. enable.auto.commit

    该参数用于指定是否启用自动提交offset。如果设置为true,则消费者会在消费消息后自动提交offset;如果设置为false,则需要手动提交offset。

  4. auto.commit.interval.ms

    该参数用于指定自动提交offset的时间间隔,单位为毫秒。只有当enable.auto.commit设置为true时,该参数才会生效。

  5. session.timeout.ms

    该参数用于指定消费者与broker之间的会话超时时间,单位为毫秒。如果消费者在该时间内没有发送心跳包,则会被认为已经失效,broker会将其从消费组中移除。

  6. max.poll.records

    该参数用于指定每次拉取消息的最大条数。如果一次拉取的消息数量超过了该参数指定的值,则消费者需要等待下一次拉取消息。

  7. auto.offset.reset

    该参数用于指定当消费者第一次加入消费组或者offset失效时,从哪个位置开始消费。可选值为latest和earliest,分别表示从最新的消息和最早的消息开始消费。

  8. max.poll.interval.ms

    该参数用于指定两次poll操作之间的最大时间间隔,单位为毫秒。如果消费者在该时间内没有进行poll操作,则被认为已经失效,broker会将其从消费组中移除。

  9. fetch.min.bytes

    该参数用于指定每次拉取消息的最小字节数。如果一次拉取的消息数量不足该参数指定的字节数,则消费者需要等待下一次拉取消息。

  10. fetch.max.wait.ms

    该参数用于指定拉取消息的最大等待时间,单位为毫秒。如果在该时间内没有获取到足够的消息,则返回已经获取到的消息。


Kafka消费者的实现

Kafka消费者的实现可以使用Kafka提供的高级API或者低级API。高级API封装了低级API,提供了更加简洁、易用的接口。下面分别介绍一下这两种API的使用方法。

高级API

使用高级API可以更加方便地实现Kafka消费者。下面是一个使用高级API实现Kafka消费者的示例代码:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

在上面的代码中,我们首先创建了一个Properties对象,用于存储Kafka消费者的配置信息。然后创建了一个KafkaConsumer对象,并指定了要消费的主题。最后使用poll方法从Broker中读取消息,并对每条消息进行处理。

低级API

使用低级API可以更加灵活地实现Kafka消费者。下面是一个使用低级API实现Kafka消费者的示例代码:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "fal	VCC se");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        consumer.commitSync(Collections.singletonMap(record.topicPartition(), new OffsetAndMetadata(record.offset() + 1)));
    }
}

在上面的代码中,我们首先创建了一个Properties对象,用于存储Kafka消费者的配置信息。然后创建了一个KafkaConsumer对象,并指定了要消费的主题。最后使用poll方法从Broker中读取消息,并对每条消息进行处理。在处理完每条消息后,我们使用commitSync方法手动提交偏移量。


导图

在这里插入图片描述

总结

Kafka消费者是Kafka消息队列系统中的重要组成部分,它能够从指定的主题中读取消息,并进行相应的处理。在使用Kafka消费者时,需要注意消费者组ID、自动提交偏移量、偏移量重置策略以及消息处理方式等配置信息。

在这里插入图片描述

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

面试前15天刷完这个笔记,拿下字节测开岗offer....

面试&#xff0c;跳槽&#xff0c;每天都在发生&#xff0c;跳槽&#xff0c;更是很常见的&#xff0c;对于每个人来说&#xff0c;跳槽的意义也各不相同&#xff0c;可能是一个人更向往一个更大的平台&#xff0c;更好的地方&#xff0c;可以通过换一个环境改变自己的现状。而…

解密Java Class文件不为人知的秘密

Java 诞生多年&#xff0c;因此在网络上&#xff0c;有关 Java Class 文件格式解析的文章有很多&#xff0c;但他们大多数都是在列举《Java 虚拟机》中定义的格式&#xff0c;通读下来&#xff0c;好像所有的东西都讲清楚了&#xff0c;但是我个人好像并没有看懂&#xff0c;不…

2. css表格属性、文本属性、列表属性、边距属性、尺寸属性

1. 表格属性 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta http-equiv"X-UA-Compatible" content"IEedge"><meta name"viewport" content"widthdevice-width…

【JavaSE】Java基础语法(十五):继承

文章目录 1. 继承的实现2. 继承的好处和弊端3. Java中继承的特点4. 继承中的成员访问特点5. super6. 继承中构造方法的访问特点7. 继承中成员方法的访问特点8. super内存图9. 方法重写10. 权限修饰符 1. 继承的实现 继承的概念 继承是面向对象三大特征之一&#xff0c;可以使得…

【状态估计】基于随机方法优化PMU优化配置(Matlab代码实现)

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…

路径规划算法:基于猫群优化的路径规划算法- 附代码

路径规划算法&#xff1a;基于猫群优化的路径规划算法- 附代码 文章目录 路径规划算法&#xff1a;基于猫群优化的路径规划算法- 附代码1.算法原理1.1 环境设定1.2 约束条件1.3 适应度函数 2.算法结果3.MATLAB代码4.参考文献 摘要&#xff1a;本文主要介绍利用智能优化算法猫群…

基于docker部署testlink并集成mantis

使用docker pull命令拉取需要的镜像。由于testlink和mantis都需要存储相关数据&#xff0c;所以这里可以看到还拉取了一个mysql镜像。 # docker pull bitnami/testlink:1.9.16-r8 # docker pull vimagick/mantisbt # docker pull mysql:5.7.20 使用docker network命令中创建…

Altium Designer 相同电路多组复制布线

在进行设计开发的时候&#xff0c;总会遇到相同的电路&#xff0c;或者模块&#xff0c;这些电路可以使用相同的布局和走线。我们可以画好其中一部分&#xff0c;然后直接复制&#xff0c;就可以提高效率。下面记录我自己的实际操作过程&#xff0c;有一些地方遇到了问题&#…

抽象轻松JavaScript

想象一样&#xff0c;现在有一个苹果&#xff0c;两个苹果&#xff0c;一箱苹果在你面前 看&#xff0c;上面的三种苹果&#xff0c;&#xff08;我写的是苹果就是苹果&#xff09; 语境1 例如你现在要搬运苹果&#xff01; 那么现在上面有苹果&#xff0c;一个&#xff0c;两…

大数据好找工作么?前景如何

大数据好不好找工作不是一概而论的&#xff0c;要根据你个人的学历情况&#xff0c;掌握技能程度&#xff0c;所在城市招聘需求&#xff0c;甚至是你的面试能力和简历是否突出优势有关。 但是毋庸置疑的是&#xff0c;大数据目前的发展前景还是相当优秀的。 我们知道&#xf…

每天一个面试题之final在java中有什么作用?

final在java中有什么作用&#xff1f; final关键字表示最终的含义 当它用来修饰一个引用时&#xff1a; <1>:如果引用为基本数据类型&#xff0c;则该引用为常量&#xff0c;该值无法被修改。<2>:如果引用为引用数据类型&#xff0c;例如&#xff0c;对象/数组等…

Java的CookieManager

文章目录 1. 简介2. CookieStore 1. 简介 Java5包括一个抽象类Java.net.CookieHandler&#xff0c;它定义了存储和获取Cookie的一个API&#xff0c;但不包括这个抽象类的实现&#xff0c;所以还有很多工作要做。Java6进一步作了补充&#xff0c;为CookieManager增加了一个可以…

基于信息间隙决策理论的碳捕集电厂调度(Matlab代码实现)

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…

unity 渲染性能分析工具

目标 既然要优化&#xff0c;肯定要有个目标&#xff1a; pc上一般要求&#xff1a;一秒渲染60帧 移动端&#xff1a;一秒渲染30帧 这应该是最低的要求&#xff0c;如果游戏运行时&#xff0c;游戏帧率有变化&#xff0c;人眼能够明显的感觉到帧率下降。 优化的首要规则是找到…

OpenCV实战(20)——图像投影关系

OpenCV实战(20)——图像投影关系 0. 前言1. 相机成像原理2. 图像对的基本矩阵3. 完整代码小结系列链接0. 前言 数码相机通过将光线通过镜头投射到图像传感器上来捕捉场景产生图像。由于通过将 3D 场景投影到 2D 平面上形成图像,因此场景与其图像之间以及同一场景的不同图像…

vscode远程到服务器(包括WSL)进行GDB调试

工欲善其事必先利其器&#xff0c;这句话不容小觑&#xff0c;调试工具做的好&#xff0c;对开发工作可起到事半功倍。 本文主要讲vscode远程到服务器进行在线GDB调试手段&#xff0c;包含对WSL的远程调试&#xff0c;可以轻松对照源码进行应用程序调试。 文章目录 一、vscode…

损失函数——交叉熵损失(Cross-entropy loss)

交叉熵损失&#xff08;Cross-entropy loss&#xff09;是深度学习中常用的一种损失函数&#xff0c;通常用于分类问题。它衡量了模型预测结果与实际结果之间的差距&#xff0c;是优化模型参数的关键指标之一。以下是交叉熵损失的详细介绍。 假设我们有一个分类问题&#xff0…

L-shape 方法

L-shape 方法是求解两阶段随机规划的一种常用方法&#xff0c;基本思想是利用切平面将第二阶段的反馈函数线性化&#xff0c;在构造切平面条件时有点类似 bender’s 方法。 注&#xff1a;这个图形中黑实线 Q ( x ) \mathcal{Q}(x) Q(x) 就是下面模型中的 L ( x ) \mathscr{L…

为什么要用线程池?

线程池是一种管理和复用线程资源的机制&#xff0c;它由一个线程池管理器和一组工作线程组成。线程池管理器负责创建和销毁线程池&#xff0c;以及管理线程池中的工作线程。工作线程则负责执行具体的任务。 线程池的主要作用是管理和复用线程资源&#xff0c;避免了线程的频繁…

IOS开发指南之自定义TableViewCell使用

演示效果: 1.自定义TableViewCell创建 File->new->File... 在iOS模板中选择Empty来创建一个空的XIB文件,然后点击下一步 输入XIB文件名Cell,然后点击Create创建 创建XIB文件成功后如下: 同时按钮Shift+command+L弹出库,然后输入 table筛选,选择Table View Cell 拖到下…