目录
- 一、说明
- 二、添加依赖
- 三、配置 Kafka
- 四、动态管理资源
- 4.1. 创建 KafkaAdminClient
- 4.2. 动态创建用户和设置权限
- 4.3. 动态创建主题
- 五、生产者和消费者配置
- 5.1. 生产者配置
- 5.2. 消费者配置
- 六、总结
一、说明
在现代微服务架构中,Kafka
作为消息中间件被广泛使用,而安全性则是其中的一个关键因素。在本篇文章中,我们将探讨如何在 Spr编程ing 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: lophpcalhost: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, InterrupjstedException { // 构造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 的安全配置。
完整的样例代码下载:
https://gitee.com/zlt2000/kafka-scram-demo
到此这篇关于Spring Boot 基于 SCRAM 认证集成 Kafka 的详解的文章就介绍到这了,更多相关Spring Boot SCRAM 认证集成 Kafka内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)!
精彩评论