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,默认是5000ms
        prop.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.同步提交offset
            consumer.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.异步提交offset
            consumer.commitAsync();
        }
    }
}

其他说明

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

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

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

相关文章

重庆大学软件工程复试怎么准备？

重庆大学软件工程复试怎么准备？

重大软件复试相对来说不算刁钻&#xff0c;关键是对自己的竞赛和项目足够了解&#xff0c;能应对老师的提问。专业课范围广&#xff0c;英文文献看个人水平&#xff0c;难度不算大&#xff0c;整体只要表现得得体从容&#xff0c;以及充分的准备&#xff0c;老师不会为难你。 …
阅读更多...
Hadoop•FinalShell连接VMware免密登录

Hadoop•FinalShell连接VMware免密登录

听说这是目录哦 FinalShell连接VMware&#x1f324;️解决重连失效FinalShell的使用 免密登录⛈️能量站&#x1f61a; FinalShell连接VMware&#x1f324;️ 保持虚拟机的开机状态&#xff0c;打开FinalShell&#xff0c;如果虚拟机关机或者挂起&#xff0c;连接就会断开。 …
阅读更多...
List-顺序表--2

List-顺序表--2

目录 1、ArrayList 2、ArrayList构造方法 3、ArrayList常见方法 4、ArrayList的遍历 5、ArrayList的扩容机制 6、ArrayList的具体使用 6.1、杨辉三角 6.2、简单的洗牌算法 1、ArrayList 在集合框架中&#xff0c;ArrayList 是一个普通的类&#xff0c;实现了 List 接口…
阅读更多...
通过串口通信控制led灯的亮灭

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

初始化led灯的gpio接口控制灯的亮灭 初始化uart1串口 将gpio9和gpio10设置为复用模式进行串口通信 通过串口的输入输出函数实现串口通信控制led灯的亮灭
阅读更多...
git知识点汇总

git知识点汇总

git init 初始化一个git仓库&#xff0c;后面可以加仓库名&#xff0c;在当前目录下创建指定名称的目录并在该目录下创建仓库&#xff0c;若不加则直接在当前目录下创建仓库。git仓库的三个区域&#xff1a;工作区&#xff08;当前目录&#xff09;、暂存区&#xff08;.git/in…
阅读更多...
电子电气架构 --- 中央HPC架构

电子电气架构 --- 中央HPC架构

我是穿拖鞋的汉子，魔都中坚持长期主义的汽车电子工程师。 老规矩，分享一段喜欢的文字，避免自己成为高知识低文化的工程师： 所谓鸡汤，要么蛊惑你认命，要么怂恿你拼命，但都是回避问题的根源，以现象替代逻辑，以情绪代替思考，把消极接受现实的懦弱，伪装成乐观面对不幸的…
阅读更多...
让 Agent 具备语音交互能力：技术突破与应用前景（16/30）

让 Agent 具备语音交互能力：技术突破与应用前景（16/30）

让 Agent 具备语音交互能力&#xff1a;技术突破与应用前景 一、引言 在当今数字化时代&#xff0c;人机交互方式正经历着深刻的变革。从早期的命令行界面到图形用户界面&#xff0c;再到如今日益普及的语音交互&#xff0c;人们对于与机器沟通的便捷性和自然性有了更高的追求…
阅读更多...
L27.【LeetCode笔记】2 的幂(五种解法)

L27.【LeetCode笔记】2 的幂(五种解法)

目录 1.题目 2.自解 方法1:调用log函数 代码 提交结果 方法2:循环 提交结果 3.优解 方法3:位运算n & (n-1) 0 代码 提交结果 方法4:位运算lowbit 代码 提交结果 4.投机取巧的方法 代码 提交结果 1.题目 https://leetcode.cn/problems/power-of-two/?env…
阅读更多...
第0章 机器人及自动驾驶SLAM定位方法全解析及入门进阶学习建议

第0章 机器人及自动驾驶SLAM定位方法全解析及入门进阶学习建议

嗨&#xff0c;各位同学大家好&#xff01;笔者自985硕士毕业后&#xff0c;在机器人算法领域已经深耕 7 年多啦。这段时间里&#xff0c;我积累了不少宝贵经验。本专栏《机器人工程师带你从零入门SLAM》将结合下面的SLAM知识体系思维导图及多年的工作实战总结&#xff0c;将逐…
阅读更多...
密码学原理技术-第十一章-Hash Functions

密码学原理技术-第十一章-Hash Functions

文章目录 总结Why we need hash functionsDigital Signature with a Hash FunctionBasic Protocol for Digital Signatures with a Hash FunctionPrincipal input–output behavior of hash functions Security propertiesThe three security requirements of hash functionsWh…
阅读更多...
Docker 远程访问完整配置教程以及核心参数理解

Docker 远程访问完整配置教程以及核心参数理解

