当前位置： 首页 > news >正文

深入解析：kafka 2.X+zookeeper3.X 权限控制

news 2026/3/26 18:12:20

Kafka 和 Zookeeper 权限控制

1️⃣ Kafka 和 Zookeeper 权限控制的概念

（1）Kafka 权限控制

Kafka 权限控制主要针对 主题（topic）和消费组（consumer group）。核心概念是 ACL（Access Control List，访问控制列表）。

  • 目的：控制谁能读写某个 topic，谁能创建或删除 topic，以及谁能操作消费组。

  • 实现方式：通过 SASL+ACL，也可以结合 SSL 客户端证书认证。

  • 操作粒度：

    • TOPIC：读（Read）、写（Write）、创建（Create）、删除（Delete）、修改配置（Alter）、描述（Describe）等。

    • GROUP：消费组操作权限，如提交 offset、加入消费组。

    • CLUSTER：集群管理操作权限，如创建 topic、删除 topic、查看 broker 状态。

示例：

资源类型用户权限
TOPICUser:producerWRITE
TOPICUser:consumerREAD
GROUPUser:consumerREAD

（2）Zookeeper 权限控制

Zookeeper 是 Kafka 的元数据存储（存储 topic、分区信息、消费者偏移量等）。如果不做权限控制，任何人都能连接 ZK 修改元数据，风险很大。

  • Zookeeper 权限控制基于 ACL + Scheme：

    • world:anyone → 任何人都可访问

    • auth → 认证用户可访问

    • digest → 用户名+密码认证

  • Kafka 使用：Kafka Broker 使用 Zookeeper 存储元数据时，会配置 zookeeper.set.acl=true，并通过 SASL/Digest 用户认证。

2️⃣ 为什么需要权限控制

权限控制的必要性主要有三点：

  1. 安全性

    • 阻止未授权用户读取或写入敏感数据，例如生产者向 topic 写入消息，消费者读取消息。

    • 阻止恶意用户篡改元数据（topic、分区、副本分配等）。

  2. 数据完整性

    • 只有有权限的用户才能修改 topic 配置、消费组偏移量，避免错误操作破坏业务。

  3. 审计与多租户支持

    • 企业环境中不同团队或服务可能共享 Kafka 集群，需要严格控制访问权限，保证每个团队只能操作自己的 topic。

3️⃣ Kafka 权限控制流程

  1. 认证（Authentication）

    • 确认客户端身份，例如：

      • SASL/SCRAM：用户名+密码

      • SSL：证书认证

    • 客户端首先通过 SASL 验证身份

  2. 授权（Authorization）

    • Broker 根据 ACL 判断该用户是否允许执行该操作：

      • 写入 topic？

      • 读取 topic？

      • 操作消费组？

  3. ACL 配置方式

    • 使用 Kafka 内置工具 kafka-acls.sh 添加/修改/删除 ACL

    • ACL 记录存储在 Zookeeper 或 Kafka 内部（取决于版本）

好的，我帮你把你整理的 Kafka 2.x + Zookeeper 3.x 权限控制步骤精简、系统化，并加上逻辑顺序说明，方便直接参考或执行：

2 Kafka + Zookeeper 权限控制实操（2.x + 3.x）

1️⃣ Zookeeper 权限控制

（1）配置 Zookeeper 启用 SASL

编辑 zoo.cfg：

tickTime=2000
initLimit=1
syncLimit=5
dataDir=/data/var/lib/zookeeper/data
dataLogDir=/data/var/lib/zookeeper/datalog
clientPort=2181
admin.serverPort=8888
maxClientCnxns=3000
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
4lw.commands.whitelist=conf,stat,srvr,mntr,envi
# 启用 SASL
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

（2）配置 Zookeeper JAAS

创建 zk_jaas.conf：

Server {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="admin"password="HDKJ!@#2011"user_kafka="HDKJ!@#2011";
};

（3）修改 Zookeeper 启动脚本

在 zkEnv.sh 最后一行添加：

export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/data/software/zookeeper/conf/zk_jaas.conf"

启动zk

./bin/zkServer.sh start

2️⃣ Kafka SASL 用户配置

（1）创建或修改 SCRAM 用户

./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name admin
./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name producer
./bin/kafka-configs.sh --zookeeper yourIP:2181 \--alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=HDKJ!@#2011],SCRAM-SHA-512=[password=HDKJ!@#2011]' \--entity-type users --entity-name consumer

（2）查看用户信息

./bin/kafka-configs.sh --zookeeper yourIP:2181 --describe --entity-type users

3️⃣ Kafka JAAS 配置

创建 kafka_server_jaas.conf：

KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="HDKJ!@#2011"user_admin="HDKJ!@#2011"user_producer="HDKJ!@#2011"user_consumer="HDKJ!@#2011";
};
KafkaClient {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="HDKJ!@#2011"user_producer="HDKJ!@#2011"user_consumer="HDKJ!@#2011";
};
Client {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="kafka"password="HDKJ!@#2011";
};

4️⃣ Kafka Broker 配置

编辑 server.properties：

broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://yourIP:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
zookeeper.connect=yourIP:2181

