kafka-消费者组（SpringBoot整合Kafka）

news2024/12/24 20:19:09

文章目录

  • 1、消费者组
    • 1.1、使用 efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本
    • 1.2、创建生产者发送消息
    • 1.3、application.yml配置
    • 1.4、创建消费者监听器
    • 1.5、创建SpringBoot启动类
    • 1.6、屏蔽 kafka debug 日志 logback.xml
    • 1.7、引入spring-kafka依赖
    • 1.8、消费者控制台：

1、消费者组

1.1、使用 efak 创建 主题 my_topic1 并建立6个分区并给每个分区建立3个副本

在这里插入图片描述

1.2、创建生产者发送消息

package com.atguigu.spring.kafka.consumer;

import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;

@SpringBootTest
class SpringKafkaConsumerApplicationTests {
    @Resource
    KafkaTemplate kafkaTemplate;

    @Test
    void contextLoads() {
        for (int i = 0; i < 10; i++) {
            kafkaTemplate.send("my_topic1",i%6,"", "消费者组"+i);
        }
    }
}

在这里插入图片描述
在这里插入图片描述

1.3、application.yml配置

server:
  port: 8120

# v1
spring:
  Kafka:
    bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097
    consumer:
      # read-committed读事务已提交的消息 解决脏读问题
      isolation-level: read-committed # 消费者的事务隔离级别：read-uncommitted会导致脏读，可以读取生产者事务还未提交的消息
      # 消费者是否自动ack ：true自动ack 消费者获取到消息后kafka提交消费者偏移量
      enable-auto-commit: true 
      # 消费者提交ack时多长时间批量提交一次
      auto-commit-interval: 1000
      # 消费者第一次消费主题消息时从哪个位置开始
      auto-offset-reset: earliest  #指定Offset消费:earliest | latest | none
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

1.4、创建消费者监听器

package com.atguigu.spring.kafka.consumer.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaListener {

    @KafkaListener(topics ={"my_topic1"},groupId = "my_group1")
    public void onMessage1(ConsumerRecord<String, String> record) {
        System.out.println("my_group1消费者1获取到消息：topic = "+ record.topic()
                +"，partition:"+record.partition()
                +"，offset = "+record.offset()
                +"，key = "+record.key()
                +"，value = "+record.value());
    }
    @KafkaListener(topics ={"my_topic1"},groupId = "my_group1")
    public void onMessage2(ConsumerRecord<String, String> record) {
        System.out.println("my_group1消费者2获取到消息：topic = "+ record.topic()
                +"，partition:"+record.partition()
                +"，offset = "+record.offset()
                +"，key = "+record.key()
                +"，value = "+record.value());
    }
    @KafkaListener(topics ={"my_topic1"},groupId = "my_group2")
    public void onMessage3(ConsumerRecord<String, String> record) {
        System.out.println("my_group2消费者获取到消息：topic = "+ record.topic()
                +"，partition:"+record.partition()
                +"，offset = "+record.offset()
                +"，key = "+record.key()
                +"，value = "+record.value());
    }

}

1.5、创建SpringBoot启动类

package com.atguigu.spring.kafka.consumer;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;


// Generated by https://start.springboot.io
// 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn
@SpringBootApplication
public class SpringKafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaConsumerApplication.class, args);
    }

}

1.6、屏蔽 kafka debug 日志 logback.xml

<configuration>　　　　　　
    <!-- 如果觉得idea控制台日志太多，src\main\resources目录下新建logback.xml
屏蔽kafka debug -->
    <logger name="org.apache.kafka.clients" level="debug" />
</configuration>

1.7、引入spring-kafka依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.0.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <!-- Generated by https://start.springboot.io -->
    <!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn -->
    <groupId>com.atguigu</groupId>
    <artifactId>spring-kafka-consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>spring-kafka-consumer</name>
    <description>spring-kafka-consumer</description>
    <properties>
        <java.version>17</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