Docker 远程访问完整配置教程 以下是配置 Docker 支持远程访问的完整教程&#xff0c;包括参数说明、配置修改、云服务器安全组设置、主机防火墙配置&#xff0c;以及验证远程访问的详细步骤。 1. 理解 -H fd:// 参数的作用&#xff08;理解了以后容易理解后面的操作&#xff…
阅读更多...
单元测试3.0+ @RunWith(JMockit.class)+mock+injectable+Expectations

单元测试3.0+ @RunWith(JMockit.class)+mock+injectable+Expectations

Jmockit使用笔记_基本功能使用Tested_Injectable_Mocked_Expectations_jmockit.class-CSDN博客 静态变量直接赋值就好&#xff0c;没必要mock了 测试框架Jmockit集合junit使用 RunWith(JMockit.class) 写在测试案例类上的注解 Tested 在测试案例中,写在我们要测试的类上…
阅读更多...
vue数据请求通用方案：axios的options都有哪些值

vue数据请求通用方案：axios的options都有哪些值

Axios 是一个基于 promise 的 HTTP 库&#xff0c;可以用在浏览器和 Node.js 中。 在使用 Axios 发送请求时&#xff0c;可以通过传递一个配置对象来指定请求的各种选项。 以下是一些常用的 Axios 配置选项及其说明&#xff1a; 1.url: &#xff08;必需&#xff09;请求的 …
阅读更多...
MySQL 08 章——聚合函数

MySQL 08 章——聚合函数

聚合函数是对一组数据进行汇总的函数&#xff0c;输入的是一组数据的集合&#xff0c;输出的是单个值 MySQL中&#xff0c;目前不能对聚合函数进行嵌套 一、聚合函数介绍 &#xff08;1&#xff09;AVG和SUM函数 举例&#xff1a;只适用于数值类型的字段&#xff08;或变量…
阅读更多...
JVM对象创建过程

JVM对象创建过程

1 类加载检查 jvm通过new指令开始创建对象jvm执行new指令时&#xff0c;首先通过指令参数从常量池中取到需要创建的类名检查该类是否被加载&#xff0c;解析&#xff0c;和初始化过如果没有&#xff0c;则执行类的加载过程new指令对应到java语言具体的操作为 new 关键字创建对象…
阅读更多...
Outlook2024版如何回到经典Outlook

Outlook2024版如何回到经典Outlook

Outlook2024版如何回到经典Outlook 如果新加入一家公司&#xff0c;拿到的电脑&#xff0c;大概率是最新版的Windows, 一切都是新的。 如果不coding, 使用国产的foxmail大概就可以解决一切问题了。可惜老程序员很多Coding都是基于传统Outlook的&#xff0c;科技公司所有人都是I…
阅读更多...
三甲医院等级评审八维数据分析应用（五）--数据集成与共享篇

三甲医院等级评审八维数据分析应用（五）--数据集成与共享篇

一、引言 1.1 研究背景与意义 随着医疗卫生体制改革的不断深化以及信息技术的飞速发展，三甲医院评审作为衡量医院综合实力与服务水平的重要标准，对数据集成与共享提出了更为严苛的要求。在传统医疗模式下，医院内部各业务系统往往各自为政，形成诸多“信息孤岛”，使得数据…
阅读更多...
Scala_【4】流程控制

Scala_【4】流程控制

第四章 分支控制if-else单分支双分支多分支返回值嵌套分支 For循环控制包含边界不包含边界循环守卫循环步长嵌套循环循环返回值 While循环Break友情链接 分支控制if-else 单分支 双分支 多分支 返回值 嵌套分支 For循环控制 Scala也为for循环这一常见的控制结构提供了非常多的…
阅读更多...
Nginx - 整合lua 实现对POST请求的参数拦截校验(不使用Openresty)

Nginx - 整合lua 实现对POST请求的参数拦截校验(不使用Openresty)

文章目录 概述步骤 1: 安装 Nginx 和 Lua 模块步骤 2: 创建 Lua 脚本用于参数校验步骤 3: 配置 Nginx 使用 Lua 脚本写法二&#xff1a; 状态码写法三 &#xff1a; 返回自定义JSON复杂的正则校验 步骤 4: 测试和验证ngx.HTTP_* 枚举值 概述 一个不使用 OpenResty 的 Nginx 集…
阅读更多...
医院机房运维：所有IT资源运行状态同一平台实时呈现

医院机房运维：所有IT资源运行状态同一平台实时呈现

在当今数字化医疗高速发展的时代&#xff0c;医院的信息化系统已然成为保障医疗服务顺畅开展、守护患者生命健康的关键基础设施。以郑州人民医院为例&#xff0c;随着医疗业务不断拓展&#xff0c;其背后支撑的机房运维面临着诸多棘手难题。 传统的分散式人工维护模式&#xff…
阅读更多...
最新文章

Copyright @ 2022~2023