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
通讯协议指定启用SASLsasl.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 的安全配置。