wget https://download.oracle.com/java/21/latest/jdk-21_linux-x64_bin.tar.gz
tar xf jdk-21_linux-x64_bin.tar.gz
mv jdk-21.0.3/ /usr/local/
echo "export JAVA_HOME=/usr/local/jdk-21.0.3" >> /etc/profile
source /etc/profile
wget https://www.apache.org/dyn/closer.lua/zookeeper/zookeeper-3.9.2/apache-zookeeper-3.9.2-bin.tar.gz
tar xf apache-zookeeper-3.9.2-bin.tar.gz
mv apache-zookeeper-3.9.2-bin/ /usr/local/
cd /usr/local/apache-zookeeper-3.9.2-bin/
cp ./conf/zoo_sample.cfg ./conf/zoo.cfg
./bin/zkServer.sh start
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz
tar xf kafka_2.13-3.7.0.tgz
mv kafka_2.13-3.7.0 /usr/local/
/usr/local/kafka_2.13-3.7.0/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.13-3.7.0/config/server.properties
server.properties
在Kafka服务器的config
目录下找到server.properties
文件,并添加或修改以下配置:
# broker监听器的安全协议配置
listeners=SASL_PLAINTEXT://:9092
# 通信时使用使用SASL安全协议进行认证
security.inter.broker.protocol=SASL_PLAINTEXT
# broker之间通信时使用的SASL机制;PLAIN(简单的用户名/密码认证机制)
sasl.mechanism.inter.broker.protocol=PLAIN
# 启用的机制,可以添加多个,用逗号分隔
sasl.enabled.mechanisms=PLAIN
# 如果需要限制哪些用户可以访问Kafka集群
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
# 没有找到匹配的ACL,则允许所有用户访问
allow.everyone.if.no.acl.found=true
# Kafka集群中的超级用户,可访问集群中所有资源的权限,不受ACL的限制
super.users=User:admin
# 如果Kafka服务器和客户端不在同一台机器上,需要设置advertised.listeners
advertised.listeners=SASL_PLAINTEXT://172.30.30.4:9092
注意:listeners
和advertised.listeners
的值应该根据你的网络配置进行调整。
创建一个JAAS配置文件(例如kafka_server_jaas.conf
),并添加以下内容:
[root@app config]# pwd
/usr/local/kafka_2.13-3.7.0/config
[root@app config]# cat kafka_server_jaas.conf
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin";
};
这里定义了Kafka服务器的SASL用户。username
和password
是Kafka服务器用于内部通信的凭据(在大多数情况下,这些凭据不会被客户端直接使用)。user_admin
定义了其他可以访问Kafka的用户及其密码。
在启动Kafka服务时,需要指定JAAS配置文件的路径。这通常是通过修改Kafka的启动脚本来实现的(如kafka-server-start.sh
)。在脚本中添加或修改KAFKA_OPTS
环境变量:
[root@app config]# tail -n1 /usr/local/kafka_2.13-3.7.0/bin/kafka-server-start.sh
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/usr/local/kafka_2.13-3.7.0/config/kafka_server_jaas.conf kafka.Kafka "$@"
配置通用认证凭据
创建pub-acl.properties
文件并添加:
[root@app config]# cat pub-acl.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin";
修改完配置后,需要重启Kafka服务以使配置生效。
使用配置好的客户端连接到Kafka服务器,检查是否能够成功连接并进行消息的生产和消费。
# 查看所有队列
[root@app ~]# /usr/local/kafka_2.13-3.7.0/bin/kafka-topics.sh --list --bootstrap-server 172.30.30.4:9092 --command-config /usr/local/kafka_2.13-3.7.0/config/pub-acl.properties
# 消费队列
[root@app ~]# /usr/local/kafka_2.13-3.7.0/bin/kafka-console-consumer.sh --bootstrap-server 172.30.30.4:9092 --topic wx_log --consumer.config /usr/local/kafka_2.13-3.7.0/config/pub-acl.properties