写点什么

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

  • 2024-08-07
    福建
  • 本文字数:2130 字

    阅读完需:约 7 分钟

一、说明


在现代微服务架构中,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 通讯协议指定启用 SASL

  • sasl.mechanism 指定 SASL 使用的具体身份验证机制

  • sasl.jaas.config 指定认证模块的处理类以及 用户名 和 密码

  • auto-offset-reset 指定偏移量的逻辑,earliest 代表新加入的消费者都是从头开始消费

 

四、动态管理资源


4.1. 创建 KafkaAdminClient


KafkaAdminClient 用于管理 Kafka 资源(用户、ACL、主题等)。以下是示例代码:

@Configurationpublic 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 生产者,用于发送消息:

@Servicepublic 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 注解实现消费消息方法:

@Servicepublic 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 的安全配置。


文章转载自:zlt2000

原文链接:https://www.cnblogs.com/zlt2000/p/18342834

体验地址:http://www.jnpfsoft.com/?from=infoq

用户头像

还未添加个人签名 2023-06-19 加入

还未添加个人简介

评论

发布
暂无评论
Spring Boot 基于 SCRAM 认证集成 Kafka 的详解_kafka_不在线第一只蜗牛_InfoQ写作社区