Spring Boot 基于 SCRAM 認(rèn)證集成 Kafka 的詳解
一、說(shuō)明
在現(xiàn)代微服務(wù)架構(gòu)中，Kafka 作為消息中間件被廣泛使用，而安全性則是其中的一個(gè)關(guān)鍵因素。在本篇文章中，我們將探討如何在 Spring Boot 應(yīng)用中集成 Kafka 并使用 SCRAM 認(rèn)證機(jī)制進(jìn)行安全連接；并實(shí)現(xiàn)動(dòng)態(tài)創(chuàng)建賬號(hào)、ACL 權(quán)限、Topic，以及生產(chǎn)者和消費(fèi)者等操作。
需要準(zhǔn)備一個(gè)配置了 SCRAM 認(rèn)證的 Kafka 環(huán)境，可參考《基于 SASL/SCRAM 讓 Kafka 實(shí)現(xiàn)動(dòng)態(tài)授權(quán)認(rèn)證》 進(jìn)行部署。
二、添加依賴
在 Spring Boot 項(xiàng)目的 pom.xml 中添加 spring-kafka 依賴
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>三、配置 Kafka
在 application.yml 中配置 Kafka 的相關(guān)屬性，包括服務(wù)器地址、認(rèn)證信息等。
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 通訊協(xié)議指定啟用SASL
- sasl.mechanism 指定 SASL 使用的具體身份驗(yàn)證機(jī)制
- sasl.jaas.config 指定認(rèn)證模塊的處理類以及 「用戶名」 和 「密碼」
- auto-offset-reset 指定偏移量的邏輯，「earliest」 代表新加入的消費(fèi)者都是從頭開(kāi)始消費(fèi)
四、動(dòng)態(tài)管理資源
4.1. 創(chuàng)建 KafkaAdminClient
KafkaAdminClient 用于管理 Kafka 資源（用戶、ACL、主題等）。以下是示例代碼：
@Configuration
public class KafkaConfig {
@Bean
public KafkaAdminClient kafkaAdminClient(KafkaAdmin kafkaAdmin) {
return (KafkaAdminClient) KafkaAdminClient.create(kafkaAdmin.getConfigurationProperties());
}
}4.2. 動(dòng)態(tài)創(chuàng)建用戶和設(shè)置權(quán)限
使用 Kafka AdminClient API 實(shí)現(xiàn)動(dòng)態(tài)創(chuàng)建用戶和設(shè)置 ACL 權(quán)限：
/**
* 創(chuàng)建用戶
*/
public void createUser(String userName, String password) throws ExecutionException, InterruptedException {
// 構(gòu)造Scram認(rèn)證機(jī)制信息
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();
}
/**
* 配置用戶只讀權(quán)限
*/
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. 動(dòng)態(tài)創(chuàng)建主題
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();
}五、生產(chǎn)者和消費(fèi)者配置
5.1. 生產(chǎn)者配置
配置 Kafka 生產(chǎn)者，用于發(fā)送消息：
@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. 消費(fèi)者配置
使用 @KafkaListener 注解實(shí)現(xiàn)消費(fèi)消息方法：
@Service
public class KafkaConsumer {
@KafkaListener(topics = "test", groupId = "test-consumer-group")
public void consume(String message) {
System.out.println("Received message: " + message);
}
}六、總結(jié)
通過(guò)以上步驟，我們成功地在 Spring Boot 應(yīng)用中集成了 Kafka，并使用 SCRAM 認(rèn)證機(jī)制進(jìn)行安全連接；確保在生產(chǎn)環(huán)境中妥善管理用戶憑證，并根據(jù)需要調(diào)整 Kafka 的安全配置。
完整的樣例代碼下載：
