目录java
@(本节目录)数组
ACL是access control list的简称,俗称访问控制列表。访问控制,基本上会涉及到用户、资源、权限、角色等概念,那在RocketMQ中上述会对应哪些对象呢?并发
另外,RocketMQ还支持按照客户端IP进行白名单设置。ide
在讲解如何使用ACL以前,咱们先简单看一下RocketMQ ACL的请求流程:
对于上述具体的实现,将在后续文章中重点讲解,本文的目的只是但愿给读者一个大概的了解。源码分析
acl默认的配置文件名:plain_acl.yml,须要放在${ROCKETMQ_HOME}/store/config目录下。下面对其配置项一一介绍。.net
全局白名单,其类型为数组,即支持多个配置。其支持的配置格式以下:3d
配置用户信息,该类型为数组类型。拥有accessKey、secretKey、whiteRemoteAddress、admin、defaultTopicPerm、defaultGroupPerm、topicPerms、groupPerms子元素。rest
登陆用户名,长度必须大于6个字符。code
登陆密码。长度必须大于6个字符。中间件
用户级别的IP地址白名单。其类型为一个字符串,其配置规则与globalWhiteRemoteAddresses,但只能配置一条规则。
boolean类型,设置是不是admin。以下权限只有admin=true时才有权限执行。
默认topic权限。该值默认为DENY(拒绝)。
默认消费组权限,该值默认为DENY(拒绝),建议值为SUB。
设置topic的权限。其类型为数组,其可选择值在下节介绍。
设置消费组的权限。其类型为数组,其可选择值在下节介绍。能够为每一消费组配置不同的权限。
上面定义了全局白名单、用户级别的白名单,用户级别的权限,为了更好的配置ACL权限规则,下面给出权限匹配逻辑。
首先,须要在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: RocketMQ secretKey: 12345678 whiteRemoteAddress: admin: false defaultTopicPerm: DENY defaultGroupPerm: SUB topicPerms: - TopicTest=PUB groupPerms: # the group should convert to retry topic - oms_consumer_group=DENY - accessKey: admin secretKey: 12345678 whiteRemoteAddress: # if it is admin, it could access all resources admin: true
从上面的配置可知,用户RocketMQ只能发送TopicTest的消息,其余topic无权限发送;拒绝oms_consumer_group消费组的消息消费,其余消费组默承认消费。
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")); } }
运行效果如图所示:
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() { @Override public 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")); } }
发现并不没有消费消息,符合预期。
关于RocketMQ ACL的使用就介绍到这里了,下一篇将介绍RocketMQ ACL实现原理。
推荐阅读:
一、RocketMQ实战:生产环境中,autoCreateTopicEnable为何不能设置为true
二、RocketMQ 消息发送system busy、broker busy缘由分析与解决方案
做者介绍:
丁威,《RocketMQ技术内幕》做者,RocketMQ 社区布道师,公众号:中间件兴趣圈 维护者,目前已陆续发表源码分析Java集合、Java 并发包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源码专栏。