kafka账号密码配置

### 如何在Kafka中设置和配置账号及密码认证 为了实现基于账号和密码的安全认证，在Kafka集群上启用SASL/SCRAM机制是一个常见的做法。这涉及到服务器端以及客户端的相应配置。 #### 服务器端配置 对于CentOS 7环境下的Kafka安装，要开启SASL/SCRAM身份验证，则需编辑`$KAFKA_HOME/config/server.properties`文件并加入如下参数： ```properties listeners=SASL_PLAINTEXT://:9092 listener.name.sasl_plaintext.scram-sha-512.sasl.enabled.mechanisms=SCRAM-SHA-512 sasl.enabled.mechanisms=SCRAM-SHA-512 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer allow.everyone.if.no.acl.found=true ``` 上述配置指定了监听器采用SASL_PLAINTEXT协议，并启用了SCRAM-SHA-512作为加密算法[^1]。 接着还需要创建JAAS（Java Authentication and Authorization Service）配置文件用于定义具体的用户名及其对应的哈希后的密钥。该文件通常位于`$KAFKA_HOME/config/kafka_server_jaas.conf`: ```plaintext KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required; }; ``` 之后重启Kafka服务使更改生效。 #### 客户端配置 当使用Spring MVC或者Spring Boot应用程序来连接已设置了账户名与密码保护的Kafka实例时，可以通过XML或Java代码形式指定必要的属性以便完成身份验证过程。 如果选择通过XML方式进行配置的话，可以在applicationContext.xml里添加类似下面的内容： ```xml <bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"> <constructor-arg> <map> <!-- other properties --> <entry key="security.protocol" value="SASL_PLAINTEXT"/> <entry key="sasl.mechanism" value="SCRAM-SHA-512"/> <entry key="sasl.jaas.config" value="org.apache.kafka.common.security.scram.ScramLoginModule required username=&quot;your_username&quot; password=&quot;your_password&quot;;"/> </map> </constructor-arg> </bean> <bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory"> <constructor-arg> <map> <!-- other properties --> <entry key="security.protocol" value="SASL_PLAINTEXT"/> <entry key="sasl.mechanism" value="SCRAM-SHA-512"/> <entry key="sasl.jaas.config" value="org.apache.kafka.common.security.scram.ScramLoginModule required username=&quot;your_username&quot; password=&quot;your_password&quot;;"/> </map> </constructor-arg> </bean> ``` 而如果是利用纯Java代码来进行相同的操作，则可以参照以下例子： ```java Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("key.deserializer", StringDeserializer.class.getName()); props.put("value.deserializer", StringDeserializer.class.getName()); // SASL/PLAIN configuration props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"your_username\" password=\"your_password\";"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); ``` 以上就是关于如何在Kafka中实施账号密码认证的相关说明。