1.8、消费者控制台：

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::                (v3.0.5)

my_group2消费者获取到消息：topic = my_topic1，partition:2，offset = 0，key = ，value = 消费者组2
my_group2消费者获取到消息：topic = my_topic1，partition:2，offset = 1，key = ，value = 消费者组8
my_group2消费者获取到消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 消费者组0
my_group2消费者获取到消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 消费者组6
my_group2消费者获取到消息：topic = my_topic1，partition:5，offset = 0，key = ，value = 消费者组5
my_group2消费者获取到消息：topic = my_topic1，partition:3，offset = 0，key = ，value = 消费者组3
my_group2消费者获取到消息：topic = my_topic1，partition:3，offset = 1，key = ，value = 消费者组9
my_group2消费者获取到消息：topic = my_topic1，partition:1，offset = 0，key = ，value = 消费者组1
my_group2消费者获取到消息：topic = my_topic1，partition:1，offset = 1，key = ，value = 消费者组7
my_group2消费者获取到消息：topic = my_topic1，partition:4，offset = 0，key = ，value = 消费者组4
my_group1消费者2获取到消息：topic = my_topic1，partition:2，offset = 0，key = ，value = 消费者组2
my_group1消费者2获取到消息：topic = my_topic1，partition:2，offset = 1，key = ，value = 消费者组8
my_group1消费者1获取到消息：topic = my_topic1，partition:5，offset = 0，key = ，value = 消费者组5
my_group1消费者1获取到消息：topic = my_topic1，partition:4，offset = 0，key = ，value = 消费者组4
my_group1消费者1获取到消息：topic = my_topic1，partition:3，offset = 0，key = ，value = 消费者组3
my_group1消费者1获取到消息：topic = my_topic1，partition:3，offset = 1，key = ，value = 消费者组9
my_group1消费者2获取到消息：topic = my_topic1，partition:1，offset = 0，key = ，value = 消费者组1
my_group1消费者2获取到消息：topic = my_topic1，partition:1，offset = 1，key = ，value = 消费者组7
my_group1消费者2获取到消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 消费者组0
my_group1消费者2获取到消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 消费者组6

在这里插入图片描述

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

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

相关文章

【Qt秘籍】[010]-Qt常用控件

【Qt秘籍】[010]-Qt常用控件

一、控件概述 在GUI&#xff08;图形用户界面&#xff09;开发领域&#xff0c;Qt无疑是众多开发者心中的首选框架之一。它不仅跨平台、功能强大&#xff0c;而且拥有丰富且灵活的控件库&#xff0c;使得开发者能够快速构建美观、高效的用户界面。对于初学者而言&#xff0…
阅读更多...
万里长城第一步——尚庭公寓【技术概述】

万里长城第一步——尚庭公寓【技术概述】

简略版&#xff1a; 项目概述主要是移动端&#xff08;房源检索&#xff1b;预约看房&#xff0c;租赁管理&#xff0c;浏览历史&#xff09;和后台管理&#xff08;管理员对房源进行操作&#xff09;&#xff1b; 项目使用前后端分离的方法&#xff0c;主要以后端为主&#xf…
阅读更多...
CSS真题合集（一）

CSS真题合集（一）

CSS真题合集&#xff08;一&#xff09; 1. 盒子模型1.1 盒子模型的基本组成1.2 盒子模型的实际大小1.3 盒子模型的两种类型1.4 设置盒子模型1.5 弹性盒子模型 2. BFC2.1 主要用途2.2 触发BFC的方法2.2 解决外边距的塌陷问题&#xff08;垂直塌陷&#xff09; 3. 响应式布局3.1…
阅读更多...
Django 外键关联数据

Django 外键关联数据

在设计数据库的时候&#xff0c;是得需要通过外键的形式将各个表进行连接。 原先的表是这样的 要想更改成这样&#xff1a; 下面是操作步骤&#xff1a; 有两张表是关联的 # 在 models.py 里创建class Department(models.Model):"""部门表""&quo…
阅读更多...
Qt for android ： 调用 OpenCV库错误集（To Be Continue）