（2）Kafka 启动脚本添加 JAAS

在 kafka-server-start.sh 最后一行：

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/data/software/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

5️⃣ Kafka 客户端配置

（1）Consumer/Producer 配置文件

consumer.conf：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \username="consumer" \password="HDKJ!@#2011";

producer.conf：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \username="producer" \password="HDKJ!@#2011";

（2）Console 工具脚本添加 JAAS

在 kafka-console-producer.sh 和 kafka-console-consumer.sh 中：

export KAFKA_OPTS="-Djava.security.auth.login.config=/data/software/kafka/config/kafka_server_jaas.conf"

6️⃣ Kafka ACL 配置

（1）创建 Topic

./bin/kafka-topics.sh --zookeeper yourIP:2181 --create --replication-factor 1 --partitions 1 --topic test_topic

（2）配置写权限（Producer）

./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:producer --operation Write --topic test_topic

（3）配置读权限（Consumer）

# Topic 读取权限
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:consumer --operation Read --topic test_topic
# 消费组权限
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=yourIP:2181 \--add --allow-principal User:consumer --operation Read --group test-group

7️⃣ 启动 Kafka 集群

./bin/kafka-server-start.sh -daemon ./config/server.properties
tail -100f logs/server.log

8️⃣ 测试

（1）写入消息

./bin/kafka-console-producer.sh --bootstrap-server yourIP:9092 --topic test_topic --producer.config /data/software/kafka/config/producer.conf

（2）读取消息

./bin/kafka-console-consumer.sh --bootstrap-server yourIP:9092 --topic test_topic --group test-group --from-beginning --consumer.config /data/software/kafka/config/consumer.conf

好的，我帮你画一张 Kafka 2.x + Zookeeper 3.x 权限控制流程图 的概念图，清楚展示认证和授权流程。

9，Kafka + Zookeeper 权限控制流程图（概念）

            +-----------------+|   Kafka Client  ||  Producer/Consumer|  JAAS 配置      ||  username/password+--------+--------+|| SASL认证请求 (SCRAM)v+-----------------+|   Kafka Broker  ||  JAAS 配置      ||  user_admin=...,||  user_producer=...|+--------+--------+|认证成功?     |        ACL检查+------------v----------------+|  ACL授权层                  ||  Topic权限/Consumer Group权限|+------------+----------------+|授权通过?    |+------------v----------------+|  Kafka Topic/Partition       ||  消息读写操作               |+-----------------------------+^|+-----------------+| Zookeeper       || 用户验证 (Digest/SASL)| 元数据管理       |+-----------------+

流程说明

  1. 客户端认证

    • 客户端通过 sasl.jaas.config 提供用户名和密码。

    • Kafka Broker 根据自身 JAAS 配置和 SCRAM 存储信息验证身份。

  2. Zookeeper 验证

    • Broker 访问 ZK 获取元数据或提交 offset，也需要 ZK SASL 认证。

  3. ACL 授权

    • Broker 检查客户端请求的 Topic/Consumer Group 是否有权限。

    • 通过 kafka-acls.sh 配置的 ACL 决定是否允许读写。

  4. 消息读写

    • 认证和授权成功后，客户端才可以进行消息生产或消费。

借鉴：Apache zookeeper kafka 开启SASL安全认证 —— 筑梦之路_kafka sasl认证-CSDN博客

http://www.jsqmd.com/news/56212/

相关文章：

  • P10470 前缀统计
  • 天气+快递查询前端页面制作步骤（HTML,css）
  • U3D测试
  • AI硬件与芯片之争：从夸克眼镜到GPU通用性
  • 数据库故障的诊断方法与分析思路：实战经验总结
  • P4391 无线传输
  • 查看虚拟环境里安装了哪些包
  • 2025/11/29
  • 代码背后的哲学
  • 编程之外的修行
  • 2025/11/28
  • 《代码大全2》读书笔记4
  • 2025/11/30
  • カワキヲアメク
  • 《代码大全2》读书笔记5
  • 完整教程：第162期 自定义目标检测的 YOLO 微调完整指南
  • OpenCSG发布AgenticHub｜一站式企业AI智能体平台
  • 读书笔记5
  • 完整教程：LeetCode 413 - 等差数列划分
  • 为什么在anacnda prompt里面运行了激活虚拟环境的命令，却还是没有切换成虚拟环境？
  • 在线调试+JMeter联动（以万维易源接口为例）
  • 征程 6 | QAT 新版 qconfig 量化模板使用教程
  • 计算机毕设java幼儿园校车管理高效的系统 基于Java的幼儿园校车信息管理系统设计与实现 Java环境下幼儿园校车运营管理平台开发
  • 在线调试+JMeter联动（以聚合数据快递接口为例）
  • 手艺文档搭建实战：基于PandaWiki的五步自动化方案
  • JMeter调用中国天气网接口查看天气预报
  • AI元人文构想：回应《自动驾驶技术的伦理认同与社会化应用治理》——规则库的范式分野与价值原语化的理论必然
  • NOIP 2025 游记 退役记 回忆录
  • ML - F1 score
  • 多模态技术深度探索：融合视觉与语言的AI新范式 - 详解