創建實例時開啟SASL_SSL訪問，則數據加密傳輸，安全性更高。
由于安全問題，支持的加密套件為TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256，TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256和TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256。
本節介紹如何使用開源的Kafka客戶端訪問開啟SASL的Kafka專享實例的方法。
說明：
使用SASL方式連接Kafka實例時，為了客戶端能夠快速解析實例的Broker，建議在客戶端所在主機的“/etc/hosts”文件中配置host和IP的映射關系。
其中，IP地址必須為實例連接地址（Broker地址），host為每個實例主機的名稱（您可以自定義主機的名稱，但不能重復）。
例如：
10.154.48.120 server01
10.154.48.121 server02
10.154.48.122 server03
已配置正確的安全組。
訪問開啟SASL的Kafka專享實例時，支持VPC內訪問。實例需要配置正確的安全組規則，具體安全組配置要求，請參考表3-2。
已獲取連接Kafka專享版實例的地址。
使用VPC內訪問，實例端口為9093，實例連接地址獲取如下圖。
獲取VPC內訪問Kafka專享實例的連接地址（實例已開啟SASL）
Kafka專享實例已創建Topic，否則請提前創建Topic。
已下載client.truststore.jks證書。如果沒有，在控制臺單擊Kafka實例名稱，進入實例詳情頁面，在“基本信息 > 高級配置 > Kafka SASL_SSL”所在行，單擊 。下載壓縮包后解壓，獲取壓縮包中的客戶端證書文件：client.truststore.jks。
已下載Kafka命令行工具1.1.0版本或者Kafka命令行工具2.3.0版本，確保Kafka實例版本與命令行工具版本相同。
已在Kafka命令行工具的使用環境中安裝Java Development Kit 1.8.111或以上版本，并完成環境變量配置。
以下操作命令以Linux系統為例進行說明。
解壓Kafka命令行工具。
進入文件壓縮包所在目錄，然后執行以下命令解壓文件。
tar -zxf [kafka_tar]
其中，[kafka_tar]表示命令行工具的壓縮包名稱。
例如：
tar -zxf kafka_2.11-1.1.0.tgz
修改Kafka命令行工具配置文件。
在Kafka命令行工具的“/config”目錄中找到“consumer.properties”和“producer.properties”文件，并分別在文件中增加如下內容。
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="**********" \
password="**********";
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
ssl.truststore.location=/opt/kafka_2.11-1.1.0/config/client.truststore.jks
ssl.truststore.password=dms@kafka
ssl.endpoint.identification.algorithm=
參數說明：
username和password為創建Kafka專享實例過程中開啟SASL_SSL時填入的用戶名和密碼。
ssl.trustore.location配置為client.truststore.jks證書的存放路徑。注意，Windows系統下證書路徑中也必須使用“/”，不能使用Windows系統中拷貝路徑時的“\”，否則客戶端獲取證書失敗。
ssl.truststore.password為服務器證書密碼，不可更改，需要保持為dms@kafka。
ssl.endpoint.identification.algorithm為證書域名校驗開關，為空則表示關閉。這里需要保持關閉狀態，必須設置為空。
進入Kafka命令行工具的“/bin”目錄下。
注意，Windows系統下需要進入“/bin/windows”目錄下。
執行如下命令進行生產消息。
./kafka-console-producer.sh --broker-list ${連接地址} --topic ${Topic名稱} --producer.config ../config/producer.properties
參數說明如下：
連接地址：從前提條件獲取的連接地址。
Topic名稱：Kafka實例下創建的Topic名稱。
如下示例，Kafka實例連接地址為“10.xxx.xxx.202:9093,10.xxx.xxx.197:9093,10.xxx.xxx.68:9093”。
執行完命令后，輸入需要生產的消息內容，按“Enter”發送消息到Kafka實例，輸入的每一行內容都將作為一條消息發送到Kafka實例。
[root@ecs-kafka bin]#./kafka-console-producer.sh --broker-list 10.xxx.xxx.202:9093,10.xxx.xxx.197:9093,10.xxx.xxx.68:9093 --topic topic-demo --producer.config ../config/producer.properties
>hello
>DMS
>Kafka!
>^C[root@ecs-kafka bin]#
如需停止生產使用Ctrl+C命令退出。
執行如下命令消費消息。
./kafka-console-consumer.sh --bootstrap-server ${連接地址} --topic ${Topic名稱} --group ${消費組名稱} --from-beginning --consumer.config ../config/consumer.properties
參數說明如下：
連接地址：從前提條件獲取的連接地址。
Topic名稱：Kafka實例下創建的Topic名稱。
消費組名稱：根據您的業務需求，設定消費組名稱。
示例如下：
[root@ecs-kafka bin]# ./kafka-console-consumer.sh --bootstrap-server 10.xxx.xxx.202:9093,10.xxx.xxx.197:9093,10.xxx.xxx.68:9093 --topic topic-demo --group order-test --from-beginning --consumer.config ../config/consumer.properties
hello
Kafka!
DMS
hello
^CProcessed a total of 4 messages
[root@ecs-kafka bin]#
如需停止消費使用Ctrl+C命令退出。
