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

如何使用@KafkaListener实现从nacos中动态获取监听的topic

news 来源：原创 2025/6/23 16:03:32

1、简介

对于经常需要变更kafka主题的场景，为了实现动态监听topic的功能，可以使用以下方式。

2、使用步骤

2.1、添加依赖

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

2.2、nacos中配置

    # kafka 配置
spring:kafka:bootstrap-servers: ip地址:9092topics: topic1,tpic2producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerenable-idempotence: trueacks: alltransactional-id: kafka-groupconsumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: kafka-clickhouse-groupauto-offset-reset: latestenable-auto-commit: falseisolation-level: read_committedallow-auto-create-topics: truelistener:ack-mode: MANUAL_IMMEDIATEconcurrency: 3

2.3、配置类

package org.aecsi.kafkadatatock.config;import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.transaction.KafkaTransactionManager;import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;@Configuration
@RequiredArgsConstructor
@EnableKafka
@RefreshScope
public class KafkaConfig {private final KafkaProperties kafkaProperties;@Beanpublic KafkaAdmin kafkaAdmin() {return new KafkaAdmin(kafkaProperties.buildAdminProperties());}@Beanpublic AdminClient adminClient(KafkaAdmin kafkaAdmin) {return AdminClient.create(kafkaAdmin.getConfigurationProperties());}@Beanpublic ProducerFactory<String, String> producerFactory() {Map<String, Object> configProps = new HashMap<>(kafkaProperties.buildProducerProperties());configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);configProps.put(ProducerConfig.ACKS_CONFIG, "all");configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "kafka-clickhouse-producer");DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(configProps);factory.setTransactionIdPrefix("kafka-clickhouse-producer-");return factory;}@Beanpublic KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {return new KafkaTemplate<>(producerFactory);}@Bean@RefreshScopepublic ConsumerFactory<String, String> consumerFactory() {Map<String, Object> configProps = new HashMap<>(kafkaProperties.buildConsumerProperties());configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);configProps.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, true);configProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");return new DefaultKafkaConsumerFactory<>(configProps);}@Beanpublic KafkaTransactionManager<String, String> transactionManager(ProducerFactory<String, String> producerFactory) {return new KafkaTransactionManager<>(producerFactory);}@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory,KafkaTransactionManager<String, String> transactionManager) {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);factory.getContainerProperties().setTransactionManager(transactionManager);return factory;}@Bean@RefreshScopepublic ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());factory.setAutoStartup(true);return factory;}@Beanpublic ApplicationRunner kafkaListenerStarter(KafkaListenerEndpointRegistry registry) {return args -> {// 启动所有 Kafka 监听器registry.start();};}
}

接收消息类

