SpringBoot集成Kafka之SASL_SSL

2024-04-15 14:20

本文主要是介绍SpringBoot集成Kafka之SASL_SSL,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

系列文章目录


文章目录

  • 系列文章目录
  • 前言


前言

前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站,这篇文章男女通用,看懂了就去分享给你的码吧。
在这里插入图片描述


Kafka 是一个由 LinkedIn 开发的分布式消息系统,它于2011年年初开源,现在由著名的 Apache 基金会维护与开发。 Kafka 使用 Scala 实现,被用作 LinkedIn 的活动流和运营数据处理的管道,现在也被诸多互联网企业广泛地用作数据流管道和消息系统, Kafka 是基于消息发布﹣订阅模式实现的消息系统。
在这里插入图片描述
Win安装:Kafka安装Windows版 - Java小强技术博客 (javacui.com)

SpringBoot集成:SpringBoot集成Kafka - Java小强技术博客 (javacui.com)

由于公司使用时,必须要求Prodcuer&Consumer must connect to Kafka via SASL_SSL,因为进行了相关集成的编码,进行记录。

首先之前Spring的配置文件我们就不需要了,为了测试方便先编码到代码中,手续可以手动指定到配置文件中。

这里需要IO包,来操作JKS文件,看其是否存在,因此POM需要引入

<dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.11.0</version>
</dependency>

然后消息监听类,需要改造下,首先指定一个ID,然后指定为手动启动

package com.example.springboot.listener;import com.alibaba.fastjson.JSON;
import com.example.springboot.model.Blog;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;@Component
public class KafkaMessListener {private static final String messTopic = "test";@KafkaListener(id="KafkaMessListener", topics = messTopic, groupId = "javagroup", autoStartup = "false")public void messListener(Blog blog) {System.out.println("消费端 收到消息:" + JSON.toJSONString(blog));}
}

编写消息提供者Producer配置类

package com.example.springboot.config;import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.boot.system.ApplicationHome;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;/*** Kafka服务端配置*/
@Slf4j
@Configuration
public class KafkaProducerConfig {@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}/*** the producer factory config*/@Beanpublic ProducerFactory<String, String> producerFactory() {Map<String, Object> props = new HashMap();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9002");props.put(ProducerConfig.ACKS_CONFIG, "all");props.put(ProducerConfig.RETRIES_CONFIG, 3);props.put(ProducerConfig.BATCH_SIZE_CONFIG, 106384);props.put(ProducerConfig.LINGER_MS_CONFIG, 1);props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonSerializer");props.put("security.protocol", "SASL_SSL");props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");//打包成jar后,无法找到文件的解决方法(复制jar包里的文件,到文件目录里)ApplicationHome applicationHome = new ApplicationHome(KafkaProducerConfig.class);//项目打包成jar包所在的根路径String rootPath = applicationHome.getSource().getParentFile().toString();String configFilePath = rootPath + "\\client_truststore.jks";log.info("证书文件地址:" + configFilePath);File configFile = new File(configFilePath);if (!configFile.exists()) {try {//获取类路径下的指定文件流  (项目目录下的:  /resource/client_truststore.jks)InputStream in = this.getClass().getClassLoader().getResourceAsStream("client_truststore.jks");FileUtils.copyInputStreamToFile(Objects.requireNonNull(in, "client_truststore.jks文件找不到"), configFile);} catch (IOException e) {throw new IllegalArgumentException("client_truststore.jks文件找不到->" + e.getMessage());}}props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, configFilePath);props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "Abc123");//注意passwod结尾的分号一定不要漏props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username='YouName' password='YouPass';");return new DefaultKafkaProducerFactory<String, String>(props);}}

这了之前在Spring中配置的属性,我们都编码在代码里面了,特别注意的是对于JKS文件的目录指定,因为SpringBoot程序最后都是打成Jar包的,因此需要上述方式来指定JKS文件位置。

然后我们编码消息监听者的配置类,因为我们已经指定了监听程序手动启动,因此配置类中需要编码手工开启某监听的启动。

