前兩篇部落格主要講配置Kafka SASL認證，已經說明了，配置後Kafka自帶的工具很多都不能使用。本文主要講三個常用的Kafka附帶工具的安全配置，在完成安全認證配置後，可以繼續使用這些工具。本文基於前面的。確保Kafka叢集已經配置了SASL PLAIN且能順利執行。

本文將說明以下三個kafka工具：

1、kafka-console-consumer.sh和kafka-console-producer.sh

一個簡單的消費者程式，通過給定的引數，消費某個kafka叢集的指定topic資料。在我們配置了安全認證策略後，無論Kafka消費還是生產者，都必須要指定使用者名稱密碼才可。而同上兩篇部落格一樣，我們的配置和基本的生產者、消費者程式沒什麼兩樣。

（1）編寫JAAS配置檔案

假如我們使用的Kafka使用者名稱是produce，密碼是prod-sec，則可以寫一個JAAS檔案，我命名為kafka_client_jaas.conf檔案：

KafkaClient {  
    org.apache.kafka.common.security.plain.PlainLoginModule required  
            username="producer"  
            password="prod-sec";  
};  

（2）加入環境變數

環境變數java.security.auth.login.config必須在執行kafka-console-consumer.sh之前被設定，且設定成我們的kafka_client_jaas.conf配置檔案路徑。為方便，我們可以將kafka-console-consumer.sh指令碼再包裝一下，在執行前設定這個變數。建立一個kafka-console-consumer-sasl.sh指令碼。

#!/bin/bash

ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_client_jaas.conf "
$ROOT/bin/kafka-console-consumer.sh --new-consumer "[email protected]"

將這個指令碼放置在Kafka部署根目錄上即可。看最後一條語句，給kafka-console-consumer.sh指令碼傳入了--new-consumer選項，這是因為之前老版本的Kafka java API不支援安全策略，屬於Kafka專案中的一些坑，大概是0.9版本以前的API。我們指定這個引數後，工具將使用新的java API開啟消費者，也支援安全機制。注意，這個選項很重要。

（3）建立配置檔案

和普通的生產者、消費者程式一樣，開啟安全策略後，必須手動指定幾個配置項，視認證機制而定，我們的Kafka配置的是SASL_PLAIN，建立一個配置檔案，名為consumer.properties，放置於kafka部署目錄的conf下，若存在這個配置檔案直接修改亦可。新增如下配置項：

security.protocol = SASL_PLAINTEXT
sasl.mechanism = PLAIN

（4）嘗試執行

按上述步驟完成後，我們嘗試消費kafka叢集中的eztest主題，切換到kafka部署根目錄，示例命令如下：

./kafka-console-consumer-sasl.sh --zookeeper vubuntuez1:2181,vdebianez3:2181,vcentosez2:2181,vdebianez4:2181 \
--bootstrap-server vubuntuez1:9092,vdebianez3:9092,vcentosez2:9092 \
--topic eztest --consumer.config ./config/consumer.properties

指令碼是第二步建立的，--zookeeper和--bootstrap-server引數都需要。--topic指定為eztest，--consumer.conf這個選項必須指定，也就是指定我們在第三步建立的consumer.properties配置檔案。如此，只要topic中有資料，都可以列印到標準輸出流。

與kafka-console-consumer.sh相同，kafka-console-producer.sh也如此配置，創建出一個kafka-console-producer-sasl.sh指令碼即可。

2、kafka-consumer-groups.sh

這個工具主要用於檢視某些消費者組，消費者組詳細消費情況等。屬於一個消費者程式。我們同樣使用如上的方法配置。編寫JAAS配置檔案和consumer.properties配置檔案，同上。再嘗試寫出一個啟動指令碼，我命名為kafka-consumer-groups-sasl.sh：

#!/bin/bash

ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_client_jaas.conf "
$ROOT/bin/kafka-consumer-groups.sh --new-consumer "[email protected]"

這裡我們也需要增加一個--new-consumer選項，非常重要。

嘗試執行如下示例命令：

./kafka-consumer-groups-sasl.sh --bootstrap-server vubuntuez1:9092,vdebianez3:9092,vcentosez2:9092 \
 --list --command-config ./config/consumer.properties

我們必須指定一個--command-config選項，值就是上面反覆提到的consumer.properties配置檔案路徑。這樣，可以檢視到當前kafka中所有的消費者組。

其他工具比如kafka-run-classes.sh有時我們會使用它啟動一些Kafka內部類，如GetOffsetShell等，但目前沒有找到這個kafka-run-classes.sh的配置檔案入口，有些內部類貌似也沒有提供這個配置檔案選項，比如GetOffsetShell。