单机安装kafka（含账号密码认证）

单机安装kafka完整版

需要准备如下：
安装JDK1.8+，配置JAVA_HOME (CentOS 64bit)
配置主机名和IP映射【可选】
关闭防火墙 | 防火墙开机自启动
安装&启动Zookeeper【使用kafka安装包自带的即可】

1、下载kafka安装包

官方下载地址

点击kafka_2.12-3.8.0.tgz自动下载

Kafka 的 Scala 版本指的是 Apache Kafka 的代码库使用 Scala 编程语言编写的部分。Kafka 是用 Java 和 Scala 混合开发的，主要的服务和功能是用 Java 实现的，但一些组件和库，特别是在流处理方面，使用了 Scala。

2、服务器部署

上传至服务器提前创建好的部署目录并解压:

tar -zxvf kafka_2.12-3.8.0.tgz

如有需要可配置软链:

ln -s kafka_2.12-3.8.0.tgz kafka

配置

KAFKA_HOME

环境变量，以及将

$KAFKA_HOME/bin

文件夹加入

PATH

环境变量中（环境变量的配置为可选项最好配置学习时可不配）:

vim /etc/profile

尾部添加如下:

exportKAFKA_HOME=/export/server/kafka
exportPATH=:$PATH:${KAFKA_HOME}

生效环境变量:

source /etc/profile

进入

config

目录下编辑

kafka

核心配置文件

server.properties

以及

zk

的核心配置文件

zookeeper.properties

:

vim server.properties

#没有单独部署zk集群的话就直接配本地zk，后面直接启动kafka自带的zk其中127.0.0.1为本机ip
zookeeper.connect=127.0.0.1:2181
#kafka数据存储目录workDir改成你的安装目录
log.dirs=/workDir/kafka/app/kafka_2.12-3.8.0/data
#配置kafka的监听端口（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
listeners=SASL_PLAINTEXT://127.0.0.1:9092

#---------以下配置是在需要用户密码认证时配置，不需要直接删除----------
#使用的认证协议（SASL_PLAINTEXT：动态增加用户协议，PLAINTEXT 不能动态增加用户）
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制,简单账号密码认证配置
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
#设置管理员账号
super.users=User:admin

vim zookeeper.properties

#zk数据存储目录
dataDir=/workDir/kafka/zookeeper/apache-zookeeper-3.5.7-bin/data

在

config

目录下创建用于用户密码认证的认证文件

kafka-server-jaas.conf

、

kafka-zk-jaas.conf

kafka-server-jaas.conf

—用于kafka的认证:

KafkaServer={
        org.apache.kafka.common.security.plain.PlainLoginModule=required
        username="admin"
        password="admin-pwd"
        user_admin="admin-pwd"
        user_broker0="broker0-pwd";
};=

kafka-zk-jaas.conf

—用于集群broker之间的认证:

zookeeper={
        org.apache.kafka.common.security.plain.PlainLoginModule=required
        username="admin"
        password="admin-pwd";
};=

其中username和password是配置超管的账号密码， user_broker0="broker0-pwd"可以配置多个，格式为user_账号名=密码

进入

bin

目录下修改

zk

和

kafka

的启动脚本:

vim zookeeper-server-start.sh

直接在首行添加如下内容:

#注意配置文件的路径改成你刚才创建的那个文件的目录exportKAFKA_OPTS=" -Djava.security.auth.login.config=/iflytek/dmxu2/kafka/app/kafka_2.12-3.8.0/config/kafka-zk-jaas.conf"

vim kafka-server-start.sh

直接在首行添加如下内容:

#注意配置文件的路径改成你刚才创建的那个文件的目录exportKAFKA_OPTS=" -Djava.security.auth.login.config=/iflytek/dmxu2/kafka/app/kafka_2.12-3.8.0/config/kafka-server-jaas.conf"

启动zk和kafka，先启动zk

#启动zk
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties
#启动kafka
./bin/kafka-server-start.sh -daemon config/server.properties

#如果不成功可以去掉-daemon改为前台启动观察日志
./bin/kafka-server-start.sh config/server.properties

kafka提供了各种脚本

./bin/kafka-server-stop.sh=#关闭kafka
kafka-console-consumer.sh=#消费命令
kafka-console-producer.sh=#生产命令
kafka-consumer-groups.sh=#查看消费者组，重置消费位点等
kafka-topics.sh=#查询topic状态，新建，删除，扩容
kafka-acls.sh=#配置，查看kafka集群鉴权信息
kafka-configs.sh=#查看，修改kafka配置
kafka-mirror-maker.sh=#kafka集群间同步命令
kafka-preferred-replica-election.sh=#重新选举topic分区leader
kafka-producer-perf-test.sh=#kafka自带生产性能测试命令
kafka-reassign-partitions.sh=#kafka数据重平衡命令
kafka-run-class.sh=#kafka执行脚本

给broker0账户赋予topic=[topic_name]的写权限

./kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:broker0 --operation Write --topic topic_name

给broker0账户赋予topic=[topic_name]的读权限

./kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:broker0 --operation Read --topic topic_name

修改consumer.properties中的group.id (默认为 test-consumer-group)

# consumer group id
group.id=group_id

给broker0账户赋予group=[group_id]读授权 (外部连接kafka需要指定group.id，此时必须对group进行授权，否则无法通过授权)

./kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=127.0.0.1:2181 --add --allow-principal User:broker0 --operation Read --group group_id

查看权限

./kafka-acls.sh --list --authorizer-properties zookeeper.connect=127.0.0.1:2181

Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=group_name, patternType=LITERAL)`:(principal=User:broker0, host=*, operation=READ, permissionType=ALLOW)
 
Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=group_id, patternType=LITERAL)`:(principal=User:broker0, host=*, operation=READ, permissionType=ALLOW)
 
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=topic_name, patternType=LITERAL)`:(principal=User:broker0, host=*, operation=READ, permissionType=ALLOW)(principal=User:broker0, host=*, operation=WRITE, permissionType=ALLOW)

ACL命令操作(–operation)对应功能：
OperationResourceAPIALTERTopicCreatePartitionsALTER_CONFIGSTopicAlterConfigsCREATETopicMetadata if auto.create.topics.enableCREATETopicCreateTopicsDELETETopicDeleteRecordsDELETETopicDeleteTopicsDESCRIBETopicListOffsetsDESCRIBETopicMetadataDESCRIBETopicOffsetFetchDESCRIBETopicOffsetForLeaderEpochDESCRIBE_CONFIGSTopicDescribeConfigsREADTopicFetchREADTopicOffsetCommitREADTopicTxnOffsetCommitWRITETopicProduceWRITETopicAddPartitionsToTxnDELETEGroupDeleteGroupsDESCRIBEGroupDescribeGroupDESCRIBEGroupFindCoordinatorDESCRIBEGroupListGroupsREADGroupAddOffsetsToTxnREADGroupHeartbeatREADGroupJoinGroupREADGroupLeaveGroupREADGroupOffsetCommitREADGroupOffsetFetchREADGroupSyncGroupREADGroupTxnOffsetCommit
启动生产者和消费者

./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic topic_name --from-beginning --consumer.config ../config/consumer.properties

./kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic topic_name --producer.config ../config/producer.properties