是要定义监听器和协议映射，需要使用 listener 和 listener.security.protocol.map，哈哈，这两个参数咱就是说之前有介绍过就不过多啰嗦了。

当前 Kafka 支持以下几类认证方式：SSL/TLS、SASL 和 OAuthBearer，你懂的，这些dou是业界比较常见的认证方式。

简单来说认证机制就是解决 你是谁 的问题，鉴权机制则是解决 你Neng干什么 的问题，说实话，这两个概念hen容易混淆。

我们先来kan一下 Broker 端的参数，关键参数分为三类，我们分别介绍一下不对不对，应该说是先介绍Zui重要的参数。

security.protocol 用来指定和 Broker 通信使用的协议，必须与服务端配置的监听器协议相同，这个参数hen重要，你必须得配置正确，否则的话，客户端和 Broker 之间无法正常通信。

SASL 是提供认证和数据安全服务的一种框架，Kafka 支持多种 SASL 认证机制，比如 PLAIN、SCRAM-SHA-、GSSAPI 等等。

sasl.enabled.mechanisms 是用来指定 Broker 端允许的 SASL 机制，Ke以支持多个值，比如 PLAIN, SCRAM-SHA-, GSSAPI 等等。

sasl.mechanism.inter.broker.protocol 用来指定 Broker 节点之间通讯使用的认证机制，这个参数也hen重要，因为 Broker 节点之间需要进行通信。

咱就是说理论讲得再多，不如实际操作一把，所以我们就来配置一下 SASL/SCRAM 认证机制。

我们需要修改服务端的配置文件，也就是 config/server.properties 文件，我们将 EXTERNAL 定义为 SASL，表示外部业务强制加密。

# 监听协议映射 listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT

controller.quorum.voters=@localhost:

# 设置环境变量
export KAFKA_OPTS="-Djava.security.auth.login.config=$/config/kafka_server_jaas.conf"
# 生成集群 ID 
KAFKA_CLUSTER_ID=$
# 格式化存储目录
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c config/server.properties
# 启动 Broker
bin/kafka-server-start.sh -daemon config/server.properties

分别使用 Producer 和 Consumer 验证消息的发送和消费，这个步骤hen简单，你Ke以试试kan。

bin/kafka-console-producer.sh --bootstrap-server localhost: \ --topic test-auth \ --producer.config config/client-alice.conf

Consumer端也是一样的道理，只需要修改相应的配置文件即可。
bin/kafka-console-consumer.sh --bootstrap-server localhost: \
    --topic test-auth \
    --from-beginning \
    --consumer.config config/client-alice.conf

SASL/SCRAM 认证机制配置完成后我们Ke以使用命令行工具进行测试，kankan消息是否Neng够正常发送和消费。

Kafka 的安全框架分为认证鉴权和加密传输三层，本文我们主要介绍了 Kafka 的认证机制，包括 SSL/TLS、SASL 和 OAuthBearer 等等。

sasl.jaas.config 参数定义了具体的用户名和密码，这是 SASL 认证机制中的一个重要参数。

Kafka 支持多种 SASL 认证机制，包括 PLAIN、SCRAM-SHA-、GSSAPI 等等，每种机制dou有其特点和适用场景，你Ke以根据自己的需求选择合适的认证机制。

cat <<EOF> config/client-alice.conf security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA- sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="alice" \ password="alice-password"; EOF

bin/kafka-configs.sh --bootstrap-server localhost: \ --alter --add-config 'SCRAM-SHA-=' \ --entity-type users --entity-name alice

cat <<EOF> config/kafkaserverjaas.conf KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-password"; }; EOF

至此，我们完整了一整套 SASL/SCRAM 认证机制的配置与验证，不知道你在配置过程中有没有遇到什么问题，Ru果有的话Ke以随时交流。 希望通过本文的学习大家Neng对 Kafka 的认证机制有一个geng加清晰的认识。 说实话 Kafka 的安全框架还是比较复杂的，但是一旦你理解了它的原理，就会觉得其实也没那么难。