Spring Boot 基于 SCRAM 認證集成 Kafka 的詳解
作者：zlt2000
我們成功地在 Spring Boot 應用中集成了 Kafka，并使用 SCRAM 認證機制進行安全連接；確保在生產環(huán)境中妥善管理用戶憑證，并根據需要調整 Kafka 的安全配置。
一、說明
在現(xiàn)代微服務架構中，Kafka 作為消息中間件被廣泛使用，而安全性則是其中的一個關鍵因素。在本篇文章中，我們將探討如何在 Spring Boot 應用中集成 Kafka 并使用 SCRAM 認證機制進行安全連接；并實現(xiàn)動態(tài)創(chuàng)建賬號、ACL 權限、Topic，以及生產者和消費者等操作。
需要準備一個配置了 SCRAM 認證的 Kafka 環(huán)境，可參考《基于 SASL/SCRAM 讓 Kafka 實現(xiàn)動態(tài)授權認證》 進行部署。
二、添加依賴
在 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 通訊協(xié)議指定啟用SASL
- sasl.mechanism 指定 SASL 使用的具體身份驗證機制
- sasl.jaas.config 指定認證模塊的處理類以及 「用戶名」 和 「密碼」
- auto-offset-reset 指定偏移量的邏輯，「earliest」 代表新加入的消費者都是從頭開始消費
四、動態(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. 動態(tài)創(chuàng)建用戶和設置權限
使用 Kafka AdminClient API 實現(xiàn)動態(tài)創(chuàng)建用戶和設置 ACL 權限：
/**
* 創(chuàng)建用戶
*/
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. 動態(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();
}五、生產者和消費者配置
5.1. 生產者配置
配置 Kafka 生產者，用于發(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. 消費者配置
使用 @KafkaListener 注解實現(xiàn)消費消息方法：
@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 認證機制進行安全連接；確保在生產環(huán)境中妥善管理用戶憑證，并根據需要調整 Kafka 的安全配置。
完整的樣例代碼下載：
責任編輯：武曉燕
來源：
陶陶技術筆記
