kafka-生产者拦截器(SpringBoot整合Kafka)
文章目录
- 1、生产者拦截器
- 1.1、创建生产者拦截器
- 1.2、KafkaTemplate配置生产者拦截器
- 1.3、使用Java代码创建主题分区副本
- 1.4、application.yml配置----v1版
- 1.5、屏蔽 kafka debug 日志 logback.xml
- 1.6、引入spring-kafka依赖
- 1.7、控制台日志
1、生产者拦截器
1.1、创建生产者拦截器
package com.atguigu.kafka.interceptor;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.stereotype.Component;
import java.util.Map;
//拦截器必须手动注册给kafka生产者(KafkaTemplate)
@Component
public class MyKafkaInterceptor implements ProducerInterceptor<String,String> {//kafka生产者发送消息前执行:拦截发送的消息预处理@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {System.out.println("生产者即将发送消息:topic = "+ producerRecord.topic()+",partition:"+producerRecord.partition()+",key = "+producerRecord.key()+",value = "+producerRecord.value());return null;}//kafka broker 给出应答后执行@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {//exception为空表示消息发送成功if(e == null){System.out.println("消息发送成功:topic = "+ recordMetadata.topic()+",partition:"+recordMetadata.partition()+",offset="+recordMetadata.offset()+",timestamp="+recordMetadata.timestamp());}}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> map) {}
}
1.2、KafkaTemplate配置生产者拦截器
package com.atguigu.kafka.producer;import com.atguigu.kafka.interceptor.MyKafkaInterceptor;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import java.io.IOException;@SpringBootTest
class KafkaProducerApplicationTests {//装配kafka模板类: springboot启动时会自动根据配置文初始化kafka模板类对象注入到容器中@ResourceKafkaTemplate kafkaTemplate;@ResourceMyKafkaInterceptor myKafkaInterceptor;@PostConstructpublic void init() {kafkaTemplate.setProducerInterceptor(myKafkaInterceptor);}@Testvoid contextLoads() throws IOException {kafkaTemplate.send("my_topic1", "spring-kafka-生产者拦截器");//回调是等kafka,ack以后才执行,需要阻塞System.in.read();}
}
1.3、使用Java代码创建主题分区副本
package com.atguigu.kafka.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.stereotype.Component;
@Component
public class KafkaTopicConfig {@Beanpublic NewTopic myTopic1() {//相同名称的主题 只会创建一次,后面创建的主题名称相同配置不同可以做增量更新(分区、副本数)return TopicBuilder.name("my_topic1")//主题名称.partitions(3)//主题分区.replicas(3)//主题分区副本数.build();//创建}
}
1.4、application.yml配置----v1版
server:port: 8110# v1
spring:kafka:bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097producer: # producer 生产者retries: 0 # 重试次数 0表示不重试acks: -1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、-1/all)batch-size: 16384 # 批次大小 单位bytebuffer-memory: 33554432 # 生产者缓冲区大小 单位bytekey-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器
1.5、屏蔽 kafka debug 日志 logback.xml
<configuration> <!-- 如果觉得idea控制台日志太多,src\main\resources目录下新建logback.xml
屏蔽kafka debug --><logger name="org.apache.kafka.clients" level="debug" />
</configuration>
1.6、引入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.kafka</groupId><artifactId>kafka-producer</artifactId><version>0.0.1-SNAPSHOT</version><name>kafka-producer</name><description>kafka-producer</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.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.7、控制台日志
生产者即将发送消息:topic = my_topic1,partition:null,key = null,value = spring-kafka-生产者拦截器
消息发送成功:topic = my_topic1,partition:0,offset=0,timestamp=1717490776329
[[{"partition": 0,"offset": 0,"msg": "spring-kafka-生产者拦截器","timespan": 1717490776329,"date": "2024-06-04 08:46:16"}]
]

相关文章:
kafka-生产者拦截器(SpringBoot整合Kafka)
文章目录 1、生产者拦截器1.1、创建生产者拦截器1.2、KafkaTemplate配置生产者拦截器1.3、使用Java代码创建主题分区副本1.4、application.yml配置----v1版1.5、屏蔽 kafka debug 日志 logback.xml1.6、引入spring-kafka依赖1.7、控制台日志 1、生产者拦截器 1.1、创建生产者拦…...
每日一题:聊聊 Redis 过期键的删除策略
聊聊 Redis 过期键的删除策略 答案 惰性删除 :只会在取出 key 的时候才对数据进行过期检查;这样对 CPU 最友好,但是可能会造成太多过期 key 没有被删除(占用内存)。 通过定时器实现(时间事件)&…...
边缘计算的AI小板——OrangePi AI Pro
简介 OrangePi AI Pro是一款基于Allwinner H6处理器的嵌入式AI计算设备,适用于物联网和边缘计算。它具有强大的性能、低功耗、多接口和小尺寸。 本文分为三个部分: 一、对该板进行简单的开箱介绍。 二、 将SD卡中的系统迁移到由于该板支持SD卡、SSD…...
RDMA (2)
iWARP(RDMA)怎么工作的 招式1:bypass内核 非iWARP时,当应用向网络适配器发出读或者写命令时,命令穿过用户空间以及内核空间,因此需要在用户空间和内核空间间进行切换。 iWARP使用RDMA,让应用直接将命令送达到网络适配器。这规避了对内核的调用,减少了开销和延迟。 招式2…...
vue.config.js中,devServer对象用于配置开发服务器的行为
devServer: {hot: true, // 启用模块热替换(Hot Module Replacement,HMR)。liveReload: true, // 启用页面自动刷新。当热更新失败时,将回退到页面自动刷新。open: true, // 启动服务器后自动打开浏览器。port: 8080, // 设置开发…...
JVM 运行流程
JVM 是 Java 运行的基础,也是实现一次编译到处执行的关键,那么 JVM 是如何执行的呢? JVM 执行流程 程序在执行之前先要把java代码转换成字节码(class 文件), JVM 首先需要把字节码通过一定的 方式 类加…...
android-JNI
1.2【静态库】的特点: (.a) ①静态库对函数库的链接是在编译期完成的。执行期间代码装载速度快。 ②使可执行文件变大,浪费空间和资源(占空间)。 ③对程序的更新、部署与发布不方便,需要全量更新…...
Go_unsafe包
是什么?为什么? 如何利用unsafe包修改私有成员? 结构体会被分配到一块连续的内存,结构体的地址也代表第一个成员的地址。 如何利用unsafe包获取slice和map的长度? // 利用unsafe包修改私有成员 type S struct {name …...
【HarmonyOS4学习笔记】《HarmonyOS4+NEXT星河版入门到企业级实战教程》课程学习笔记(十三)
课程地址: 黑马程序员HarmonyOS4NEXT星河版入门到企业级实战教程,一套精通鸿蒙应用开发 (本篇笔记对应课程第 20 - 21节) P20《19.ArkUI-属性动画和显式动画》 本节先来学习属性动画和显式动画: 在代码中定义动画&am…...
企业建站响应式网站建设平台版源码系统 海量模版可选择 带完整的安装代码以及搭建教程
系统概述 企业建站响应式网站建设平台版源码系统是一款集创新性、实用性和便捷性于一体的建站解决方案。它旨在为用户提供一站式的网站建设服务,无论你是新手还是经验丰富的开发者,都能通过该系统轻松实现网站的构建与部署。 该系统采用先进的技术架构…...
在 VSCode 中搭建 Flutter 开发环境并运行项目
要在 Visual Studio Code (VSCode) 中运行 Flutter 项目并启动虚拟机(例如 Android Emulator),可以按照以下步骤进行设置和操作: 一、安装 Flutter 和 Dart 插件 安装 Flutter SDK: 前往 Flutter 官网 下载并安装 Flu…...
如何执行VMware P2V迁移|VMware Converter和替代方案
VMware中的P2V是什么? 我们常说的VMware P2V其实指的就是“物理到虚拟”,将工作负载从物理机器转换或迁移到虚拟机(VM)的过程,能够使您无需从头开始费力地创建和配置新虚拟机。 就像您可以使用Disk2vhd执行Hyper-V物理…...
03-3.2.3 队列的链式存储的实现
👋 Hi, I’m Beast Cheng👀 I’m interested in photography, hiking, landscape…🌱 I’m currently learning python, javascript, kotlin…📫 How to reach me --> 458290771qq.com 喜欢《数据结构》部分笔记的小伙伴可以订…...
Spring AI 第二讲 之 Chat Model API 第八节Anthropic 3 Chat
Anthropic Claude 是一系列基础人工智能模型,可用于各种应用。对于开发人员和企业来说,您可以利用 API 访问,直接在 Anthropic 的人工智能基础架构之上进行构建。 Spring AI 支持用于同步和流式文本生成的 Anthropic 消息 API。 Anthropic …...
【ARM 常见汇编指令学习 6.2 -- ARMv8 汇编指令 SDIV 详细介绍】
文章目录 SDIV指令格式使用示例注意事项总结 SDIV ARMv8 架构中的 SDIV 指令用于执行带符号整数除法操作。这意味着它可以处理负数除法,与 UDIV(执行无符号整数除法)形成对比。SDIV 将两个寄存器中的带符号整数相除,将除法结果存…...
【ArcGIS微课1000例】0113:大地测量要素概述与构建
文章目录 一、大地测量要素描述1. 大地要素的概念2. 大地要素的类型二、创建大地测量要素1. 创建要素类2. 创建大地要素一、大地测量要素描述 1. 大地要素的概念 大地测量要素的测量值考虑了投影空间的固有变形。如果要创建一个空间跨度较大的要素(例如一条横跨大洋的飞行路…...
【记录】LangChain+本地模型的文档问答(webUI)
已在notebook测试无误。 包安装 pip install langchain langchain_community transformers InstructorEmbedding sentence_transformers2.2.2 faiss-gpu PyPDF2 streamlit pyngrok gradio fitz frontend 环境变量设置 huggingface连不上无法下载模型,需要设置镜像。…...
Winddow系统下关于Golang使用Cgo的配置
1.配置CGO_ENABLED为1 go env -w CGO_ENABLED1 2.安装gcc环境,否则出现cgo: C compiler "gcc" not found: exec: "gcc": executable file not found in %PATH%错误 安装包:链接:https://pan.baidu.com/s/1sgF9lijqGeP…...
python面向过程与初始面向对象编程
让我们穿越到《龙珠》世界,一起揭开 面向对象编程 的神秘面纱吧。 面向过程编程与面向对象编程 天下第一武道会 选手登记 第 22 届天下第一武道会即将召开,各路武术高手齐聚一堂,其中最受瞩目的,当属卡卡罗特(孙悟…...
vue3 实现自定义指令封装 --- 通俗易懂
1、局部自定义指令 1.1 在<script setup>定义组件内的指令,任何以v开头的驼峰式命名的变量都可以被用作一个自定义指令 <template><div><h3>使用自定义指令</h3><div>########################## start 局部自定义指令</d…...
UE5 学习系列(二)用户操作界面及介绍
这篇博客是 UE5 学习系列博客的第二篇,在第一篇的基础上展开这篇内容。博客参考的 B 站视频资料和第一篇的链接如下: 【Note】:如果你已经完成安装等操作,可以只执行第一篇博客中 2. 新建一个空白游戏项目 章节操作,重…...
ubuntu搭建nfs服务centos挂载访问
在Ubuntu上设置NFS服务器 在Ubuntu上,你可以使用apt包管理器来安装NFS服务器。打开终端并运行: sudo apt update sudo apt install nfs-kernel-server创建共享目录 创建一个目录用于共享,例如/shared: sudo mkdir /shared sud…...
Xshell远程连接Kali(默认 | 私钥)Note版
前言:xshell远程连接,私钥连接和常规默认连接 任务一 开启ssh服务 service ssh status //查看ssh服务状态 service ssh start //开启ssh服务 update-rc.d ssh enable //开启自启动ssh服务 任务二 修改配置文件 vi /etc/ssh/ssh_config //第一…...
shell脚本--常见案例
1、自动备份文件或目录 2、批量重命名文件 3、查找并删除指定名称的文件: 4、批量删除文件 5、查找并替换文件内容 6、批量创建文件 7、创建文件夹并移动文件 8、在文件夹中查找文件...
vscode(仍待补充)
写于2025 6.9 主包将加入vscode这个更权威的圈子 vscode的基本使用 侧边栏 vscode还能连接ssh? debug时使用的launch文件 1.task.json {"tasks": [{"type": "cppbuild","label": "C/C: gcc.exe 生成活动文件"…...
【第二十一章 SDIO接口(SDIO)】
第二十一章 SDIO接口 目录 第二十一章 SDIO接口(SDIO) 1 SDIO 主要功能 2 SDIO 总线拓扑 3 SDIO 功能描述 3.1 SDIO 适配器 3.2 SDIOAHB 接口 4 卡功能描述 4.1 卡识别模式 4.2 卡复位 4.3 操作电压范围确认 4.4 卡识别过程 4.5 写数据块 4.6 读数据块 4.7 数据流…...
Auto-Coder使用GPT-4o完成:在用TabPFN这个模型构建一个预测未来3天涨跌的分类任务
通过akshare库,获取股票数据,并生成TabPFN这个模型 可以识别、处理的格式,写一个完整的预处理示例,并构建一个预测未来 3 天股价涨跌的分类任务 用TabPFN这个模型构建一个预测未来 3 天股价涨跌的分类任务,进行预测并输…...
华硕a豆14 Air香氛版,美学与科技的馨香融合
在快节奏的现代生活中,我们渴望一个能激发创想、愉悦感官的工作与生活伙伴,它不仅是冰冷的科技工具,更能触动我们内心深处的细腻情感。正是在这样的期许下,华硕a豆14 Air香氛版翩然而至,它以一种前所未有的方式&#x…...
C#中的CLR属性、依赖属性与附加属性
CLR属性的主要特征 封装性: 隐藏字段的实现细节 提供对字段的受控访问 访问控制: 可单独设置get/set访问器的可见性 可创建只读或只写属性 计算属性: 可以在getter中执行计算逻辑 不需要直接对应一个字段 验证逻辑: 可以…...
【Nginx】使用 Nginx+Lua 实现基于 IP 的访问频率限制
使用 NginxLua 实现基于 IP 的访问频率限制 在高并发场景下,限制某个 IP 的访问频率是非常重要的,可以有效防止恶意攻击或错误配置导致的服务宕机。以下是一个详细的实现方案,使用 Nginx 和 Lua 脚本结合 Redis 来实现基于 IP 的访问频率限制…...
