devbox
Li_阴宅
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot项目连接,有Kerberos认证的Kafka_springboot kafka kerberos

作者：Li_阴宅 | 2024-08-09 02:26:50

赞

踩

springboot kafka kerberos

在连接Kerberos认证kafka之前，需要了解Kerberos协议

二、什么是Kerberos协议

Kerberos是一种计算机网络认证协议 ，其设计目标是通过密钥系统为网络中通信的客户机(Client)/服务器(Server)应用程序提供严格的身份验证服务，确保通信双方身份的真实性和安全性。不同于其他网络服务，Kerberos协议中不是所有的客户端向想要访问的网络服务发起请求，他就能建立连接然后进行加密通信，而是在发起服务请求后必须先进行一系列的身份认证，包括客户端和服务端两方的双向认证，只有当通信双方都认证通过对方身份之后，才可以互相建立起连接，进行网络通信。即Kerberos协议的侧重在于认证通信双方的身份，客户端需要确认即将访问的网络服务就是自己所想要访问的服务而不是一个伪造的服务器，而服务端需要确认这个客户端是一个身份真实，安全可靠的客户端，而不是一个想要进行恶意网络攻击的用户。

三、Kerberos协议角色组成
Kerberos协议中存在三个角色，分别是：

客户端(Client)：发送请求的一方
服务端(Server)：接收请求的一方
密钥分发中心(Key distribution KDC)

一，首先需要准备三个文件

（user.keytab，krb5.conf，jass.conf）

其中user.keytab和krb5.conf是两个认证文件，需要厂商提供，就是你连接谁的kafka，让谁提供

jass.conf文件需要自己在本地创建

jass.conf文件内容如下，具体路径和域名需要换成自己的：

  1. debug: true
  2.  
  3. fusioninsight:
  4.   kafka:
  5.     bootstrap-servers: 10.80.10.3:21007,10.80.10.181:21007,10.80.10.52:21007
  6.     security:
  7.       protocol: SASL_PLAINTEXT
  8.     kerberos:
  9.       domain:
  10.         name: hadoop.798687_97_4a2b_9510_00359f31c5ec.com
  11.     sasl:
  12.       kerberos:
  13.         service:
  14.           name: kafka

其中kerberos.domain.name:hadoop.798687_97_4a2b_9510_00359f31c5ec.com

hadoop.798687_97_4a2b_9510_00359f31c5ec.com需要根据现场提供给你的域名

二、文件准备好后可以将三个配置文件，放在自己项目中，也可以放在服务器的某个目录下，只要确保项目启动后能读取到即可

我的目录结构如下：

pom依赖：

