RocketMQ中ACL权限控制

2024-09-04 03:08
文章标签 控制 rocketmq 权限 acl

本文主要是介绍RocketMQ中ACL权限控制，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

1、什么是ACL?

ACL是access control list的简称，俗称访问控制列表。访问控制，基本上会涉及到用户、资源、权限、角色等概念，那在RocketMQ中上述会对应哪些对象呢？

  • 用户
    用户是访问控制的基础要素，也不难理解，RocketMQ ACL必然也会引入用户的概念，即支持用户名、密码。
  • 资源
    资源，需要保护的对象，在RocketMQ中，消息发送涉及的Topic、消息消费涉及的消费组，应该进行保护，故可以抽象成资源。
  • 权限
    针对资源，能进行的操作，
  • 角色
    RocketMQ中，只定义两种角色：是否是管理员。

另外，RocketMQ还支持按照客户端IP进行白名单设置。

2、ACL基本流程图

在讲解如何使用ACL之前，我们先简单看一下RocketMQ ACL的请求流程：
在这里插入图片描述
对于上述具体的实现，将在后续文章中重点讲解，本文的目的只是希望给读者一个大概的了解。

3、如何配置ACL

3.1 acl配置文件

acl默认的配置文件名：plain_acl.yml,需要放在${ROCKETMQ_HOME}/store/config目录下。下面对其配置项一一介绍。

3.1.1 globalWhiteRemoteAddresses

全局白名单，其类型为数组，即支持多个配置。其支持的配置格式如下：

  • 空
    表示不设置白名单，该条规则默认返回false。
  • "*"
    表示全部匹配，该条规则直接返回true，将会阻断其他规则的判断，请慎重使用。
  • 192.168.0.{100,101}
    多地址配置模式，ip地址的最后一组，使用{}，大括号中多个ip地址，用英文逗号(,)隔开。
  • 192.168.1.100,192.168.2.100
    直接使用,分隔，配置多个ip地址。
  • 192.168..或192.168.100-200.10-20
    每个IP段使用 "*" 或"-"表示范围。

3.1.2 accounts

配置用户信息，该类型为数组类型。拥有accessKey、secretKey、whiteRemoteAddress、admin、defaultTopicPerm、defaultGroupPerm、topicPerms、groupPerms子元素。

3.1.2.1 accessKey

登录用户名，长度必须大于6个字符。

3.1.2.2 secretKey

登录密码。长度必须大于6个字符。

3.1.2.3 whiteRemoteAddress

用户级别的IP地址白名单。其类型为一个字符串，其配置规则与globalWhiteRemoteAddresses，但只能配置一条规则。

3.1.2.4 admin

boolean类型，设置是否是admin。如下权限只有admin=true时才有权限执行。

  • UPDATE_AND_CREATE_TOPIC
    更新或创建主题。
  • UPDATE_BROKER_CONFIG
    更新Broker配置。
  • DELETE_TOPIC_IN_BROKER
    删除主题。
  • UPDATE_AND_CREATE_SUBSCRIPTIONGROUP
    更新或创建订阅组信息。
  • DELETE_SUBSCRIPTIONGROUP
    删除订阅组信息。

3.1.2.5 defaultTopicPerm

默认topic权限。该值默认为DENY(拒绝)。

3.1.2.6 defaultGroupPerm

默认消费组权限，该值默认为DENY(拒绝)，建议值为SUB。

3.1.2.7 topicPerms

设置topic的权限。其类型为数组，其可选择值在下节介绍。

3.1.2.8 groupPerms

设置消费组的权限。其类型为数组，其可选择值在下节介绍。可以为每一消费组配置不一样的权限。

3.2 RocketMQ ACL权限可选值

  • DENY
    拒绝。
  • PUB
    拥有发送权限。
  • SUB
    拥有订阅权限。

3.3、权限验证流程

上面定义了全局白名单、用户级别的白名单，用户级别的权限，为了更好的配置ACL权限规则，下面给出权限匹配逻辑。
在这里插入图片描述

4、使用示例

4.1 Broker端安装

首先，需要在broker.conf文件中，增加参数aclEnable=true。并拷贝distribution/conf/plain_acl.yml文件到${ROCKETMQ_HOME}/conf目录。

broker.conf的配置文件如下：

brokerClusterName = DefaultCluster
brokerName = broker-b
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
listenPort=10915
storePathRootDir=E:/SH2019/tmp/rocketmq_home/rocketmq4.5MB/store
storePathCommitLog=E:/SH2019/tmp/rocketmq_home/rocketmq4.5MB/store/commitlog
namesrvAddr=127.0.0.1:9876
autoCreateTopicEnable=false
aclEnable=true

