当前位置: 首页 > news >正文

java 中 main 方法使用 KafkaConsumer 拉取 kafka 消息如何禁止输出 debug 日志

pom 依赖:

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.5.14.RELEASE</version>
</dependency>

 或者

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.5.1</version>
</dependency>

 ps:前面的 spring-kafka 依赖中已经包含了后面的 kafka-clients

KafkaConsumerDemo.java:


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.PartitionInfo;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.LoggerFactory;import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.LoggerContext;import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.logging.Logger;public class KafkaConsumerDemo {static Map<String,Object> properties = new HashMap<String,Object>();private static KafkaConsumer kafkaConsumer = null;/**** windows 环境需要将下面 8 行添加到 "C:\Windows\System32\drivers\etc\hosts" 文件中:*      xxx.xxx.xxx.xxx1 xxx-data01*      xxx.xxx.xxx.xxx2 xxx-data02*      xxx.xxx.xxx.xxx3 xxx-data03*      xxx.xxx.xxx.xxx4 xxx-data04*      xxx.xxx.xxx.xxx5 xxx-data05*      xxx.xxx.xxx.xxx6 xxx-data06*      xxx.xxx.xxx.xxx7 xxx-data07*      xxx.xxx.xxx.xxx8 xxx-data08* @param args*/public static void main(String[] args) {// 禁止控制台输出一些 org.apache.kafka.xxx 相关的日志LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();loggerContext.getLogger("org.apache.kafka.clients.consumer.internals.ConsumerCoordinator").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.FetchSessionHandler").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.internals.Fetcher").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.internals.AbstractCoordinator").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.NetworkClient").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.common.network.Selector").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.internals.OffsetsForLeaderEpochClient").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.Metadata").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.KafkaConsumer").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.common.utils.AppInfoParser").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.KafkaConsumer").setLevel(Level.OFF);loggerContext.getLogger("org.apache.kafka.clients.consumer.ConsumerConfig").setLevel(Level.OFF);properties.put("bootstrap.servers","127.0.0.1:9192,127.0.0.1:9192,127.0.0.1:9192");  // 指定 Brokerproperties.put("group.id", "11111111111111111111111");              // 指定消费组群 ID,为防止自己启动拉取消息导致其他生产环境的消费者无法消费该消息,请设置一个绝对不重复的值,以起到隔离的作用properties.put("max.poll.records", "1000");// todo 设置可批量拉取???properties.put("enable.auto.commit", "false");properties.put("key.deserializer", StringDeserializer.class); // 将 key 的字节数组转成 Java 对象properties.put("value.deserializer", StringDeserializer.class);  // 将 value 的字节数组转成 Java 对象kafkaConsumer = new KafkaConsumer<String, String>(properties);// List<String> topics = queryAllTopics( consumer );kafkaConsumer.subscribe( Collections.singletonList( "ods_carbon_rfid_device_record" ) );  // 订阅主题 order-eventsnew Thread(new Runnable() {@Overridepublic void run() {receiveMessage();}}).start();}/*** 查询全部的主题(topic)列表* @param kafkaConsumer* @return*/private static List<String> queryAllTopics(KafkaConsumer kafkaConsumer) {if( kafkaConsumer == null ){return null;}Map<String, List<PartitionInfo>> map = kafkaConsumer.listTopics();if( map == null ){return null;}return new ArrayList<String>( map.keySet() );}public static void receiveMessage() {try {while ( true ){synchronized (KafkaConsumerDemo.class) {// ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));// 30L 表示超时时间为 30秒,有消息立即返回,没消息最多等 30 秒后返回SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");ConsumerRecords<String,String> records = kafkaConsumer.poll(Duration.ofSeconds(30L));String date = sdf.format(new Date());if( records == null ){System.out.println( date + " 本次未拉取到任何消息" );}else {System.out.println( date + " 本次拉取到 " + records.count() + " 条消息" );int i = 1;for (ConsumerRecord<String,String> record: records) {String info = String.format("[Topic: %s][Partition:%d][Offset:%d][Key:%s][Message:%s]", record.topic(), record.partition(), record.offset(), record.key(), record.value());System.out.println( "第" + i + "条消息:" + info );i++;}kafkaConsumer.commitSync();}/*** 当你用 KafkaConsumer从Kafka里读取消息并且处理完后,commitSync 方法会帮你把这些消息的处理进度(也就是偏移量 offset )同步地告诉 Kafka 服务器。* 这样,Kafka 就知道你已经处理到哪儿了。如果消费者(也就是读取消息的程序)突然崩溃或者重启,Kafka 就能根据最后一次提交的偏移量,让你从上一次处理* 完的地方继续开始,而不会漏掉或者重复处理消息。* 简单来说,commitSync 方 法就是用来“保存进度”的,确保消息处理的可靠性和顺序性。*/// Thread.sleep( 5000L );}}} catch (Exception e){e.printStackTrace();} finally {kafkaConsumer.close();}}
}

相关文章:

java 中 main 方法使用 KafkaConsumer 拉取 kafka 消息如何禁止输出 debug 日志

pom 依赖&#xff1a; <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.5.14.RELEASE</version> </dependency> 或者 <dependency><groupId>org.ap…...

【后端面试总结】Golang可能的内存泄漏场景及应对策略

Golang可能的内存泄漏场景及应对策略 一、引言 Golang作为一种高性能、并发友好的编程语言&#xff0c;其内置的垃圾回收机制极大地简化了内存管理。然而&#xff0c;这并不意味着开发者可以完全忽视内存泄漏问题。在实际开发中&#xff0c;由于不当的资源管理、循环引用、以…...

Java 反射机制详解

在 Java 编程世界中&#xff0c;反射机制犹如一把神奇的钥匙&#xff0c;它能够打开许多隐藏在代码深处的 “大门”&#xff0c;让开发者突破常规的限制&#xff0c;实现一些极具灵活性的功能。今天&#xff0c;就跟随我一同深入探究 Java 反射机制的奥秘。 一、什么是反射 反…...

【k8s】scc权限 restricted、anyuid、privileged

文章目录 概述1. 内置的scc2. OpenShift如何确定pod的scc2.1 Pod未带SCC标签的情况2.2. Pod带有SCC标签的情况 参考 概述 在OpenShift&#xff08;后文简称OCP&#xff09;中&#xff0c;很早就一个概念&#xff1a;Security Context Constraints &#xff0c;简称SCC&#xf…...

2025华数杯国际赛A题完整论文讲解(含每一问python代码+数据+可视化图)

大家好呀&#xff0c;从发布赛题一直到现在&#xff0c;总算完成了2025“华数杯”国际大学生数学建模竞赛A题Can He Swim Faster的完整的成品论文。 本论文可以保证原创&#xff0c;保证高质量。绝不是随便引用一大堆模型和代码复制粘贴进来完全没有应用糊弄人的垃圾半成品论文…...

ThreadLocal 的使用场景

在现代电商平台中&#xff0c;ThreadLocal 常用于以下场景&#xff0c;特别是与线程隔离相关的业务中&#xff0c;以提高性能和简化上下文传递。 1. 用户上下文信息管理 场景&#xff1a;在用户发起的每次请求中&#xff0c;需要携带用户 ID、角色、权限等信息&#xff0c;而这…...

后端开发 Springboot整合Redis Spring Data Redis 模板

目录 redis 配置 RedisConfig 类 完整代码 代码讲解 1. 类定义和注解 2. 定义 RedisTemplate Bean 3. 配置 JSON 序列化 4. 配置 Redis 的 key 和 value 序列化方式 5. 完成配置并返回 RedisTemplate 总结 redis 服务接口实现类 类级别 注入 RedisTemplate 常用 Re…...

代码随想录算法训练营第 4 天(链表 2)| 24. 两两交换链表中的节点19.删除链表的倒数第N个节点 -

一、24. 两两交换链表中的节点 题目&#xff1a;24. 两两交换链表中的节点 - 力扣&#xff08;LeetCode&#xff09; 视频&#xff1a;帮你把链表细节学清楚&#xff01; | LeetCode&#xff1a;24. 两两交换链表中的节点_哔哩哔哩_bilibili 讲解&#xff1a;代码随想录 dummy-…...

【RDMA学习笔记】1:RDMA(Remote Direct Memory Access)介绍

从帝国理工的PPT学习。 什么是RDMA Remote Direct Memory Access&#xff0c;也就是Remote的DMA&#xff0c;是一种硬件机制&#xff0c;能直接访问远端结点的内存&#xff0c;而不需要处理器介入。 其中&#xff1a; Remote&#xff1a;跨node进行数据传输Direct&#xff…...

网络安全常见的35个安全框架及模型

大家读完觉得有帮助记得关注和点赞&#xff01;&#xff01;&#xff01; 01、概述 网络安全专业机构制定的一套标准、准则和程序&#xff0c;旨在帮助组织了解和管理面临的网络安全风险。优秀的安全框架及模型应该为用户提供一种可靠方法&#xff0c;帮助其实现网络安全建设…...

Elasticsearch介绍及使用

Elasticsearch 是一款基于 Lucene 库构建的开源、分布式、RESTful 风格的搜索引擎和分析引擎&#xff0c;具有强大的全文搜索、数据分析、机器学习等功能&#xff0c;广泛应用于日志分析、实时数据分析、全文检索等场景。 核心概念 索引&#xff08;Index&#xff09;&#xf…...

Leetocde516. 最长回文子序列 动态规划

原题链接&#xff1a;Leetocde516. 最长回文子序列 class Solution { public:int longestPalindromeSubseq(string s) {int n s.size();vector<vector<int>> dp(n, vector<int>(n, 1));for (int i 0; i < n; i) {dp[i][i] 1;if (i 1 < n &&…...

iOS 逆向学习 - Inter-Process Communication:进程间通信

iOS 逆向学习 - Inter-Process Communication&#xff1a;进程间通信 一、进程间通信概要二、iOS 进程间通信机制详解1. URL Schemes2. Pasteboard3. App Groups 和 Shared Containers4. XPC Services 三、不同进程间通信机制的差异四、总结 一、进程间通信概要 进程间通信&am…...

高级生化大纲

一&#xff0c;蛋白质化学&#xff1a; 蛋白质分离是生物化学和分子生物学研究中的一项基本技术&#xff0c;用于根据蛋白质的物理和化学特性将其从混合物中分离出来。 1. 离心分离法 离心分离法利用离心力来分离不同质量或密度的颗粒和分子。 差速离心&#xff1a;通过逐…...

YARN WebUI 服务

一、WebUI 使用 与HDFS一样&#xff0c;YARN也提供了一个WebUI服务&#xff0c;可以使用YARN Web用户界面监视群集、队列、应用程序、服务、流活动和节点信息。还可以查看集群详细配置的信息&#xff0c;检查各种应用程序和服务的日志。 1.1 首页 浏览器输入http://node2.itc…...

【Unity3D】利用IJob、Burst优化处理切割物体

参考文章&#xff1a; 【Unity】切割网格 【Unity3D】ECS入门学习&#xff08;一&#xff09;导入及基础学习_unity ecs教程-CSDN博客 【Unity3D】ECS入门学习&#xff08;十二&#xff09;IJob、IJobFor、IJobParallelFor_unity ijobparallelfor-CSDN博客 工程资源地址&…...

【大前端】Vue3 工程化项目使用详解

目录 一、前言 二、前置准备 2.1 环境准备 2.1.1 create-vue功能 2.1.2 nodejs环境 2.1.3 配置nodejs的环境变量 2.1.4 更换安装包的源 三、工程化项目创建与启动过程 3.1 创建工程化项目 3.2 项目初始化 3.3 项目启动 3.4 核心文件说明 四、VUE两种不同的API风格 …...

基于文件系统分布式锁原理

分布式锁&#xff1a;在一个公共的存储服务上打上一个标记&#xff0c;如Redis的setnx命令&#xff0c;是先到先得方式获得锁&#xff0c;ZooKeeper有点像下面的demo,比较大小的方式判决谁获得锁。 package com.ldj.mybatisflex.demo;import java.util.*; import java.util.co…...

简历整理YH

一&#xff0c;订单中心 1&#xff0c;调拨单 融通(Rocketmq)-订单中心&#xff1a;ECC_BMS123(已出单)&#xff0c;125(分配),127(发货),129(收货) 通过RocketMq接入多场景订单数据 2&#xff0c;销售单 sap&#xff08;FTP&#xff09;-订单中心&#xff0c;下发1002,1003,…...

Kotlin 协程基础三 —— 结构化并发(二)

Kotlin 协程基础系列&#xff1a; Kotlin 协程基础一 —— 总体知识概述 Kotlin 协程基础二 —— 结构化并发&#xff08;一&#xff09; Kotlin 协程基础三 —— 结构化并发&#xff08;二&#xff09; Kotlin 协程基础四 —— CoroutineScope 与 CoroutineContext Kotlin 协程…...

vscode里如何用git

打开vs终端执行如下&#xff1a; 1 初始化 Git 仓库&#xff08;如果尚未初始化&#xff09; git init 2 添加文件到 Git 仓库 git add . 3 使用 git commit 命令来提交你的更改。确保在提交时加上一个有用的消息。 git commit -m "备注信息" 4 …...

Cursor实现用excel数据填充word模版的方法

cursor主页&#xff1a;https://www.cursor.com/ 任务目标&#xff1a;把excel格式的数据里的单元格&#xff0c;按照某一个固定模版填充到word中 文章目录 注意事项逐步生成程序1. 确定格式2. 调试程序 注意事项 直接给一个excel文件和最终呈现的word文件的示例&#xff0c;…...

SkyWalking 10.2.0 SWCK 配置过程

SkyWalking 10.2.0 & SWCK 配置过程 skywalking oap-server & ui 使用Docker安装在K8S集群以外&#xff0c;K8S集群中的微服务使用initContainer按命名空间将skywalking-java-agent注入到业务容器中。 SWCK有整套的解决方案&#xff0c;全安装在K8S群集中。 具体可参…...

基于ASP.NET+ SQL Server实现(Web)医院信息管理系统

医院信息管理系统 1. 课程设计内容 在 visual studio 2017 平台上&#xff0c;开发一个“医院信息管理系统”Web 程序。 2. 课程设计目的 综合运用 c#.net 知识&#xff0c;在 vs 2017 平台上&#xff0c;进行 ASP.NET 应用程序和简易网站的开发&#xff1b;初步熟悉开发一…...

Oracle查询表空间大小

1 查询数据库中所有的表空间以及表空间所占空间的大小 SELECTtablespace_name,sum( bytes ) / 1024 / 1024 FROMdba_data_files GROUP BYtablespace_name; 2 Oracle查询表空间大小及每个表所占空间的大小 SELECTtablespace_name,file_id,file_name,round( bytes / ( 1024 …...

关于nvm与node.js

1 安装nvm 安装过程中手动修改 nvm的安装路径&#xff0c; 以及修改 通过nvm安装node后正在使用的node的存放目录【这句话可能难以理解&#xff0c;但接着往下看你就了然了】 2 修改nvm中settings.txt文件配置 nvm安装成功后&#xff0c;通常在该文件中会出现以下配置&…...

蓝桥杯 2024 15届国赛 A组 儿童节快乐

P10576 [蓝桥杯 2024 国 A] 儿童节快乐 题目描述 五彩斑斓的气球在蓝天下悠然飘荡&#xff0c;轻快的音乐在耳边持续回荡&#xff0c;小朋友们手牵着手一同畅快欢笑。在这样一片安乐祥和的氛围下&#xff0c;六一来了。 今天是六一儿童节&#xff0c;小蓝老师为了让大家在节…...

聊一聊接口测试的意义有哪些?

目录 一、隔离性 & 早期测试 二、保障系统集成质量 三、验证业务逻辑的核心层 四、提升测试效率与覆盖度 五、系统稳定性的守护者 六、驱动团队协作与契约管理 七、性能与扩展性的前置评估 八、持续交付的核心支撑 接口测试的意义可以从四个维度展开&#xff0c;首…...

是否存在路径(FIFOBB算法)

题目描述 一个具有 n 个顶点e条边的无向图&#xff0c;该图顶点的编号依次为0到n-1且不存在顶点与自身相连的边。请使用FIFOBB算法编写程序&#xff0c;确定是否存在从顶点 source到顶点 destination的路径。 输入 第一行两个整数&#xff0c;分别表示n 和 e 的值&#xff08;1…...

论文阅读笔记——Muffin: Testing Deep Learning Libraries via Neural Architecture Fuzzing

Muffin 论文 现有方法 CRADLE 和 LEMON&#xff0c;依赖模型推理阶段输出进行差分测试&#xff0c;但在训练阶段是不可行的&#xff0c;因为训练阶段直到最后才有固定输出&#xff0c;中间过程是不断变化的。API 库覆盖低&#xff0c;因为各个 API 都是在各种具体场景下使用。…...