转载请注明出处：http://blog.csdn.net/u012842205/article/details/73648170

前两篇博客主要讲配置Kafka SASL认证，已经说明了，配置后Kafka自带的工具很多都不能使用。本文主要讲三个常用的Kafka附带工具的安全配置，在完成安全认证配置后，可以继续使用这些工具。本文基于前面的集群Kafka配置SASL用户名密码认证。确保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 "$@"

将这个脚本放置在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 "$@"

这里我们也需要增加一个--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。