plain_acl.yml文件内容如下：

globalWhiteRemoteAddresses:accounts:
- accessKey: RocketMQsecretKey: 12345678whiteRemoteAddress:admin: falsedefaultTopicPerm: DENYdefaultGroupPerm: SUBtopicPerms:- TopicTest=PUBgroupPerms:# the group should convert to retry topic- oms_consumer_group=DENY- accessKey: adminsecretKey: 12345678whiteRemoteAddress:# if it is admin, it could access all resourcesadmin: true

从上面的配置可知，用户RocketMQ只能发送TopicTest的消息，其他topic无权限发送；拒绝oms_consumer_group消费组的消息消费，其他消费组默认可消费。

4.2 消息发送端示例

public class AclProducer {public static void main(String[] args) throws MQClientException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name", getAclRPCHook());producer.setNamesrvAddr("127.0.0.1:9876");producer.start();for (int i = 0; i < 1; i++) {try {Message msg = new Message("TopicTest3" ,"TagA" , ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);} catch (Exception e) {e.printStackTrace();Thread.sleep(1000);}}producer.shutdown();}static RPCHook getAclRPCHook() {return new AclClientRPCHook(new SessionCredentials("rocketmq","12345678"));}
}

运行效果如图所示：
在这里插入图片描述

4.3 消息消费端示例

public class AclConsumer {public static void main(String[] args) throws InterruptedException, MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4", getAclRPCHook(),new AllocateMessageQueueAveragely());consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);consumer.subscribe("TopicTest", "*");consumer.setNamesrvAddr("127.0.0.1:9876");consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");}static RPCHook getAclRPCHook() {return new AclClientRPCHook(new SessionCredentials("rocketmq","12345678"));}
}

发现并不没有消费消息，符合预期。

参考Demo：消费者

