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

技术分享 1个月前 (08-06) 0 999+

一、说明

在现代微服务架构中，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);     } }