==作者：YB-Chi==

[toc]

所需组件

JDK1.8
Zookeeper3.4.6
kafka_2.11-0.10.0.0

安装jdk

配置jdk要先卸载centos自带的openjdk 先查看 rpm -qa | grep java 显示如下信息：

[root@localhost /]# rpm -qa | grep java
java-1.8.0-openjdk-headless-1.8.0.65-3.b17.el7.x86_64
javapackages-tools-3.4.1-11.el7.noarch
java-1.8.0-openjdk-1.8.0.65-3.b17.el7.x86_64
tzdata-java-2015g-1.el7.noarch
java-1.7.0-openjdk-1.7.0.91-2.6.2.3.el7.x86_64
java-1.7.0-openjdk-headless-1.7.0.91-2.6.2.3.el7.x86_64
python-javapackages-3.4.1-11.el7.noarch

卸载

rpm -e --nodeps java-1.8.0-openjdk-headless-1.8.0.65-3.b17.el7.x86_64
rpm -e --nodeps java-1.8.0-openjdk-1.8.0.65-3.b17.el7.x86_64
rpm -e --nodeps java-1.7.0-openjdk-1.7.0.91-2.6.2.3.el7.x86_64
rpm -e --nodeps java-1.7.0-openjdk-headless-1.7.0.91-2.6.2.3.el7.x86_64

解压jdk在/usr/local并配置环境变量

vim /etc/profile
在最后添加
JAVA_HOME=/usr/local/jdk_1.8
PATH=$PATH:$JAVA_HOME/bin:
CLASSPATH=.:$JAVA_HOME/lib
export JAVA_HOME  PATH CLASSPATH
更新配置文件
source /etc/profile
验证
java -version
javac

安裝Zookeeper

解压tar zxvf zookeeper-3.4.6.tar.gz

重命名mv zookeeper-3.4.6 zookeeper

修改配置文件
cp zoo_simple.cfg zoo.cfg
vim zoo.cfg
修改dataDir	zk的目录
dataDir=/usr/local/zookeeper/data
创建data目录

配置zk的环境变量(参考添加)
vim /etc/profile
export ZK_HOME=/usr/local/zookeeper
PATH=$PATH:$JAVA_HOME/bin:$ZK_HOME/bin:

更新配置文件
source /etc/profile

启动
zkServer.sh start
zk使用
可以使用命令zkCli.sh进入zk客户端

图示成功

安装Kafka

解压在/usr/local 重命名为kafka方便修改

mkdir -p /var/local/kafka/data/#创建kafka的数据目录
cd /usr/local/kafka/config #进入配置目录 

vi server.properties #编辑修改相应的参数 

broker.id=0 #注意此id为kafka服务器的唯一标示  不要与客户kafka 的id重复  只要是整数即可
port=9092								 #这个参数很有意思  即使listeners里有端口号  也必须配置port 否则偶尔会出现问题
listeners = PLAINTEXT://192.168.56.212:9092
log.dirs=/var/local/kafka/data/ #日志存放路径，上面创建的目录 
zookeeper.connect=192.168.56.212:2181 #zookeeper地址和端口，单机配置部署，ip:2181 

启动
cd /usr/local/kafka
bin/kafka-server-start.sh config/server.properties

脚本

cd /usr/local/kafka

vim kafkastart.sh
#启动kafka
/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

vim kafkastop.sh
#关闭kafka
/usr/local/kafka/bin/kafka-server-stop.sh /usr/local/kafka/config/server.properties &

#添加脚本执行权限
chmod +x kafkastart.sh
chmod +x kafkastop.sh

开启SASL认证(未成功)

• 添加配置文件

cd config/
#服务端配置
vim kafka_server_jaas.conf

KafkaServer{
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_admin="admin"
    user_alice="alice";
};#此处的用户和密码和linux账号无任何关系   

#客户端配置
vim kafka_client_jaas.conf

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="admin";
};#必须为服务端配置里存在的账号和密码

• 更改资源文件

#核心文件server.properties
vim server.properties
#在修改原有的listeners=PLAINTEXT://192.168.56.212:9092   并在其下加入sasl认证的配置
listeners=SASL_PLAINTEXT://192.168.56.212:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin

#客户端资源文件   用于命令启动sh客户端进行验证使用
vim producer.properties
#在最后加入
security.protocol = SASL_PLAINTEXT
sasl.mechanism = PLAIN

vim consumer.properties
#在最后加入
security.protocol = SASL_PLAINTEXT
sasl.mechanism = PLAIN

• 添加启动脚本

cd ../bin/

vim kafka-run-class.sh
#在最后加入KAFKA_SASL_OPTS变量值并在下方引用    注意变量引用的配置文件路径

# Launch mode
KAFKA_SASL_OPTS='-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_server_jaas.conf'
if [ "x$DAEMON_MODE" = "xtrue" ]; then
  nohup $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_SASL_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
  exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_SASL_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@"
fi

#建立sasl认证的生产者客户端
vim kafka-console-producer-sasl.sh

#!/bin/bash
#启用sasl验证生产脚本
export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_client_jaas.conf"
/usr/local/kafka/bin/kafka-console-producer.sh "$@"

#建立sasl认证的消费者客户端
vim kafka-console-consumer-sasl.sh

#!/bin/bash
#启用sasl验证消费脚本
export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_client_jaas.conf"
/usr/local/kafka/bin/kafka-console-consumer.sh --new-consumer "$@"

• 验证

使用命令对topic赋予指定用户的读写权限
启动sasl的客户端进行认证 
本文不提供命令    参考kafka记录中SHELL标题