Qt for android ： 调用 OpenCV库错误集（To Be Continue）

error: error: undefined symbol: AMediaExtractor_new 在调用 VideoCapture等库时出现 解决: LIBS -lmediandk
阅读更多...
【ROS2大白话】四、ROS2非常简单的传参方式

【ROS2大白话】四、ROS2非常简单的传参方式

系列文章目录 【ROS2大白话】一、ROS2 humble及cartorgrapher安装 【ROS2大白话】二、turtlebot3安装 【ROS2大白话】三、给turtlebot3安装realsense深度相机 【ROS2大白话】四、ROS2非常简单的传参方式 文章目录 系列文章目录前言一、launch文件传参的demo1. 编写launch.py文…
阅读更多...
Python语法详解module4（函数）

Python语法详解module4（函数）

目录 一、函数基础1. 函数的概念和作用2. 函数的定义和调用3. 参数传递 二、返回值和文档字符串返回值的概念和用法1. 返回值的概念2. 使用 return 关键字返回值&#xff1a;3. 多个返回值的情况&#xff1a; 文档字符串&#xff08;docstring&#xff09;的作用和使用方法1. 文…
阅读更多...
linux 下修改屏幕分辨率

linux 下修改屏幕分辨率

在使用麒麟虚拟机时&#xff0c;不知道咋回事&#xff0c;会自动改变分辨率。 使用界面设置分辨率选项修改时&#xff0c;下面的保存修改按钮显示不出来&#xff0c;无法完成设置。 所以需要使用命令行修改一下分辨率&#xff0c;修改命令如下所示&#xff1a; 1、执行xrand…
阅读更多...
Questflow借助MongoDB Atlas以AI重新定义未来工作方式

Questflow借助MongoDB Atlas以AI重新定义未来工作方式

MongoDB客户案例导读 Questflow借助MongoDB Atlas赋能AI员工&#xff0c;助力中小型初创企业自动化工作流程&#xff0c;简化数据分析&#xff0c;提升客户体验&#xff0c;推动AI与员工的协作&#xff0c;重新定义未来工作方式。 协作式AI自动化平台 无需编码即可拥有自己的…
阅读更多...
动态规划9：LCR 099. 最小路径和

动态规划9：LCR 099. 最小路径和

动态规划解题步骤&#xff1a; 1.确定状态表示&#xff1a;dp[i]是什么 2.确定状态转移方程&#xff1a;dp[i]等于什么 3.初始化&#xff1a;确保状态转移方程不越界 4.确定填表顺序&#xff1a;根据状态转移方程即可确定填表顺序 5.确定返回值 题目链接&#xff1a;LCR …
阅读更多...
Leetcode3170. 删除星号以后字典序最小的字符串

Leetcode3170. 删除星号以后字典序最小的字符串

Every day a Leetcode 题目来源&#xff1a;3170. 删除星号以后字典序最小的字符串 解法1&#xff1a;栈 由于要去掉最小的字母&#xff0c;为了让字典序尽量小&#xff0c;相比去掉前面的字母&#xff0c;去掉后面的字母更好。 从左到右遍历字符串 s&#xff0c;用 26 个栈…
阅读更多...
AI大底座核心平台：百度百舸AI异构计算平台（AI IaaS）与AI中台（AI PaaS）

AI大底座核心平台：百度百舸AI异构计算平台（AI IaaS）与AI中台（AI PaaS）

AI大底座正是整合了以上端到端全要素技术能力&#xff0c;将基础架构IaaS与应用平台PaaS能力深度融合&#xff0c;面向企业和产业AI生 产与应用的全生命周期提供完整解决方案。 百舸AI异构计算平台是AI IaaS层的核心平台&#xff0c;包括AI计算、AI存储、AI加速、AI容器四层套件…
阅读更多...
frp开启Dashboard

frp开启Dashboard

