我的单机zookeeper安装目录：/usr/local/zookeeper

我的kafka安装目录：/usr/local/kafka

一、

配置zookeeper的配置文件

1、如果是嵌入式zookeeper(kafka自带的zookeeper)

修改/usr/local/kafka/config/zookeeper.properties文件，添加如下配置

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

2、如果是非嵌入式zookeeper(单机安装的zookeeper服务)

修改/usr/local/zookeeper/conf/zoo.cfg文件，添加和上面一样的配置

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

二、创建jaas文件

在zookeeper的安装目录下的conf目录下创建zk_jaas.conf文件，并编辑内容

嵌入式zookeeper目录：/usr/local/kafka/config/zk_jaas.conf

单机zookeeper目录：/usr/local/zookeeper/conf/zk_jaas.conf

Server {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="admin"
   password="admin"
   user_admin="admin";
};

Client {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       username="admin"
       password="admin";
};

user_admin="admin"的含义是user_username="password",添加一个用户名为admin，密码为admin的认证用户，用户名和密码可以自己定义。

三、配置zookeeper客户端环境变量

因为如果要连接zookeeper的话是需要通过SASL认证的，所以需要配置环境变量，这里的环境变量主要是使用到了zk_jaas.conf文件中的Client配置，会在连接时使用用户名和密码。

1、嵌入式zookeeper：

修改/usr/local/kafka/bin/zookeeper-server-start.sh文件

添加配置后如下：

#...前面的内容省略
COMMAND=$1
case $COMMAND in
  -daemon)
     EXTRA_ARGS="-daemon "$EXTRA_ARGS
     shift
     ;;
 *)
     ;;
esac
export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/shakespeare/soft/zookeeper/apache-zookeeper-3.8.0-bin/conf/zk_jaas.conf ${KAFKAOPTS}" #这一行是要添加的内容
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS org.apache.zookeeper.server.quorum.QuorumPeerMain "$@"

2、单机版zookeeper：

在/user/local/zookeeper/conf目录下创建一个java.env的文件，并且编辑添加如下内容

java.env：

export JVMFLAGS="-Djava.security.auth.login.config=/usr/local/zookeeper/conf/zk_jaas.conf ${JVMFLAGS}"

四、重启zookeeper服务

正常启动一般便无问题。

五、通过客户端连接文件连接zookeeper服务

1、嵌入式zookeeper：

/usr/local/kafka/bin/zookeeper-shell.sh {hostname}:{port}

例如：./zookeeper-shell.sh 192.168.2.2:2181

2、单机版zookeeper

/usr/local/zookeeper/bin/zkCli.sh -server {hostname}:{port}

例如：./zkCli.sh -server 192.168.2.2:2181

一般连接成功会有如下日志显示：

Welcome to ZooKeeper!
2022-09-02 11:42:06,369 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.Login@317] - Client successfully logged in.
JLine support is enabled
2022-09-02 11:42:06,375 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.u.SecurityUtils@70] - Client will use DIGEST-MD5 as SASL mechanism.
2022-09-02 11:42:06,543 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.ClientCnxn$SendThread@1171] - Opening socket connection to server 192.168.2.2:2181.
2022-09-02 11:42:06,543 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.ClientCnxn$SendThread@1173] - SASL config status: Will attempt to SASL-authenticate using Login Context section 'Client'
[zk: 192.168.2.2:2181(CONNECTING) 0] 2022-09-02 11:42:06,564 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.ClientCnxn$SendThread@1005] - Socket connection established, initiating session, client: 192.168.2.2:56776, server: 192.168.2.2:2181
2022-09-02 11:42:06,585 [myid:192.168.2.2:2181] - INFO  [main-SendThread(192.168.2.2:2181):o.a.z.ClientCnxn$SendThread@1444] - Session establishment complete on server 192.168.2.2:2181, session id = 0x10003a960540000, negotiated timeout = 30000

WATCHER::

WatchedEvent state:SyncConnected type:None path:null

WATCHER::

WatchedEvent state:SaslAuthenticated type:None path:null

有较明显的Will attempt to SASL-authenticate using Login Context section 'Client'的提示

至此zookeeper的SASL配置完毕。

六、如何设置权限

zookeeper的sasl和普通的认证授权还不太一样

1、zookeeper默认允许匿名用户访问，如果不想让匿名用户访问某些则需要给节点单独设置ACL权限，配置完SASL后，连接zk客户端，然后可以给每个节点设置acl权限

setAcl /path sasl:admin:crdwa

2、zookeeper的3.6.0版本之后新增了一个环境变量sessionRequireClientSASLAuth

这个环境变量为true时要求客户端连接zk时必须进行SASL认证才可以连接成功，也就是说没有进行SASL认证的匿名用户就无法连接了，比第一步给每个节点设置ACL更方便一些，相当于在连接时设置了一个登录密码。

可以在单机版的zoo.cfg或嵌入式的zookeeper.properties里添加如下配置：

sessionRequireClientSASLAuth=true

7、kafka如何连接

配置完以上的认证后，启动kafka可能会报错。

原因：因为kafka依赖于zookeeper，要进行节点的访问，但是配置完后kafka也失去了权限，因此无法正常启动。

解决方法：在kafka安装目录下：/usr/local/kafka/bin/kafka-run-class.class目录下添加环境变量，如下：

#...前面的内容省略
base_dir=$(dirname $0)/..
KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/zookeeper/conf/zk_jaas.conf ${KAFKA_OPTS}" #要添加的行

if [ -z "$SCALA_VERSION" ]; then
  SCALA_VERSION=2.12.10
fi

然后重启kafka即可解决，原理和zk客户端连接原理一样，需要配置连接时的Client的用户名和密码，用户名和密码都在zk_jaas.conf文件里。