Spring Boot 基于 SCRAM 认证集成 Kafka 的详解

这篇文章介绍了Spring Boot 基于 SCRAM 认证集成 Kafka 的详解，分享给大家做个参考，收藏极客之家收获更多编程知识

一、说明

在现代微服务架构中，Kafka 作为消息中间件被广泛使用，而安全性则是其中的一个关键因素。在本篇文章中，我们将探讨如何在 Spring Boot 应用中集成 Kafka 并使用 SCRAM 认证机制进行安全连接；并实现动态创建账号、ACL 权限、Topic，以及生产者和消费者等操作。

需要准备一个配置了 SCRAM 认证的 Kafka 环境，可参考《基于 SASL/SCRAM 让 Kafka 实现动态授权认证》 进行部署。

 

二、添加依赖

在 Spring Boot 项目的 pom.xml 中添加 spring-kafka 依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

 

三、配置 Kafka

在 application.yml 中配置 Kafka 的相关属性，包括服务器地址、认证信息等。

spring:  
  kafka:  
    bootstrap-servers: localhost:9092  
    properties:  
      security.protocol: SASL_PLAINTEXT  
      sasl.mechanism: SCRAM-SHA-256 
      sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="your_username" password="your_password";
    consumer:
      group-id: test-consumer-group
      auto-offset-reset: earliest
      properties:
        sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="test" password="test";
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

• bootstrap-servers Kafka 的集群地址
• security.protocol 通讯协议指定启用SASL
• sasl.mechanism 指定 SASL 使用的具体身份验证机制
• sasl.jaas.config 指定认证模块的处理类以及 用户名 和 密码
• auto-offset-reset 指定偏移量的逻辑，earliest 代表新加入的消费者都是从头开始消费

 

四、动态管理资源

4.1. 创建 KafkaAdminClient

KafkaAdminClient 用于管理 Kafka 资源（用户、ACL、主题等）。以下是示例代码：

@Configuration
public class KafkaConfig {
    @Bean
    public KafkaAdminClient kafkaAdminClient(KafkaAdmin kafkaAdmin) {
        return (KafkaAdminClient) KafkaAdminClient.create(kafkaAdmin.getConfigurationProperties());
    }
}

 

4.2. 动态创建用户和设置权限

使用 Kafka AdminClient API 实现动态创建用户和设置 ACL 权限：

/**
 * 创建用户
 */
public void createUser(String userName, String password) throws ExecutionException, InterruptedException {
    // 构造Scram认证机制信息
    ScramCredentialInfo info = new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_256, 8192);
    //用户信息
    UserScramCredentialAlteration userScramCredentialAdd = new UserScramCredentialUpsertion(userName, info, password);
    AlterUserScramCredentialsResult result = kafkaAdminClient.alterUserScramCredentials(List.of(userScramCredentialAdd));
    result.all().get();
}

/**
 * 配置用户只读权限
 */
public void createAcl(String account, String topicName, String consumerGroup) {
    AclBinding aclBindingTopic = genAclBinding(account, ResourceType.TOPIC, topicName, AclOperation.READ);
    AclBinding aclBindingGroup = genAclBinding(account, ResourceType.GROUP, consumerGroup, AclOperation.READ);
    kafkaAdminClient.createAcls(List.of(aclBindingTopic, aclBindingGroup));
}

 

4.3. 动态创建主题

public void createTopic(String topicName, int partitions, short replicationFactor) throws ExecutionException, InterruptedException {
    NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor);
    CreateTopicsResult result = kafkaAdminClient.createTopics(List.of(newTopic));
    result.all().get();
}

 

五、生产者和消费者配置

5.1. 生产者配置

配置 Kafka 生产者，用于发送消息：

@Service
public class KafkaProducer {
    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(String message) {
        kafkaTemplate.send("test", message);
    }
}

 

5.2. 消费者配置

使用 @KafkaListener 注解实现消费消息方法：

@Service
public class KafkaConsumer {
    @KafkaListener(topics = "test", groupId = "test-consumer-group")
    public void consume(String message) {
        System.out.println("Received message: " + message);
    }
}

 

六、总结

通过以上步骤，我们成功地在 Spring Boot 应用中集成了 Kafka，并使用 SCRAM 认证机制进行安全连接；确保在生产环境中妥善管理用户凭证，并根据需要调整 Kafka 的安全配置。

完整的样例代码下载：
https://gitee.com/zlt2000/kafka-scram-demo

版权声明：本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！