Kafka 消费者专题

news2025/1/8 11:42:43

目录

  • 消费者
    • 消费者组
    • 消费方式
    • 消费规则
    • 独立消费主题
      • 代码示例（极简）
      • 代码示例（独立消费分区）
    • 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. 一个消费者允许消费多个主题

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

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

相关文章

解决 ：VS code右键没有go to definition选项（转到定义选项）

解决 ：VS code右键没有go to definition选项（转到定义选项）

问题背景&#xff1a; VScode 右键没有“go to definition”选项了&#xff0c;情况如图所示&#xff1a; 问题解决办法&#xff1a; 第一步&#xff1a;先检查没有先安装C/C插件&#xff0c;没有安装就先安装下。 第二步&#xff1a; 打开VS CODE设置界面&#xff1a;文件->…
阅读更多...
网络安全的学习与实践经验（附资料合集）

网络安全的学习与实践经验（附资料合集）

学习资源 在线学习平台&#xff1a; Hack This Site&#xff1a;提供从初学者到高级难度的挑战任务&#xff0c;适合练习各种网络安全技术。XCTF_OJ&#xff1a;由XCTF组委会开发的免费在线网络安全网站&#xff0c;提供丰富的培训材料和资源。SecurityTube&#xff1a;提供丰…
阅读更多...
《Rust权威指南》学习笔记（五）

《Rust权威指南》学习笔记（五）

高级特性 1.在Rust中&#xff0c;unsafe是一种允许绕过Rust的安全性保证的机制&#xff0c;用于执行一些Rust默认情况下不允许的操作。unsafe存在的原因是&#xff1a;unsafe 允许执行某些可能被 Rust 的安全性检查阻止的操作&#xff0c;从而可以进行性能优化&#xff0c;如手…
阅读更多...
使用R语言绘制标准的中国地图和世界地图

使用R语言绘制标准的中国地图和世界地图

在日常的学习和生活中&#xff0c;有时我们常常需要制作带有国界线的地图。这个时候绘制标准的国家地图就显得很重要。目前国家标准地图服务系统向全社会公布的标准中国地图数据&#xff0c;是最权威的地图数据。 今天介绍的R包“ggmapcn”&#xff0c;就是基于最新公布的地图…
阅读更多...
Flutter踩坑记-第三方SDK不兼容Gradle 8.0，需适配namespace

Flutter踩坑记-第三方SDK不兼容Gradle 8.0，需适配namespace

最近需要集成Flutter作为Module&#xff0c;Flutter依赖了第三方库&#xff0c;Gradle是8.0版本。 编译报错&#xff1a; 解决办法是在.android根目录下的build.gradle下新增一行代码&#xff1a; buildscript {ext.kotlin_version "1.8.22"repositories {google()…
阅读更多...
golang 编程规范 - 项目目录结构

golang 编程规范 - 项目目录结构

原文&#xff1a;https://makeoptim.com/golang/standards/project-layout 目录结构 Go 目录 cmdinternalpkgvendor 服务端应用程序目录 api Web 应用程序目录 web 通用应用程序目录 buildconfigsdeploymentsinitscriptstest 其他目录 assetsdocsexamplesgithooksthird_par…
阅读更多...
蓝桥杯备赛：C++基础，顺序表和vector（STL）

蓝桥杯备赛：C++基础，顺序表和vector（STL）

目录 一.C基础 1.第一个C程序&#xff1a; 2.头文件&#xff1a; 3.cin和cout初识&#xff1a; 4.命名空间&#xff1a; 二.顺序表和vector&#xff08;STL&#xff09; 1.顺序表的基本操作&#xff1a; 2.封装静态顺序表&#xff1a; 3.动态顺序表--vector&#xff1a;…
阅读更多...
node.js之---事件循环机制

node.js之---事件循环机制

事件循环机制 Node.js 事件循环机制&#xff08;Event Loop&#xff09;是其核心特性之一&#xff0c;它使得 Node.js 能够高效地处理大量并发的 I/O 操作。Node.js 基于 非阻塞 I/O&#xff0c;使用事件驱动的模型来实现异步编程。事件循环是 Node.js 实现异步编程的基础&…
阅读更多...
如何在 Ubuntu 22.04 上部署 Nginx 并优化以应对高流量网站教程

如何在 Ubuntu 22.04 上部署 Nginx 并优化以应对高流量网站教程

简介 本教程将教你如何优化 Nginx&#xff0c;使其能够高效地处理高流量网站。 Nginx 是一个强大且高性能的 Web 服务器&#xff0c;以其高效处理大量并发连接的能力而闻名&#xff0c;这使得它成为高流量网站的流行选择。 正确优化 Nginx 可以显著提高服务器的性能&#xff0…
阅读更多...
AIRemoveBackground：用 AI 技术轻松去除背景图的前端程序

AIRemoveBackground：用 AI 技术轻松去除背景图的前端程序

在当今数字化时代&#xff0c;图像处理技术不断发展&#xff0c;其中 AI 去除背景图的功能备受关注。本文将介绍一款名为 AIRemoveBackground 的前端程序&#xff0c;它利用人工智能技术&#xff0c;为用户提供便捷、高效的背景去除解决方案。 一、简介 随着互联网的普及和多媒…
阅读更多...
【踩坑指南2.0 2025最新】Scala中如何在命令行传入参数以运行主函数