我用的是华为云的Kafka依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3. 	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4. 	<modelVersion>4.0.0</modelVersion>
  5.  
  6. 	<groupId>com.example</groupId>
  7. 	<artifactId>kafka-sample-01</artifactId>
  8. 	<version>2.3.1.RELEASE</version>
  9. 	<packaging>jar</packaging>
  10.  
  11. 	<name>kafka-sample-01</name>
  12. 	<description>Kafka Sample 1</description>
  13.  
  14. 	<parent>
  15. 		<groupId>org.springframework.boot</groupId>
  16. 		<artifactId>spring-boot-starter-parent</artifactId>
  17. 		<version>2.2.0.RELEASE</version>
  18. 		<relativePath/> <!-- lookup parent from repository -->
  19. 	</parent>
  20.  
  21. 	<properties>
  22. 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  23. 		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  24. 		<java.version>1.8</java.version>
  25. 	</properties>
  26.  
  27. 	<dependencies>
  28.  
  29. 		<dependency>
  30. 			<groupId>org.springframework.kafka</groupId>
  31. 			<artifactId>spring-kafka</artifactId>
  32. 			<exclusions>
  33. 				<exclusion>
  34. 					<groupId>org.apache.kafka</groupId>
  35. 					<artifactId>kafka-clients</artifactId>
  36. 				</exclusion>
  37. 			</exclusions>
  38. 		</dependency>
  39.  
  40. 		<dependency>
  41. 			<groupId>org.apache.kafka</groupId>
  42. 			<artifactId>kafka-clients</artifactId>
  43. 			<version>2.4.0-hw-ei-302002</version>
  44. 		</dependency>
  45.  
  46. 		<dependency>
  47. 			<groupId>org.springframework.boot</groupId>
  48. 			<artifactId>spring-boot-starter-test</artifactId>
  49. 			<scope>test</scope>
  50. 		</dependency>
  51. 		<dependency>
  52. 			<groupId>org.springframework.boot</groupId>
  53. 			<artifactId>spring-boot-starter-web</artifactId>
  54. 		</dependency>
  55.  
  56.  
  57.  
  58. 		<!-- 华为 组件 kafka  start -->
  59. <!--		<dependency>-->
  60. <!--			<groupId>com.huawei</groupId>-->
  61. <!--			<artifactId>kafka-clients</artifactId>-->
  62. <!--			<version>2.4.0</version>-->
  63. <!--			<scope>system</scope>-->
  64. <!--			<systemPath>${project.basedir}/lib/kafka-clients-2.4.0-hw-ei-302002.jar</systemPath>-->
  65. <!--		</dependency>-->
  66. 	</dependencies>
  67.  
  68. 	<build>
  69. 		<plugins>
  70. 			<plugin>
  71. 				<groupId>org.springframework.boot</groupId>
  72. 				<artifactId>spring-boot-maven-plugin</artifactId>
  73. 			</plugin>
  74. 		</plugins>
  75. 	</build>
  76.  
  77. 	<repositories>
  78.  
  79. 		<repository>
  80. 			<id>huaweicloudsdk</id>
  81. 			<url>https://mirrors.huaweicloud.com/repository/maven/huaweicloudsdk/</url>
  82. 			<releases><enabled>true</enabled></releases>
  83. 			<snapshots><enabled>true</enabled></snapshots>
  84. 		</repository>
  85.  
  86. 		<repository>
  87. 			<id>central</id>
  88. 			<name>Mavn Centreal</name>
  89. 			<url>https://repo1.maven.org/maven2/</url>
  90. 		</repository>
  91.  
  92. 	</repositories>
  93. </project>