@KafkaListener(topics = "#{'${spring.kafka.topics}'.split(',')}", autoStartup = "false")@Transactional(transactionManager = "transactionManager")public void processMessage(ConsumerRecord<String, String> record,Acknowledgment acknowledgment,@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,@Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp) {try {log.info("kafka 接受 topic: {} 消息", topic);
//          处理消息acknowledgment.acknowledge();} catch (Exception e) {log.error("Error processing message for topic {}: {}", topic, e.getMessage());throw e;}}

主启动类添加一个注解

@EnableConfigurationProperties(KafkaProperties.class)

3、总结

实现kafka动态获取topic还有其他方式，博主目前只验证这一种，其他方式待更新。

如何使用@KafkaListener实现从nacos中动态获取监听的topic

1、简介对于经常需要变更kafka主题的场景，为了实现动态监听topic的功能，可以使用以下方式。 2、使用步骤 2.1、添加依赖 <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactI…...

编程日记 2025/6/23 16:03:32

区块链如何达成共识：PoW/PoS/DPoS的原理、争议与适用场景全解

引言区块链技术的核心在于去中心化网络中的信任机制，而共识算法是实现这一目标的关键。不同的共识机制在效率、安全性、去中心化程度等方面各有取舍。本文将深入解析三种主流共识机制——工作量证明（PoW）、权益证明&am…...

编程日记 2025/6/4 4:14:48

Oracle for Linux安装和配置（11）——Oracle安装和配置

11.3. Oracle安装和配置 Linux上Oracle的安装及配置与Windows上差不多，只是安装软件的准备等有所不同，下面只对不同于Windows的部分进行较为详细的讲解，其他类似部分不再赘述。另外，无论选择使用虚机还是物理机，Oracle安装、配置和使用等方面几乎都是完全一样的。 11.3.…...

编程日记 2025/6/22 14:14:25

http协议详解附带c/c++服务器和客户端源码

http详解代码部分HTTP请求结构示例HTTP客户端实现（使用Linux系统调用）简易HTTP服务器实现深入解析HTTP协议：从基础到实践1. HTTP协议核心概念1.1 协议本质解读1.2 通信模型详解 2. 抓包分析实战2.1 Fiddler工具妙用2.2 报文结构拆解 3. 请求…...

编程日记 2025/6/20 20:26:20

JavaScript性能优化实战（5）：数据结构与算法性能优化

JavaScript中常用数据结构性能对比数据结构的选择对JavaScript应用的性能有着决定性的影响。不同的数据结构在不同操作上各有优劣，选择合适的数据结构能显著提升应用性能。本节将对JavaScript中常用的数据结构进行全面的性能对比分析。基本数据结构时间复杂度概览首先，…...

编程日记 2025/6/23 15:58:02

uniapp小程序开发入门01-快速搭建一个空白的项目并预览它

uniapp小程序开发入门01-快速搭建一个空白的项目并预览它！由于近期有市场需求和计划，构建一套自己的小程序，所以再次带领大家系统的过一遍，如何使用uniapp程序快速构建一套完整的项目。今天是第一小节，带领大家快速构建…...

编程日记 2025/5/30 22:34:16

UR5 UR5e机器人URDF文件

URDF全称为Unified Robot Description Format，中文可以翻译为“统一机器人描述格式”。与计算机文件中的.txt文本格式、.jpg图像格式等类似，URDF是一种基于XML规范、用于描述机器人结构的格式。根据该格式的设计者所言，设计这一格式的目的在于提供一种尽可能通用（as genera…...

编程日记 2025/6/23 15:51:18

ubuntu20.04安装x11vnc远程桌面

x11vnc是一个VNC服务器, 安装后我们可以不依赖外部的显示设备, 通过网络远程登录ubuntu桌面。安装x11vnc sudo apt-get install x11vnc 设置VNC登录密码 sudo x11vnc -storepasswd /etc/x11vnc.pwd 设置x11vnc在开机时自动启动新建如下文件: sudo vi /lib/systemd/sys…...

编程日记 2025/6/23 15:53:54

AKM旭化成微电子全新推出能量收集IC“AP4413系列”

旭化成微电子开始批量生产用于环保发电的电荷控制集成电路！优化充电电池的充放电，广泛应用于智能遥控器和蓝牙TMTag等设备。 01 概述旭化成微电子株式会社（AKM）开发出面向小型二次电池（充电电池）的环境…...

编程日记 2025/6/23 15:31:38

机器人行业研究系列报告

新质生产力系列报告：2024年人形机器人核心场景发展洞察研究报告具身机器人行业现状及未来趋势分析 2025 2025年人形机器人投资策略，量产元年，全球共振，百家争鸣人形机器人行业深度报告（一）&#xff1a…...

编程日记 2025/6/1 1:48:38

利用JMeter代理服务器方式实现高效压测

前言在当今快节奏的互联网时代，确保Web应用和服务能够在高负载下稳定运行变得至关重要。无论是电子商务平台、社交媒体网络还是在线教育服务，用户对网站响应速度和稳定性的期望从未如此之高。因此，性能测试不再是一个可选项，而是…...

编程日记 2025/6/20 5:29:43

NLP高频面试题（五十五）——DeepSeek系列概览与发展背景

大型模型浪潮背景近年来，大型语言模型（Large Language Model, LLM）领域发展迅猛，从GPT-3等超大规模模型的崛起到ChatGPT的横空出世，再到GPT-4的问世，模型参数规模和训练数据量呈指数级增长。以GPT-3为例，参数高达1750亿，在570GB文本数据上训练，显示出模型规模、数据…...

编程日记 2025/6/22 1:33:50

2015-2023 各省 GDP 数据，用QuickBI 进行数据可视化——堆叠图！

嘿，数据爱好者们！今天咱要来一场刺激的数据冒险，深入剖析全国各省的 GDP 数据，而且会借助强大的 QuickBI 工具，用超酷炫的堆叠图让这些数据 “活” 起来，带你一眼看清经济格局！ 地址&#xff1…...

编程日记 2025/6/21 5:45:27

MySQL优化（持续更新）笔记

一、insert优化 ： 之前：项目通常是一条insert一条的执行，每一次都需要与MySQL进行建立连接进行网络传输，效率很低现在： 1.- 批量插入（一条sql就行，一次500-1000） 可以与MyBatis…...

编程日记 2025/5/30 3:49:43

目录 1. 表的创建2. 表的查看3. 表的修改4. 表的删除5. 总结 1. 表的创建 1.查看字符集及效验规则 2. 表的创建 CREATE TABLE table_name ( field1 datatype, field2 datatype, field3 datatype ) character set 字符集 collate 校验规则 engine 存储引擎;创建用户表1 创建用…...

编程日记 2025/5/29 9:39:13

Java 数组：深度解析

前言数组作为Java中最基础也是最强大的数据结构之一，其高效性和灵活性在性能关键型应用中无可替代。本文将从进阶使用开始，逐步深入探索Java数组的高级特性和大师级技巧，帮助开发者全面掌握数组技术的精髓。一、数组基础回顾与性能特性 1.1 数组基本特性对比特性Java数…...

编程日记 2025/6/23 15:57:28

【基于Qt的QQMusic项目演示第一章】从界面交互到核心功能实现

🌹 作者: 云小逸 🤟 个人主页: 云小逸的主页 🤟 motto: 要敢于一个人默默的面对自己，强大自己才是核心。不要等到什么都没有了，才下定决心去做。种一颗树，最好的时间是十年前，其次就是现在&…...

编程日记 2025/6/19 20:20:10

[Mybatis-plus]

简介 MyBatis-Plus （简称 MP）是一个 MyBatis的增强工具，在 MyBatis 的基础上只做增强不做改变。Mybatis-plus官网地址注意，在引入了mybatis-plus之后，不要再额外引入mybatis和mybatis-spring，避免因为版本…...

编程日记 2025/6/20 5:01:03

【EDA】EDA中聚类（Clustering）和划分（Partitioning）的应用场景

在VLSI物理设计自动化中，聚类（Clustering）和划分（Partitioning）是两个互补但目标和应用场景截然不同的关键步骤，其核心区别如下： 一、应用阶段与核心目标 1. 聚类（Clustering&…...

编程日记 2025/5/28 16:04:49

PySide与PyQt对比：为何PySide是更优选择

PySide与PyQt对比：为何PySide是更优选择引言在Python桌面应用开发领域，Qt框架的绑定库一直是首选方案。两大主要选择—PySide和PyQt，虽然功能相似，但在许可证、性能和支持方面存在显著差异。本文将深入探讨为何PySide通常是更…...

编程日记 2025/5/30 7:04:41

LVGL移植高通矢量字库GT5SLAD3BFA

字库芯片: GT5SLAD3BFA MCU: STM32F429 LVGL版本：V8.4 一，实现gt_read_data() gt_read_data()函数的作用：与字库flash进行通信，函数的定义里调用spi发送数据和接收数据的接口。用户只需要实现该函数，就可以…...

编程日记 2025/6/23 7:20:09

7.0 sharpScada的sql数据的安装

本文介绍开源库SharpScada的配置过程。 1，还原数据库 2.打开SQL server2014配置启动器，并启用Named Pipes,以及TCP/IP 3.启动SQL Server服务中的SQL Server Browser 4.允许远程连接...

编程日记 2025/5/29 18:05:27

杂项知识点

杂项 1 激活函数1.1 sigmoid1.2 tanh1.3 Relu1.4 leakRelu 1 激活函数常用的激活函数包括sigmoid tanh Relu leakRelu 1.1 sigmoid import torch import numpy as np import matplotlib.pyplot as plt # sigmoid tanh Relu leakRelu ## 1 sigmoid ### 1.1 代码复现sig…...

编程日记 2025/6/19 20:51:23

Android项目升级插件到kotlin 2.1.0后混淆网络请求异常

背景项目kt插件1.9.24升级到2.1.0后打包编译release网络请求失败了。 retrofit版本2.9.0 错误详情 java.lang.ClassCastException: java.lang.Class cannot be cast to java.lang.reflect.ParameterizedTypeat retrofit2.m.a(Unknown Source:2477)at retrofit2.K.invoke(U…...

编程日记 2025/6/20 16:49:35

uniapp 仿企微左边公司切换页

示例代码： <template><view class"container"><view class"mask" v-if"showSidebar" click"closeSidebar"></view><view class"sidebar"…...

编程日记 2025/6/1 17:51:54

Milvus（7）：Schema、主字段和自动识别

1 Schema Schema 定义了 Collections 的数据结构。在创建一个 Collection 之前，你需要设计出它的 Schema。本页将帮助你理解 Collections 模式，并自行设计一个示例模式。在 Zilliz Cloud 上，Collection Schema 是关系数据库中一个表的组合&a…...

编程日记 2025/6/20 17:38:18

Liunx服务上MySQL服务导致CPU炸了，使用kill -9 mysqld进程id后，无法启动MySQL

1.top命令后，可以看到mysqld沾满了cpu 2.然后我使用了kill -9 16594，杀死了mysqld进程 3.之后，查看mysql服务状态，发现对应的 www/serve/mysqld 目录不存在 sudo systemctl status mysqld4.使用命令查看操作 www/serve 目录的历…...

编程日记 2025/6/23 10:11:15

Java使用IText7动态生成带审批文本框的PDF文档

Java使用IText7动态生成带审批文本框的PDF文档文章目录 Java使用IText7动态生成带审批文本框的PDF文档1.构建第一个框的起始坐标2.渲染第一个框3.渲染其他的审批框测试结果示例实现思路使用Canvas进行相对定位和绝对定位来确定文本框内文字位置，用Rectangle通…...

编程日记 2025/5/29 13:36:07

【音视频】AVIO输入模式

内存IO模式 AVIOContext *avio_alloc_context( unsigned char *buffer, int buffer_size, int write_flag, void *opaque, int (*read_packet)(void *opaque, uint8_t *buf, int buf_size), int (*write_packet)(void *opaque, uint8_t *buf, int buf_size), int64_t (*seek)(…...

编程日记 2025/6/20 9:34:46