公网服务器 公网服务器frps.toml bindPort 81 # 127.0.0.1 Dashboard 端口&#xff0c;后面会进行内网穿透&#xff0c;因此不用配置ip为0.0.0.0 webServer.port 82auth.token "token令牌"公网服务器frpc.toml # 因为是在公网服务器的frpc所以直接配127.0.0.1 …
阅读更多...
计算机科学（学习笔记三）

计算机科学（学习笔记三）

内容来源&#xff1a;计算机科学 指令和程序 指令&#xff1a;指示计算机要做什么的代码&#xff0c;多条指令共同组成程序。 计算机指令长度 由于早期计算机每个字只有8位&#xff0c;指令只占4位&#xff0c;意味着只能有16个指令&#xff0c;这远远不够。 现代计算机有两…
阅读更多...
【Hive SQL 每日一题】统计各个商品今年销售额与去年销售额的增长率及排名变化

【Hive SQL 每日一题】统计各个商品今年销售额与去年销售额的增长率及排名变化

文章目录 测试数据需求说明需求实现分步解析 测试数据 -- 创建商品表 DROP TABLE IF EXISTS products; CREATE TABLE products (product_id INT,product_name STRING );INSERT INTO products VALUES (1, Product A), (2, Product B), (3, Product C), (4, Product D), (5, Pro…
阅读更多...
【调试笔记-20240606-Linux-为 OpenWrt 的 nginx 服务器添加Shell CGI 支持】

【调试笔记-20240606-Linux-为 OpenWrt 的 nginx 服务器添加Shell CGI 支持】

调试笔记-系列文章目录 调试笔记-20240606-Linux-为 OpenWrt 的 nginx 服务器添加Shell CGI 支持 文章目录 调试笔记-系列文章目录调试笔记-20240606-Linux-为 OpenWrt 的 nginx 服务器添加Shell CGI 支持 前言一、调试环境操作系统&#xff1a;Windows 10 专业版调试环境调试…
阅读更多...
【Qt知识】部分QWidget属性表格

【Qt知识】部分QWidget属性表格

QWidget是Qt库中所有图形用户界面组件的基类&#xff0c;它提供了大量属性以供自定义和配置控件的行为和外观。下面列出了一些主要的QWidget属性及其作用。 属性 作用 accessibleName 控件的辅助技术名称&#xff0c;用于无障碍访问。 accessibleDescription 控件的辅助技…
阅读更多...
Codeforces Round 951 (Div. 2)（A~C）

Codeforces Round 951 (Div. 2)（A~C）

目录 A. Guess the Maximum B. XOR Sequences C. Earning on Bets 这次比赛也是打的稀碎了&#xff0c;第二个少个break检查了15分钟才检查出来&#xff0c;第三个符号搞错了&#xff0c;错了两次&#xff0c;道心直接破碎了 A. Guess the Maximum 题意&#xff1a;我们对…
阅读更多...
机器学习的热门领域及应用趋势

机器学习的热门领域及应用趋势

机器学习的热门领域及应用趋势 近年来&#xff0c;机器学习&#xff08;Machine Learning, ML&#xff09;已经成为科技领域的热门话题&#xff0c;其在各个行业的应用越来越广泛和深入。本文将详细介绍当前机器学习的几个热门领域&#xff0c;以及人们在这些领域中使用的机器…
阅读更多...
【Mac】Alfred 5 for Mac（苹果效率提升工具）v5.5软件介绍及安装教程

【Mac】Alfred 5 for Mac（苹果效率提升工具）v5.5软件介绍及安装教程

软件介绍 Alfred 是适用于 Mac 操作系统的流行生产力应用程序。它旨在帮助用户在 Mac 电脑上更高效地启动应用程序、搜索文件和文件夹以及执行各种任务。借助 Alfred&#xff0c;用户可以创建自定义键盘快捷方式、设置自定义工作流程并使用热键访问功能。 Alfred for Mac 的一…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023