package com.example.springboot.config;import com.example.springboot.listener.KafkaMessListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.system.ApplicationHome;
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.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonDeserializer;import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;/*** Kafka消费端配置*/
@Configuration
@EnableKafka
@Slf4j
public class KafkaConsumerConfig {@Beanpublic ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {return args -> {MessageListenerContainer kafkaMessListener = registry.getListenerContainer("KafkaMessListener");kafkaMessListener.start();};}/*** 配置监听,将消费工厂信息配置进去*/@Beanpublic ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<Integer, String>();factory.setConsumerFactory(consumerFactory());return factory;}/*** 消费 消费工厂*/@Beanpublic ConsumerFactory<Integer, String> consumerFactory() {return new DefaultKafkaConsumerFactory<Integer, String>(consumerConfigs());}/*** 消费配置*/public Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9002");props.put(ConsumerConfig.GROUP_ID_CONFIG, "YouGroup_ID");props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");// earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费// latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据// none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer");props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 信任所有包可以序列化props.put("security.protocol", "SASL_SSL");props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");//打包成jar后,无法找到文件的解决方法(复制jar包里的文件,到文件目录里)ApplicationHome applicationHome = new ApplicationHome(KafkaConsumerConfig.class);//项目打包成jar包所在的根路径String rootPath = applicationHome.getSource().getParentFile().toString();String configFilePath = rootPath + "\\client_truststore.jks";log.info("证书文件地址:" + configFilePath);File configFile = new File(configFilePath);if (!configFile.exists()) {try {//获取类路径下的指定文件流  (项目目录下的:  /resource/client_truststore.jks)InputStream in = this.getClass().getClassLoader().getResourceAsStream("client_truststore.jks");FileUtils.copyInputStreamToFile(Objects.requireNonNull(in, "client_truststore.jks文件找不到"), configFile);} catch (IOException e) {throw new IllegalArgumentException("client_truststore.jks文件找不到->" + e.getMessage());}}props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, configFilePath);props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "Abc123");//注意passwod结尾的分号一定不要漏props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username='YouName' password='YouPass';");return props;}
}

如上可以手工启动消息开始处理消息。

这篇关于SpringBoot集成Kafka之SASL_SSL的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/906059

相关文章

Java学习手册之Filter和Listener使用方法

《Java学习手册之Filter和Listener使用方法》:本文主要介绍Java学习手册之Filter和Listener使用方法的相关资料,Filter是一种拦截器,可以在请求到达Servl... 目录一、Filter(过滤器)1. Filter 的工作原理2. Filter 的配置与使用二、Listen

Spring Boot中JSON数值溢出问题从报错到优雅解决办法

《SpringBoot中JSON数值溢出问题从报错到优雅解决办法》:本文主要介绍SpringBoot中JSON数值溢出问题从报错到优雅的解决办法,通过修改字段类型为Long、添加全局异常处理和... 目录一、问题背景:为什么我的接口突然报错了?二、为什么会发生这个错误?1. Java 数据类型的“容量”限制

Java对象转换的实现方式汇总

《Java对象转换的实现方式汇总》:本文主要介绍Java对象转换的多种实现方式,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录Java对象转换的多种实现方式1. 手动映射(Manual Mapping)2. Builder模式3. 工具类辅助映

SpringBoot请求参数接收控制指南分享

《SpringBoot请求参数接收控制指南分享》:本文主要介绍SpringBoot请求参数接收控制指南,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Spring Boot 请求参数接收控制指南1. 概述2. 有注解时参数接收方式对比3. 无注解时接收参数默认位置

SpringBoot基于配置实现短信服务策略的动态切换

《SpringBoot基于配置实现短信服务策略的动态切换》这篇文章主要为大家详细介绍了SpringBoot在接入多个短信服务商(如阿里云、腾讯云、华为云)后,如何根据配置或环境切换使用不同的服务商,需... 目录目标功能示例配置(application.yml)配置类绑定短信发送策略接口示例:阿里云 & 腾

SpringBoot项目中报错The field screenShot exceeds its maximum permitted size of 1048576 bytes.的问题及解决

《SpringBoot项目中报错ThefieldscreenShotexceedsitsmaximumpermittedsizeof1048576bytes.的问题及解决》这篇文章... 目录项目场景问题描述原因分析解决方案总结项目场景javascript提示:项目相关背景:项目场景:基于Spring

Spring Boot 整合 SSE的高级实践(Server-Sent Events)

《SpringBoot整合SSE的高级实践(Server-SentEvents)》SSE(Server-SentEvents)是一种基于HTTP协议的单向通信机制,允许服务器向浏览器持续发送实... 目录1、简述2、Spring Boot 中的SSE实现2.1 添加依赖2.2 实现后端接口2.3 配置超时时

Spring Boot读取配置文件的五种方式小结

《SpringBoot读取配置文件的五种方式小结》SpringBoot提供了灵活多样的方式来读取配置文件,这篇文章为大家介绍了5种常见的读取方式,文中的示例代码简洁易懂,大家可以根据自己的需要进... 目录1. 配置文件位置与加载顺序2. 读取配置文件的方式汇总方式一:使用 @Value 注解读取配置方式二

一文详解Java异常处理你都了解哪些知识

《一文详解Java异常处理你都了解哪些知识》:本文主要介绍Java异常处理的相关资料,包括异常的分类、捕获和处理异常的语法、常见的异常类型以及自定义异常的实现,文中通过代码介绍的非常详细,需要的朋... 目录前言一、什么是异常二、异常的分类2.1 受检异常2.2 非受检异常三、异常处理的语法3.1 try-

Java中的@SneakyThrows注解用法详解

《Java中的@SneakyThrows注解用法详解》:本文主要介绍Java中的@SneakyThrows注解用法的相关资料,Lombok的@SneakyThrows注解简化了Java方法中的异常... 目录前言一、@SneakyThrows 简介1.1 什么是 Lombok?二、@SneakyThrows