使用SASL_SSL协议访问集群
更新时间：2025-08-06
本文介绍创建完集群后，如何使用Kafka脚本通过SASL_SSL协议访问Kafka集群，SASL_SSL协议需要在公网网络下进行使用。如果想使用Java、Go、PHP等语言访问Kakfa集群，请参考开发指南。
前提条件
- 步骤一：创建Kafka集群
- 步骤二：创建主题
- 步骤三：配置权限认证
- 安装JDK1.8及以上版本：JDK下载地址。
具体操作步骤
- 下载Kafka 2.7.2安装包，并进行解压：kafka_2.13-2.7.2.tgz，其中解压后的bin目录为Kafka提供的可执行脚本。
- 在集群详情中，点击接入点查看SASL_SSL协议所使用的访问地址和端口。
- 创建JAAS配置文件：kafka_client_jaas.conf，认证机制支持PLAIN、SCRAM-SHA-512两种机制，根据集群所使用的认证方式进行选择内容格式如下：
SCRAM-SHA-512：
Java
1KafkaClient {
2 org.apache.kafka.common.security.scram.ScramLoginModule required
3 username="{username}"
4 password="{password}";
5};PLAIN：
Java
1KafkaClient {
2 org.apache.kafka.common.security.plain.PlainLoginModule required
3 username="{username}"
4 password="{password}";
5};其中，{username}和{password}为步骤三：配置权限认证时创建的用户名和密码。
- 下载证书文件：如何下载证书？。
- 创建kafka.properties配置文件，认证机制支持PLAIN、SCRAM-SHA-512两种机制，根据集群所使用的认证方式选择。内容格式如下：
Java
1security.protocol=SASL_SSL
2sasl.mechanism=SCRAM-SHA-512
3#sasl.mechanism=PLAIN
4#第4步下载的SSL证书文件路径
5ssl.truststore.location=/***/client.truststore.jks
6#SSL证书的密码，固定为bms@kafka
7ssl.truststore.password=bms@kafka
8#固定为空
9ssl.endpoint.identification.algorithm=- 指定第3步中所配置的JAAS文件，需要设置完整的文件路径：
Java
1export KAFKA_OPTS='-Djava.security.auth.login.config=/***/kafka_client_jaas.conf'- 进入Kafka安装包解压后的bin目录下，使用Kafka脚本访问集群。
- 查看Topic列表，需要使用--command-config配置项来指定第5步中创建的kafka.properties文件，需要设置完整的文件路径：
Java
1kafka-topics.sh --bootstrap-server <SASL_SSL接入点地址> --list --command-config /***/kafka.properties- 生产者发送消息，输入命令后会出现">"的标识，此时输入内容并且按下回车即可发送消息：
Java
1kafka-console-producer.sh --bootstrap-server <SASL_SSL接入点地址> --topic <Topic名称> --producer.config /***/kafka.properties
- 消费者消费消息：
Java
1kafka-console-consumer.sh --bootstrap-server <SASL_SSL接入点地址> --topic <Topic名称> --group <消费组ID> --from-beginning --consumer.config /***/kafka.properties
- 如果通过上述命令能够正常发送和消费消息，则说明当前网络环境下能够通过SASL_SSL协议正常访问Kafka集群，可以参考开发指南中各种语言访问Kafka集群来实现更复杂的逻辑。