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

一、阐明

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

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

 
二、添加依赖

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

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4. </dependency>

复制代码

 
三、配置 Kafka

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

1. spring:  
2.   kafka:  
3.     bootstrap-servers: localhost:9092  
4.     properties:  
5.       security.protocol: SASL_PLAINTEXT  
6.       sasl.mechanism: SCRAM-SHA-256
7.       sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="your_username" password="your_password";
8.     consumer:
9.       group-id: test-consumer-group
10.       auto-offset-reset: earliest
11.       properties:
12.         sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="test" password="test";
13.     producer:
14.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
15.       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、主题等）。以下是示例代码：

1. @Configuration
2. public class KafkaConfig {
3.     @Bean
4.     public KafkaAdminClient kafkaAdminClient(KafkaAdmin kafkaAdmin) {
5.         return (KafkaAdminClient) KafkaAdminClient.create(kafkaAdmin.getConfigurationProperties());
6.     }
7. }

复制代码

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

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

1. /**
2. * 创建用户
3. */
4. public void createUser(String userName, String password) throws ExecutionException, InterruptedException {
5.     // 构造Scram认证机制信息
6.     ScramCredentialInfo info = new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_256, 8192);
7.     //用户信息
8.     UserScramCredentialAlteration userScramCredentialAdd = new UserScramCredentialUpsertion(userName, info, password);
9.     AlterUserScramCredentialsResult result = kafkaAdminClient.alterUserScramCredentials(List.of(userScramCredentialAdd));
10.     result.all().get();
11. }
13. /**
14. * 配置用户只读权限
15. */
16. public void createAcl(String account, String topicName, String consumerGroup) {
17.     AclBinding aclBindingTopic = genAclBinding(account, ResourceType.TOPIC, topicName, AclOperation.READ);
18.     AclBinding aclBindingGroup = genAclBinding(account, ResourceType.GROUP, consumerGroup, AclOperation.READ);
19.     kafkaAdminClient.createAcls(List.of(aclBindingTopic, aclBindingGroup));
20. }

复制代码

 
4.3. 动态创建主题

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

复制代码

 
五、生产者和消耗者配置

5.1. 生产者配置

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

1. @Service
2. public class KafkaProducer {
3.     private final KafkaTemplate<String, String> kafkaTemplate;
5.     public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
6.         this.kafkaTemplate = kafkaTemplate;
7.     }
9.     public void sendMessage(String message) {
10.         kafkaTemplate.send("test", message);
11.     }
12. }

复制代码

 
5.2. 消耗者配置

利用 @KafkaListener 注解实现消耗消息方法：

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

复制代码

 
六、总结

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

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

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。