SpringBoot项目连接,有Kerberos认证的Kafka

这篇具有很好参考价值的文章主要介绍了SpringBoot项目连接,有Kerberos认证的Kafka。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

在连接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文件内容如下,具体路径和域名需要换成自己的:

debug: true
 
fusioninsight:
  kafka:
    bootstrap-servers: 10.80.10.3:21007,10.80.10.181:21007,10.80.10.52:21007
    security:
      protocol: SASL_PLAINTEXT
    kerberos:
      domain:
        name: hadoop.798687_97_4a2b_9510_00359f31c5ec.com
    sasl:
      kerberos:
        service:
          name: kafka


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

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

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

springboot连接kafka keytab认证,linq,c#
pom依赖:
我用的是华为云的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>
 
    <groupId>com.example</groupId>
    <artifactId>kafka-sample-01</artifactId>
    <version>2.3.1.RELEASE</version>
    <packaging>jar</packaging>
 
    <name>kafka-sample-01</name>
    <description>Kafka Sample 1</description>
 
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
 
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>
 
    <dependencies>
 
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
 
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.4.0-hw-ei-302002</version>
        </dependency>
 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
 
 
 
        <!-- 华为 组件 kafka  start -->
<!--        <dependency>-->
<!--            <groupId>com.huawei</groupId>-->
<!--            <artifactId>kafka-clients</artifactId>-->
<!--            <version>2.4.0</version>-->
<!--            <scope>system</scope>-->
<!--            <systemPath>${project.basedir}/lib/kafka-clients-2.4.0-hw-ei-302002.jar</systemPath>-->
<!--        </dependency>-->
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
 
    <repositories>
 
        <repository>
            <id>huaweicloudsdk</id>
            <url>https://mirrors.huaweicloud.com/repository/maven/huaweicloudsdk/</url>
            <releases><enabled>true</enabled></releases>
            <snapshots><enabled>true</enabled></snapshots>
        </repository>
 
        <repository>
            <id>central</id>
            <name>Mavn Centreal</name>
            <url>https://repo1.maven.org/maven2/</url>
        </repository>
 
    </repositories>
</project>

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

package com.example;
 
import com.common.Foo1;
 
 
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.util.backoff.FixedBackOff;
 
import java.io.File;
import java.util.HashMap;
import java.util.Map;
 
/**
 * @author
 */
@SpringBootApplication
public class Application {
 
    private final Logger logger = LoggerFactory.getLogger(Application.class);
 
    @Value("${fusioninsight.kafka.bootstrap-servers}")
    public String boostrapServers;
 
    @Value("${fusioninsight.kafka.security.protocol}")
    public String securityProtocol;
 
    @Value("${fusioninsight.kafka.kerberos.domain.name}")
    public String kerberosDomainName;
 
    @Value("${fusioninsight.kafka.sasl.kerberos.service.name}")
    public String kerberosServiceName;
 
    public static void main(String[] args) {
//        String filePath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "main"
//        String filePath = "D:\\Java\\workspace\\20231123MOSPT4eB\\sample-01\\src\\main\\resources\\";
        String filePath = "/home/yxxt/";
        System.setProperty("java.security.auth.login.config", filePath + "jaas.conf");
        System.setProperty("java.security.krb5.conf", filePath + "krb5.conf");
        SpringApplication.run(Application.class, args);
    }
 
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory, KafkaTemplate<String, String> template) {
        System.out.println(boostrapServers);
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory
            = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(template),
            new FixedBackOff(0L, 2))); // dead-letter after 3 tries
        return factory;
    }
 
    @Bean
    public RecordMessageConverter converter() {
        return new StringJsonMessageConverter();
    }
 
    // 指定消费监听,该topic有消息时立刻消费
    @KafkaListener(id = "fooGroup1", topics = "topic_ypgk")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.println("监听到了消息-----");
        logger.info("Received:消息监听成功! " );
        System.out.println("监听到了-----");
        System.out.println(record);
//        if (foo.getFoo().startsWith("fail")) {
//            // 触发83行的 ErrorHandler,将异常数据写入 topic名称+.DLT的新topic中
//            throw new RuntimeException("failed");
//        }
    }
 
    // 创建topic,指定分区数、副本数
//    @Bean
//    public NewTopic topic() {
//        return new NewTopic("topic1", 1, (short) 1);
//    }
 
    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, boostrapServers);
        configs.put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, securityProtocol);
        configs.put("sasl.kerberos.service.name", kerberosServiceName);
        configs.put("kerberos.domain.name", kerberosDomainName);
        return new KafkaAdmin(configs);
    }
 
    @Bean
    public ConsumerFactory<Object, Object> consumerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put("security.protocol", securityProtocol);
        configs.put("kerberos.domain.name", kerberosDomainName);
        configs.put("bootstrap.servers", boostrapServers);
        configs.put("sasl.kerberos.service.name", kerberosServiceName);
        configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return new DefaultKafkaConsumerFactory<>(configs);
    }
 
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        Map<String, Object> configs = new HashMap<>();
        configs.put("security.protocol", securityProtocol);
        configs.put("kerberos.domain.name", kerberosDomainName);
        configs.put("bootstrap.servers", boostrapServers);
        configs.put("sasl.kerberos.service.name", kerberosServiceName);
        configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(configs);
        return new KafkaTemplate<>(producerFactory);
    }
}

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

package com.example;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
 
