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 依赖: <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.5.14.RELEASE</version> </dependency> 或者 <dependency><groupId>org.ap…...
【后端面试总结】Golang可能的内存泄漏场景及应对策略
Golang可能的内存泄漏场景及应对策略 一、引言 Golang作为一种高性能、并发友好的编程语言,其内置的垃圾回收机制极大地简化了内存管理。然而,这并不意味着开发者可以完全忽视内存泄漏问题。在实际开发中,由于不当的资源管理、循环引用、以…...
Java 反射机制详解
在 Java 编程世界中,反射机制犹如一把神奇的钥匙,它能够打开许多隐藏在代码深处的 “大门”,让开发者突破常规的限制,实现一些极具灵活性的功能。今天,就跟随我一同深入探究 Java 反射机制的奥秘。 一、什么是反射 反…...
【k8s】scc权限 restricted、anyuid、privileged
文章目录 概述1. 内置的scc2. OpenShift如何确定pod的scc2.1 Pod未带SCC标签的情况2.2. Pod带有SCC标签的情况 参考 概述 在OpenShift(后文简称OCP)中,很早就一个概念:Security Context Constraints ,简称SCC…...
2025华数杯国际赛A题完整论文讲解(含每一问python代码+数据+可视化图)
大家好呀,从发布赛题一直到现在,总算完成了2025“华数杯”国际大学生数学建模竞赛A题Can He Swim Faster的完整的成品论文。 本论文可以保证原创,保证高质量。绝不是随便引用一大堆模型和代码复制粘贴进来完全没有应用糊弄人的垃圾半成品论文…...
ThreadLocal 的使用场景
在现代电商平台中,ThreadLocal 常用于以下场景,特别是与线程隔离相关的业务中,以提高性能和简化上下文传递。 1. 用户上下文信息管理 场景:在用户发起的每次请求中,需要携带用户 ID、角色、权限等信息,而这…...
后端开发 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. 两两交换链表中的节点 题目:24. 两两交换链表中的节点 - 力扣(LeetCode) 视频:帮你把链表细节学清楚! | LeetCode:24. 两两交换链表中的节点_哔哩哔哩_bilibili 讲解:代码随想录 dummy-…...
【RDMA学习笔记】1:RDMA(Remote Direct Memory Access)介绍
从帝国理工的PPT学习。 什么是RDMA Remote Direct Memory Access,也就是Remote的DMA,是一种硬件机制,能直接访问远端结点的内存,而不需要处理器介入。 其中: Remote:跨node进行数据传输Directÿ…...
网络安全常见的35个安全框架及模型
大家读完觉得有帮助记得关注和点赞!!! 01、概述 网络安全专业机构制定的一套标准、准则和程序,旨在帮助组织了解和管理面临的网络安全风险。优秀的安全框架及模型应该为用户提供一种可靠方法,帮助其实现网络安全建设…...
Elasticsearch介绍及使用
Elasticsearch 是一款基于 Lucene 库构建的开源、分布式、RESTful 风格的搜索引擎和分析引擎,具有强大的全文搜索、数据分析、机器学习等功能,广泛应用于日志分析、实时数据分析、全文检索等场景。 核心概念 索引(Index)…...
Leetocde516. 最长回文子序列 动态规划
原题链接: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:进程间通信 一、进程间通信概要二、iOS 进程间通信机制详解1. URL Schemes2. Pasteboard3. App Groups 和 Shared Containers4. XPC Services 三、不同进程间通信机制的差异四、总结 一、进程间通信概要 进程间通信&am…...
高级生化大纲
一,蛋白质化学: 蛋白质分离是生物化学和分子生物学研究中的一项基本技术,用于根据蛋白质的物理和化学特性将其从混合物中分离出来。 1. 离心分离法 离心分离法利用离心力来分离不同质量或密度的颗粒和分子。 差速离心:通过逐…...
YARN WebUI 服务
一、WebUI 使用 与HDFS一样,YARN也提供了一个WebUI服务,可以使用YARN Web用户界面监视群集、队列、应用程序、服务、流活动和节点信息。还可以查看集群详细配置的信息,检查各种应用程序和服务的日志。 1.1 首页 浏览器输入http://node2.itc…...
【Unity3D】利用IJob、Burst优化处理切割物体
参考文章: 【Unity】切割网格 【Unity3D】ECS入门学习(一)导入及基础学习_unity ecs教程-CSDN博客 【Unity3D】ECS入门学习(十二)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风格 …...
基于文件系统分布式锁原理
分布式锁:在一个公共的存储服务上打上一个标记,如Redis的setnx命令,是先到先得方式获得锁,ZooKeeper有点像下面的demo,比较大小的方式判决谁获得锁。 package com.ldj.mybatisflex.demo;import java.util.*; import java.util.co…...
简历整理YH
一,订单中心 1,调拨单 融通(Rocketmq)-订单中心:ECC_BMS123(已出单),125(分配),127(发货),129(收货) 通过RocketMq接入多场景订单数据 2,销售单 sap(FTP)-订单中心,下发1002,1003,…...
Kotlin 协程基础三 —— 结构化并发(二)
Kotlin 协程基础系列: Kotlin 协程基础一 —— 总体知识概述 Kotlin 协程基础二 —— 结构化并发(一) Kotlin 协程基础三 —— 结构化并发(二) Kotlin 协程基础四 —— CoroutineScope 与 CoroutineContext Kotlin 协程…...
openclaw 配置教程:本地安装、网关接入与模型 API 配置完整说明
如果你在折腾 openclaw 配置,通常会发现真正影响使用体验的,不是把程序装上去,而是后面的模型来源怎么接、网关怎么起、控制面板怎么进,以及默认模型如何切换。只要这些环节没有理顺,就算安装完成,后续也很…...
Qwen3.5-9B-AWQ-4bit多场景应用:法律合同截图关键条款提取+风险提示生成
Qwen3.5-9B-AWQ-4bit多场景应用:法律合同截图关键条款提取风险提示生成 1. 法律合同处理的痛点与解决方案 在法律实务工作中,合同审查是一项高频且重要的工作。传统方式下,律师需要: 逐页阅读纸质或电子版合同手动标记关键条款…...
RoboStudio6.08学习记录(2)
工业机器人工作站的构建1.在文件功能选项卡中,选择“创建”,单击“创建”或“空工作站”,创建一个新的工作站,如图2-1所示。图2-1 创建新工作站2.在“基本”功能选项卡中,打开“ABB模型库”,如图2-2所示。…...
Qwen3-ForcedAligner-0.6B与LaTeX的学术工作流整合
Qwen3-ForcedAligner-0.6B与LaTeX的学术工作流整合 1. 引言 学术研究过程中,我们经常需要处理大量的访谈录音、讲座内容或实验讨论。传统的手工转录不仅耗时耗力,更让人头疼的是如何在最终论文中精准引用特定时间点的对话内容。想象一下,你…...
读懂 ABAP 调试器里的 ()XVBRP[]:这不是新语法,而是旧式内表加调试器命名表示法的组合
有朋友问我下面这个截图里的变量名是什么语法? 你这张截图里的 ()XVBRP[],结论上并不是一种新的 ABAP 变量声明语法。把它拆开看,更容易理解: XVBRP[] 这一段,核心含义是:XVBRP 是一个带 header line 的旧式内表,而 [] 明确表示你看到的是内表体 table body,不是同名的…...
智能型碧蓝航线自动化助手:AzurLaneAutoScript全方位游戏管理方案
智能型碧蓝航线自动化助手:AzurLaneAutoScript全方位游戏管理方案 【免费下载链接】AzurLaneAutoScript Azur Lane bot (CN/EN/JP/TW) 碧蓝航线脚本 | 无缝委托科研,全自动大世界 项目地址: https://gitcode.com/gh_mirrors/az/AzurLaneAutoScript …...
GHelper硬件控制工具:华硕笔记本性能优化与系统管理完全指南
GHelper硬件控制工具:华硕笔记本性能优化与系统管理完全指南 【免费下载链接】g-helper Lightweight, open-source control tool for ASUS laptops and ROG Ally. Manage performance modes, fans, GPU, battery, and RGB lighting across Zephyrus, Flow, TUF, Str…...
忍者像素绘卷镜像免配置:内置Prompt语法校验器防无效输入机制
忍者像素绘卷镜像免配置:内置Prompt语法校验器防无效输入机制 1. 产品概述 忍者像素绘卷是一款基于Z-Image-Turbo深度优化的图像生成工作站,专为像素艺术创作而设计。它融合了16-Bit复古游戏美学与现代AI图像生成技术,为用户提供了一个直观…...
javaee-网络原理2
⽹络原理-TCP/IP ①应用层:规则 → 格式 → 实际用途讲解↓ (1)定义应用之间怎么通信比如:浏览器怎么请求网页、APP 怎么跟服务器发数据。 谁先说话什么时候发请求什么时候回响应出现错误怎么办一次会话怎么开始、怎么结束 比如 HTTP 协议就明确规定&…...
Qwen3.5-2B模型环境搭建保姆级教程:从Anaconda安装到模型调用
Qwen3.5-2B模型环境搭建保姆级教程:从Anaconda安装到模型调用 1. 开篇:为什么选择这个教程? 如果你刚接触AI大模型,可能会被各种环境配置问题搞得头大。别担心,这篇教程就是为你准备的。我们将从最基础的Anaconda安装…...