然后再SpringBoot项目启动类如下：

  1. package com.example;
  2.  
  3. import com.common.Foo1;
  4.  
  5.  
  6. import org.apache.kafka.clients.admin.AdminClientConfig;
  7. import org.apache.kafka.clients.admin.NewTopic;
  8. import org.apache.kafka.clients.consumer.ConsumerRecord;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.factory.annotation.Value;
  12. import org.springframework.boot.SpringApplication;
  13. import org.springframework.boot.autoconfigure.SpringBootApplication;
  14. import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
  15. import org.springframework.context.annotation.Bean;
  16. import org.springframework.kafka.annotation.KafkaListener;
  17. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
  18. import org.springframework.kafka.core.ConsumerFactory;
  19. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
  20. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  21. import org.springframework.kafka.core.KafkaAdmin;
  22. import org.springframework.kafka.core.KafkaTemplate;
  23. import org.springframework.kafka.core.ProducerFactory;
  24. import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
  25. import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
  26. import org.springframework.kafka.support.converter.RecordMessageConverter;
  27. import org.springframework.kafka.support.converter.StringJsonMessageConverter;
  28. import org.springframework.util.backoff.FixedBackOff;
  29.  
  30. import java.io.File;
  31. import java.util.HashMap;
  32. import java.util.Map;
  33.  
  34. /**
  35.  * @author
  36.  */
  37. @SpringBootApplication
  38. public class Application {
  39.  
  40.     private final Logger logger = LoggerFactory.getLogger(Application.class);
  41.  
  42.     @Value("${fusioninsight.kafka.bootstrap-servers}")
  43.     public String boostrapServers;
  44.  
  45. 	@Value("${fusioninsight.kafka.security.protocol}")
  46. 	public String securityProtocol;
  47.  
  48. 	@Value("${fusioninsight.kafka.kerberos.domain.name}")
  49. 	public String kerberosDomainName;
  50.  
  51. 	@Value("${fusioninsight.kafka.sasl.kerberos.service.name}")
  52. 	public String kerberosServiceName;
  53.  
  54.     public static void main(String[] args) {
  55. //        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main"
  56. //        String filePath = "D:\\Java\\workspace\\20231123MOSPT4eB\\sample-01\\src\\main\\resources\\";
  57.         String filePath = "/home/yxxt/";
  58.         System.setProperty("java.security.auth.login.config", filePath + "jaas.conf");
  59.         System.setProperty("java.security.krb5.conf", filePath + "krb5.conf");
  60.         SpringApplication.run(Application.class, args);
  61.     }
  62.  
  63.     @Bean
  64.     public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
  65.         ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
  66.         ConsumerFactory<Object, Object> kafkaConsumerFactory, KafkaTemplate<String, String> template) {
  67.         System.out.println(boostrapServers);
  68.         ConcurrentKafkaListenerContainerFactory<Object, Object> factory
  69.             = new ConcurrentKafkaListenerContainerFactory<>();
  70.         configurer.configure(factory, kafkaConsumerFactory);
  71.         factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(template),
  72.             new FixedBackOff(0L, 2))); // dead-letter after 3 tries
  73.         return factory;
  74.     }
  75.  
  76.     @Bean
  77.     public RecordMessageConverter converter() {
  78.         return new StringJsonMessageConverter();
  79.     }
  80.  
  81.     // 指定消费监听，该topic有消息时立刻消费
  82.     @KafkaListener(id = "fooGroup1", topics = "topic_ypgk")
  83.     public void listen(ConsumerRecord<String, String> record) {
  84.         System.out.println("监听到了消息-----");
  85.         logger.info("Received:消息监听成功！ " );
  86.         System.out.println("监听到了-----");
  87.         System.out.println(record);
  88. //        if (foo.getFoo().startsWith("fail")) {
  89. //            // 触发83行的 ErrorHandler，将异常数据写入 topic名称+.DLT的新topic中
  90. //            throw new RuntimeException("failed");
  91. //        }
  92.     }
  93.  
  94.     // 创建topic，指定分区数、副本数
  95. //    @Bean
  96. //    public NewTopic topic() {
  97. //        return new NewTopic("topic1", 1, (short) 1);
  98. //    }
  99.  
  100.     @Bean
  101.     public KafkaAdmin kafkaAdmin() {
  102.         Map<String, Object> configs = new HashMap<>();
  103.         configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, boostrapServers);
  104.         configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, securityProtocol);
  105.         configs.put("sasl.kerberos.service.name", kerberosServiceName);
  106.         configs.put("kerberos.domain.name", kerberosDomainName);
  107.         return new KafkaAdmin(configs);
  108.     }
  109.  
  110.     @Bean
  111.     public ConsumerFactory<Object, Object> consumerFactory() {
  112.         Map<String, Object> configs = new HashMap<>();
  113.         configs.put("security.protocol", securityProtocol);
  114.         configs.put("kerberos.domain.name", kerberosDomainName);
  115.         configs.put("bootstrap.servers", boostrapServers);
  116.         configs.put("sasl.kerberos.service.name", kerberosServiceName);
  117.         configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  118.         configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  119.         return new DefaultKafkaConsumerFactory<>(configs);
  120.     }
  121.  
  122.     @Bean
  123.     public KafkaTemplate<String, String> kafkaTemplate() {
  124.         Map<String, Object> configs = new HashMap<>();
  125.         configs.put("security.protocol", securityProtocol);
  126.         configs.put("kerberos.domain.name", kerberosDomainName);
  127.         configs.put("bootstrap.servers", boostrapServers);
  128.         configs.put("sasl.kerberos.service.name", kerberosServiceName);
  129.         configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  130.         configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  131.         ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(configs);
  132.         return new KafkaTemplate<>(producerFactory);
  133.     }
  134. }
  135.  
  136.

生产者：通过发送请求进行向主题里发送消息

  1. package com.example;
  2.  
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.kafka.core.KafkaTemplate;
  5. import org.springframework.web.bind.annotation.PathVariable;
  6. import org.springframework.web.bind.annotation.PostMapping;
  7. import org.springframework.web.bind.annotation.RestController;
  8.  
  9. import com.common.Foo1;
  10.  
  11. /**
  12.  * @author haosuwei
  13.  *
  14.  */
  15. @RestController
  16. public class Controller {
  17.  
  18. 	@Autowired
  19. 	private KafkaTemplate<String, String> template;
  20.  
  21. 	@PostMapping(path = "/send/foo/{what}")
  22. 	public void sendFoo(@PathVariable String what) {
  23. 		Foo1 foo1 = new Foo1(what);
  24. 		this.template.send("topic1", foo1.toString());
  25. 	}
  26.  
  27. }

运行成功，就可以监听到主题消息了

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Li_阴宅/article/detail/950994
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号