import com.common.Foo1;
 
/**
 * @author haosuwei
 *
 */
@RestController
public class Controller {
 
    @Autowired
    private KafkaTemplate<String, String> template;
 
    @PostMapping(path = "/send/foo/{what}")
    public void sendFoo(@PathVariable String what) {
        Foo1 foo1 = new Foo1(what);
        this.template.send("topic1", foo1.toString());
    }
 
}

运行成功,就可以监听到主题消息了文章来源地址https://www.toymoban.com/news/detail-815595.html

到了这里,关于SpringBoot项目连接,有Kerberos认证的Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • dbever连接kerberos认证的hive

    下载地址:https://web.mit.edu/kerberos/dist/index.html 安装:下一步或者自定义安装即可 安装后会自动生成配置文件:C:ProgramDataMITKerberos5krb5.ini 向管理员要krb5.conf配置文件,将krb5.conf配置文件的内容覆盖到C:ProgramDataMITKerberos5krb5.ini里 创建路径:C:temp 配置环境变量: ps:C:te

    2023年04月12日
    浏览(82)
  • jdbc通过kerberos认证连接hive

    根据实际情况添加依赖,主要看服务器hive版本 配置文件krb5.conf,认证文件krb5.keytab,一般由服务器生成后获取 放到resources目录下     指定krb5配置文件:krb5.conf,根据实际情况替换 认证文件:krb5.keytab,根据实际情况替换 认证用户:hive,根据实际情况修改 这里是通过将配置文

    2024年02月10日
    浏览(63)
  • Kafka-配置Kerberos安全认证(JDK8、JDK11)

    一、相关配置 1、JAAS 配置文件 2、keytab 文件(kafka.service.keytab) 从 Kerberos 服务器上拷贝到目标机器 或 找运维人员要一份 3、Kerberos 配置文件(krb5.conf) krb5文件参数说明:krb5.conf(5) 从 Kerberos 服务器上拷贝到目标机器 或 找运维人员要一份 Tip: JDK11版本 sun.security.krb5.Config 类

    2024年02月15日
    浏览(56)
  • Java 访问连接Hive的Kerberos认证前提

    The reason why authentication needs to be added when connecting to Hive in the Java code is that Hive is a distributed data processing system that is typically deployed in a multi-user environment, where access to data is controlled by policies and permissions. Therefore, to ensure the security and integrity of the data, it is necessary to authenticate users

    2024年02月07日
    浏览(47)
  • 【dbeaver】win环境的kerberos认证和Clouders/cdh集群中Kerberos认证使用Dbeaver连接Hive、Impala和Phoenix

    1.1 下载安装MIT KERBEROS客户端 MIT KERBEROS 下载较新的版本即可。 下载之后一路默认安装即可。 注意:不要修改软件安装位置。 修改系统环境变量中的Path。将刚刚的安装路径置顶。(不置顶,也要比 %JAVA_HOME%bin 和 anaconda 相关的高) 使用CMD命令确认下: 1.2 修改 krb5.conf 文件并

    2024年02月02日
    浏览(54)
  • 企业级大数据安全架构(十)DBeaver连接Hive的Kerberos认证配置

    1.配置本地hosts 因为Kerberos认证过程及集群服务中,很多是以主机名的形式进行访问的,所以工作机要设置hosts. 域名映射,我们通过部署CDH的集群的每一台机器都已经配置了host(文件为/etc/hosts),工作机也需要配置window的host文件,如果提示无法修改,一般是需要管理员权限的原

    2024年02月21日
    浏览(46)
  • kafka开kerberos认证报错the client is being asked for a password

    @Kafka kerberos认证错误记录TOC kafka 开发调试,开 kerberos情况下遇到的错误。 Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner authentication information from the user Caused by: javax.security.auth.login.LoginExcepti

    2023年04月21日
    浏览(38)
  • CDH数仓项目(三) —— Kerberos安全认证和Sentry权限管理

    本文基于《CDH数仓项目(一) —— CDH安装部署搭建详细流程》和《CDH数仓项目(二) —— 用户行为数仓和业务数仓搭建》和搭建CDH数仓。本章节主要介绍基于CDH数仓的Kerberos认证和Sentry权限管理 Kerberos是一种计算机网络授权协议,用来在非安全网络中,对个人通信以安全的手段进

    2023年04月22日
    浏览(54)
  • Kafka3.4 SASL/kerberos/ACL 证以及 SSL 加密连接

    前面我们使用 kafka3.3.1 on zookeeper 的模式进行多网段监听的 kafka 集群,顺便搭建起 kafkaui 后发现一些问题,我们 kafka 集群没有连接认证,万一谁知道了我们的 kafka 连接地址,岂不是随随便便就能消费数据、清空数据、胡乱修改数据了吗? 所以本章节进行认证连接的搭建,参

    2024年02月14日
    浏览(40)
  • JDK8 和 JDK17 下基于JDBC连接Kerberos认证的Hive(代码已测试通过)

    之前自研平台是基于jdk8开发的,连接带Kerberos的hive也是jdk8,现在想升级jdk到17,发现过Kerberos的hive有点不一样,特地记录 连接Kerberos, krb5.conf 和对应服务的 keytab 文件以及对应的 principal 肯定是需要提前准备的, 一般从服务器或者运维那里获取 krb5.conf 与 Hive对应的 keytab 文件。 注意

    2024年02月20日
    浏览(49)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包