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

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确认(可选01-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 过期键的删除策略 答案 惰性删除 &#xff1a;只会在取出 key 的时候才对数据进行过期检查&#xff1b;这样对 CPU 最友好&#xff0c;但是可能会造成太多过期 key 没有被删除&#xff08;占用内存&#xff09;。 通过定时器实现&#xff08;时间事件&#xff09;&…...

边缘计算的AI小板——OrangePi AI Pro

简介 OrangePi AI Pro是一款基于Allwinner H6处理器的嵌入式AI计算设备&#xff0c;适用于物联网和边缘计算。它具有强大的性能、低功耗、多接口和小尺寸。 本文分为三个部分&#xff1a; 一、对该板进行简单的开箱介绍。 二、 将SD卡中的系统迁移到由于该板支持SD卡、SSD…...

RDMA (2)

iWARP(RDMA)怎么工作的 招式1:bypass内核 非iWARP时,当应用向网络适配器发出读或者写命令时,命令穿过用户空间以及内核空间,因此需要在用户空间和内核空间间进行切换。 iWARP使用RDMA,让应用直接将命令送达到网络适配器。这规避了对内核的调用,减少了开销和延迟。 招式2…...

vue.config.js中,devServer对象用于配置开发服务器的行为

devServer: {hot: true, // 启用模块热替换&#xff08;Hot Module Replacement&#xff0c;HMR&#xff09;。liveReload: true, // 启用页面自动刷新。当热更新失败时&#xff0c;将回退到页面自动刷新。open: true, // 启动服务器后自动打开浏览器。port: 8080, // 设置开发…...

JVM 运行流程

JVM 是 Java 运行的基础&#xff0c;也是实现一次编译到处执行的关键&#xff0c;那么 JVM 是如何执行的呢&#xff1f; JVM 执行流程 程序在执行之前先要把java代码转换成字节码&#xff08;class 文件&#xff09;&#xff0c; JVM 首先需要把字节码通过一定的 方式 类加…...

android-JNI

1.2【静态库】的特点&#xff1a; &#xff08;.a&#xff09; ①静态库对函数库的链接是在编译期完成的。执行期间代码装载速度快。 ②使可执行文件变大&#xff0c;浪费空间和资源&#xff08;占空间&#xff09;。 ③对程序的更新、部署与发布不方便&#xff0c;需要全量更新…...

Go_unsafe包

是什么&#xff1f;为什么&#xff1f; 如何利用unsafe包修改私有成员&#xff1f; 结构体会被分配到一块连续的内存&#xff0c;结构体的地址也代表第一个成员的地址。 如何利用unsafe包获取slice和map的长度&#xff1f; // 利用unsafe包修改私有成员 type S struct {name …...

【HarmonyOS4学习笔记】《HarmonyOS4+NEXT星河版入门到企业级实战教程》课程学习笔记(十三)

课程地址&#xff1a; 黑马程序员HarmonyOS4NEXT星河版入门到企业级实战教程&#xff0c;一套精通鸿蒙应用开发 &#xff08;本篇笔记对应课程第 20 - 21节&#xff09; P20《19.ArkUI-属性动画和显式动画》 本节先来学习属性动画和显式动画&#xff1a; 在代码中定义动画&am…...

企业建站响应式网站建设平台版源码系统 海量模版可选择 带完整的安装代码以及搭建教程

系统概述 企业建站响应式网站建设平台版源码系统是一款集创新性、实用性和便捷性于一体的建站解决方案。它旨在为用户提供一站式的网站建设服务&#xff0c;无论你是新手还是经验丰富的开发者&#xff0c;都能通过该系统轻松实现网站的构建与部署。 该系统采用先进的技术架构…...

在 VSCode 中搭建 Flutter 开发环境并运行项目

要在 Visual Studio Code (VSCode) 中运行 Flutter 项目并启动虚拟机&#xff08;例如 Android Emulator&#xff09;&#xff0c;可以按照以下步骤进行设置和操作&#xff1a; 一、安装 Flutter 和 Dart 插件 安装 Flutter SDK&#xff1a; 前往 Flutter 官网 下载并安装 Flu…...

如何执行VMware P2V迁移|VMware Converter和替代方案

VMware中的P2V是什么&#xff1f; 我们常说的VMware P2V其实指的就是“物理到虚拟”&#xff0c;将工作负载从物理机器转换或迁移到虚拟机&#xff08;VM&#xff09;的过程&#xff0c;能够使您无需从头开始费力地创建和配置新虚拟机。 就像您可以使用Disk2vhd执行Hyper-V物理…...

03-3.2.3 队列的链式存储的实现

&#x1f44b; Hi, I’m Beast Cheng&#x1f440; I’m interested in photography, hiking, landscape…&#x1f331; I’m currently learning python, javascript, kotlin…&#x1f4eb; How to reach me --> 458290771qq.com 喜欢《数据结构》部分笔记的小伙伴可以订…...

Spring AI 第二讲 之 Chat Model API 第八节Anthropic 3 Chat

Anthropic Claude 是一系列基础人工智能模型&#xff0c;可用于各种应用。对于开发人员和企业来说&#xff0c;您可以利用 API 访问&#xff0c;直接在 Anthropic 的人工智能基础架构之上进行构建。 Spring AI 支持用于同步和流式文本生成的 Anthropic 消息 API。 Anthropic …...

【ARM 常见汇编指令学习 6.2 -- ARMv8 汇编指令 SDIV 详细介绍】

文章目录 SDIV指令格式使用示例注意事项总结 SDIV ARMv8 架构中的 SDIV 指令用于执行带符号整数除法操作。这意味着它可以处理负数除法&#xff0c;与 UDIV&#xff08;执行无符号整数除法&#xff09;形成对比。SDIV 将两个寄存器中的带符号整数相除&#xff0c;将除法结果存…...

【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连不上无法下载模型&#xff0c;需要设置镜像。…...

Winddow系统下关于Golang使用Cgo的配置

1.配置CGO_ENABLED为1 go env -w CGO_ENABLED1 2.安装gcc环境&#xff0c;否则出现cgo: C compiler "gcc" not found: exec: "gcc": executable file not found in %PATH%错误 安装包&#xff1a;链接&#xff1a;https://pan.baidu.com/s/1sgF9lijqGeP…...

python面向过程与初始面向对象编程

让我们穿越到《龙珠》世界&#xff0c;一起揭开 面向对象编程 的神秘面纱吧。 面向过程编程与面向对象编程 天下第一武道会 选手登记 第 22 届天下第一武道会即将召开&#xff0c;各路武术高手齐聚一堂&#xff0c;其中最受瞩目的&#xff0c;当属卡卡罗特&#xff08;孙悟…...

vue3 实现自定义指令封装 --- 通俗易懂

1、局部自定义指令 1.1 在<script setup>定义组件内的指令&#xff0c;任何以v开头的驼峰式命名的变量都可以被用作一个自定义指令 <template><div><h3>使用自定义指令</h3><div>########################## start 局部自定义指令</d…...

label-studio的使用教程(导入本地路径)

文章目录 1. 准备环境2. 脚本启动2.1 Windows2.2 Linux 3. 安装label-studio机器学习后端3.1 pip安装(推荐)3.2 GitHub仓库安装 4. 后端配置4.1 yolo环境4.2 引入后端模型4.3 修改脚本4.4 启动后端 5. 标注工程5.1 创建工程5.2 配置图片路径5.3 配置工程类型标签5.4 配置模型5.…...

python/java环境配置

环境变量放一起 python&#xff1a; 1.首先下载Python Python下载地址&#xff1a;Download Python | Python.org downloads ---windows -- 64 2.安装Python 下面两个&#xff0c;然后自定义&#xff0c;全选 可以把前4个选上 3.环境配置 1&#xff09;搜高级系统设置 2…...

【磁盘】每天掌握一个Linux命令 - iostat

目录 【磁盘】每天掌握一个Linux命令 - iostat工具概述安装方式核心功能基础用法进阶操作实战案例面试题场景生产场景 注意事项 【磁盘】每天掌握一个Linux命令 - iostat 工具概述 iostat&#xff08;I/O Statistics&#xff09;是Linux系统下用于监视系统输入输出设备和CPU使…...

Auto-Coder使用GPT-4o完成:在用TabPFN这个模型构建一个预测未来3天涨跌的分类任务

通过akshare库&#xff0c;获取股票数据&#xff0c;并生成TabPFN这个模型 可以识别、处理的格式&#xff0c;写一个完整的预处理示例&#xff0c;并构建一个预测未来 3 天股价涨跌的分类任务 用TabPFN这个模型构建一个预测未来 3 天股价涨跌的分类任务&#xff0c;进行预测并输…...

vue3+vite项目中使用.env文件环境变量方法

vue3vite项目中使用.env文件环境变量方法 .env文件作用命名规则常用的配置项示例使用方法注意事项在vite.config.js文件中读取环境变量方法 .env文件作用 .env 文件用于定义环境变量&#xff0c;这些变量可以在项目中通过 import.meta.env 进行访问。Vite 会自动加载这些环境变…...

初探Service服务发现机制

1.Service简介 Service是将运行在一组Pod上的应用程序发布为网络服务的抽象方法。 主要功能&#xff1a;服务发现和负载均衡。 Service类型的包括ClusterIP类型、NodePort类型、LoadBalancer类型、ExternalName类型 2.Endpoints简介 Endpoints是一种Kubernetes资源&#xf…...

C++ 设计模式 《小明的奶茶加料风波》

&#x1f468;‍&#x1f393; 模式名称&#xff1a;装饰器模式&#xff08;Decorator Pattern&#xff09; &#x1f466; 小明最近上线了校园奶茶配送功能&#xff0c;业务火爆&#xff0c;大家都在加料&#xff1a; 有的同学要加波霸 &#x1f7e4;&#xff0c;有的要加椰果…...

【Elasticsearch】Elasticsearch 在大数据生态圈的地位 实践经验

Elasticsearch 在大数据生态圈的地位 & 实践经验 1.Elasticsearch 的优势1.1 Elasticsearch 解决的核心问题1.1.1 传统方案的短板1.1.2 Elasticsearch 的解决方案 1.2 与大数据组件的对比优势1.3 关键优势技术支撑1.4 Elasticsearch 的竞品1.4.1 全文搜索领域1.4.2 日志分析…...

Oracle11g安装包

Oracle 11g安装包 适用于windows系统&#xff0c;64位 下载路径 oracle 11g 安装包...

Spring Security 认证流程——补充

一、认证流程概述 Spring Security 的认证流程基于 过滤器链&#xff08;Filter Chain&#xff09;&#xff0c;核心组件包括 UsernamePasswordAuthenticationFilter、AuthenticationManager、UserDetailsService 等。整个流程可分为以下步骤&#xff1a; 用户提交登录请求拦…...