【踩坑指南2.0 2025最新】Scala中如何在命令行传入参数以运行主函数

这个地方基本没有任何文档记录&#xff0c;在学习的过程中屡屡碰壁&#xff0c;因此记录一下这部分的内容&#xff0c;懒得看可以直接跳到总结看结论。 踩坑步骤 首先来看看书上让我们怎么写&#xff1a; //main.scala object Start {def main(args:Array[String]) {try {v…
阅读更多...
Excel VBA 自动填充空白并合并相同值的解决方案

Excel VBA 自动填充空白并合并相同值的解决方案

文章目录 Excel VBA: 自动填充空白并合并相同值的解决方案问题背景解决方案1. VBA代码实现2. 代码说明3. 使用方法4. 注意事项 扩展优化总结 Excel VBA: 自动填充空白并合并相同值的解决方案 问题背景 在Excel中经常会遇到这样的数据处理需求&#xff1a;一列数据中存在多个空…
阅读更多...
SpringSecurity中的过滤器链与自定义过滤器

SpringSecurity中的过滤器链与自定义过滤器

关于 Spring Security 框架中的过滤器的使用方法，系列文章： 《SpringSecurity中的过滤器链与自定义过滤器》 《SpringSecurity使用过滤器实现图形验证码》 1、Spring Security 中的过滤器链 Spring Security 中的过滤器链（Filter Chain）是一个核心的概念，它定义了一系列过…
阅读更多...
【STC库函数】Compare比较器的使用

【STC库函数】Compare比较器的使用

如果我们需要比较两个点的电压&#xff0c;当A点高于B点的时候我们做一个操作&#xff0c;当B点高于A点的时候做另一个操作。 我们除了加一个运放或者比较器&#xff0c;还可以直接使用STC内部的一个比较器。 正极输入端可以是P37、P50、P51&#xff0c;或者从ADC的十六个通道…
阅读更多...
Postgresql 命令还原数据库

Postgresql 命令还原数据库

因为PgAdmin打不开&#xff0c;但是数据库已经安装成功了&#xff0c;这里借助Pg命令来还原数据库 C:\Program Files\PostgreSQL\15\bin\psql.exe #链接数据库 psql -U postgres -p 5432#创建数据库 CREATE DATABASE "数据库名称"WITHOWNER postgresENCODING UTF8…
阅读更多...
Backend - C# 的日志 NLog日志

Backend - C# 的日志 NLog日志

目录 一、注入依赖和使用 logger 二、配置记录文件 1.安装插件 NLog 2.创建 nlog.config 配置文件 3. Programs配置日志信息 4. 设置 appsettings.json 的 LogLevel 5. 日志设定文件和日志级别的优先级 &#xff08;1&#xff09;常见的日志级别优先级 &#xff08;2&…
阅读更多...
急需升级，D-Link 路由器漏洞被僵尸网络广泛用于 DDoS 攻击

急需升级，D-Link 路由器漏洞被僵尸网络广泛用于 DDoS 攻击

僵尸网络活动增加 &#xff1a;新的“FICORA”和“CAPSAICIN”僵尸网络&#xff08;Mirai 和 Kaiten 的变体&#xff09;的活动激增。 被利用的漏洞 &#xff1a;攻击者利用已知的 D-Link 路由器漏洞&#xff08;例如 CVE-2015-2051、CVE-2024-33112&#xff09;来执行恶意命…
阅读更多...
[ubuntu-22.04]ubuntu不识别rtl8153 usb转网口

[ubuntu-22.04]ubuntu不识别rtl8153 usb转网口

问题描述 ubuntu22.04插入rtl8153 usb转网口不识别 解决方案 安装依赖包 sudo apt-get install libelf-dev build-essential linux-headers-uname -r sudo apt-get install gcc-12 下载源码 Realtek USB FE / GBE / 2.5G / 5G Ethernet Family Controller Softwarehttps:/…
阅读更多...
WinForm开发-自定义组件-1. 工具栏: UcompToolStrip

WinForm开发-自定义组件-1. 工具栏: UcompToolStrip

这里写自定义目录标题 1. 工具栏: UcompToolStrip1.1 展示效果1.2 代码UcompToolStrip.csUcompToolStrip.Designer.cs 1. 工具栏: UcompToolStrip 自定义一些Winform组件 1.1 展示效果 1&#xff09;使用效果 2&#xff09;控件事件 1.2 代码 设计 编码 UcompToolStrip.…
阅读更多...
Hypium纯血鸿蒙系统 HarmonyOS NEXT自动化测试框架

Hypium纯血鸿蒙系统 HarmonyOS NEXT自动化测试框架

1、什么是Hypium Hypium是华为官方为鸿蒙操作系统开发的一款以python为语言的自动化测试框架。 引用华为官网介绍如下&#xff1a; DevEco Testing Hypium(以下简称Hypium)是HarmonyOS平台的UI自动化测试框架&#xff0c;支持开发者使用python语言为应用编写UI自动化测试脚本…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023