@Overridepublic void run(String... args) throws Exception {System.out.println("通过实现CommandLineRunner接口，在spring boot项目启动后打印参数");// AccessKey 身份验证AclClientRPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));consumer = new DefaultMQPushConsumer(consumerGroupId, rpcHook,new AllocateMessageQueueAveragely());//consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);//设置 TCP 接入域名consumer.setNamesrvAddr(nameserverMQ);//消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);consumer.setVipChannelEnabled(false);consumer.setConsumeTimeout(3);consumer.setMessageModel(MessageModel.BROADCASTING);//广播订阅方式//订阅主题和 标签（ * 代表所有标签)下信息String later = "_" + rocketmqAreacode + "_" + hikvisionId;consumer.subscribe(MQIOTDEV001 + later,  MqServerTypeConstant.YCCK);// //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {// msgs中只收集同一个topic，同一个tag，并且key相同的message// 会把不同的消息分别放置到不同的队列中try {for (Message msg : msgs) {//消费者获取消息 这里只输出 不做后面逻辑处理String body = new String(msg.getBody(), "utf-8");log.info("Consumer-获取消息-主题topic为={}, 消费消息为={}", msg.getTopic(), body);} catch (Exception e) {log.error("0x80517800消费远程操控接口失败"+e.getMessage());}}} catch (UnsupportedEncodingException e) {e.printStackTrace();return ConsumeConcurrentlyStatus.RECONSUME_LATER;}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();System.out.println("消费者 启动成功=======");}

消费者：

public void sendControl(BaseResponse baseResponse) {try {String later = "_" + rocketmqAreacode + "_" + hikvisionId;
//        		DefaultMQPushConsumer dconsumer = new DefaultMQPushConsumer("my_test_group");AclClientRPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(remoteProperties.getAccessKey(),remoteProperties.getSecretKey()));DefaultMQProducer producer = new DefaultMQProducer(producerGroupIOTDEV001, rpcHook);producer.setNamesrvAddr(nameserverMQ);producer.setInstanceName(RunTimeUtil.getRocketMqUniqeInstanceName());producer.start();Message msg = new Message(MQIOTDEV001 + later, MqServerTypeConstant.YCCK,// Message Tag,可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤MqServerTypeConstant.YCCK,// Message Body// 任何二进制形式的数据， MQ不做任何干预，// 需要Producer与Consumer协商好一致的序列化和反序列化方式JsonUtils.object2Json(baseResponse).getBytes("UTF-8"));//同步发送消息producer.send(msg);// 在callback返回之前即可取得msgId。// 在应用退出前，销毁Producer对象。注意：如果不销毁也没有问题log.info("推送反控响应信息到rocketmq:同步发送消息");producer.shutdown();}catch (Exception e) {log.error("0x80517800推送响应信息到rocketmq失败："+e.getMessage());e.printStackTrace();}}

 

关于RocketMQ ACL的使用就介绍到这里了，下一篇将介绍RocketMQ ACL实现原理。

 

这篇关于RocketMQ中ACL权限控制的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1134827。

相关文章

用js控制视频播放进度基本示例代码

用js控制视频播放进度基本示例代码

《用js控制视频播放进度基本示例代码》写前端的时候,很多的时候是需要支持要网页视频播放的功能,下面这篇文章主要给大家介绍了关于用js控制视频播放进度的相关资料,文中通过代码介绍的非常详细,需要的朋友可... 目录前言html部分:JavaScript部分:注意：总结前言在javascript中控制视频播放
阅读更多...
Python异步编程中asyncio.gather的并发控制详解

Python异步编程中asyncio.gather的并发控制详解

《Python异步编程中asyncio.gather的并发控制详解》在Python异步编程生态中,asyncio.gather是并发任务调度的核心工具,本文将通过实际场景和代码示例,展示如何结合信号量... 目录一、asyncio.gather的原始行为解析二、信号量控制法：给并发装上"节流阀"三、进阶控制
阅读更多...
使用DrissionPage控制360浏览器的完美解决方案

使用DrissionPage控制360浏览器的完美解决方案

《使用DrissionPage控制360浏览器的完美解决方案》在网页自动化领域,经常遇到需要保持登录状态、保留Cookie等场景,今天要分享的方案可以完美解决这个问题：使用DrissionPage直接... 目录完整代码引言为什么要使用已有用户数据？核心代码实现1. 导入必要模块2. 关键配置（重点！）3.
阅读更多...
SpringSecurity 认证、注销、权限控制功能(注销、记住密码、自定义登入页)

SpringSecurity 认证、注销、权限控制功能(注销、记住密码、自定义登入页)

《SpringSecurity认证、注销、权限控制功能(注销、记住密码、自定义登入页)》SpringSecurity是一个强大的Java框架,用于保护应用程序的安全性,它提供了一套全面的安全解决方案... 目录简介认识Spring Security“认证”（Authentication）“授权” （Auth
阅读更多...
python之流程控制语句match-case详解

python之流程控制语句match-case详解

《python之流程控制语句match-case详解》：本文主要介绍python之流程控制语句match-case使用,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录match-case 语法详解与实战一、基础值匹配（类似 switch-case）二、数据结构解构匹
阅读更多...
Spring Security注解方式权限控制过程

Spring Security注解方式权限控制过程

《SpringSecurity注解方式权限控制过程》：本文主要介绍SpringSecurity注解方式权限控制过程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、摘要二、实现步骤2.1 在配置类中添加权限注解的支持2.2 创建Controller类2.3 Us
阅读更多...
Redis实现RBAC权限管理

Redis实现RBAC权限管理

《Redis实现RBAC权限管理》本文主要介绍了Redis实现RBAC权限管理,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1. 什么是 RBAC？2. 为什么使用 Redis 实现 RBAC？3. 设计 RBAC 数据结构
阅读更多...
Python中如何控制小数点精度与对齐方式

Python中如何控制小数点精度与对齐方式

《Python中如何控制小数点精度与对齐方式》在Python编程中,数据输出格式化是一个常见的需求,尤其是在涉及到小数点精度和对齐方式时,下面小编就来为大家介绍一下如何在Python中实现这些功能吧... 目录一、控制小数点精度1. 使用 round() 函数2. 使用字符串格式化二、控制对齐方式1. 使用
阅读更多...
Springboot控制反转与Bean对象的方法

Springboot控制反转与Bean对象的方法

《Springboot控制反转与Bean对象的方法》文章介绍了SpringBoot中的控制反转（IoC）概念,描述了IoC容器如何管理Bean的生命周期和依赖关系,它详细讲解了Bean的注册过程,包括... 目录1 控制反转1.1 什么是控制反转1.2 SpringBoot中的控制反转2 Ioc容器对Bea
阅读更多...
Vue中动态权限到按钮的完整实现方案详解

Vue中动态权限到按钮的完整实现方案详解

《Vue中动态权限到按钮的完整实现方案详解》这篇文章主要为大家详细介绍了Vue如何在现有方案的基础上加入对路由的增、删、改、查权限控制,感兴趣的小伙伴可以跟随小编一起学习一下... 目录一、数据库设计扩展1.1 修改路由表（routes）1.2 修改角色与路由权限表（role_routes）二、后端接口设计
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2