一、ACL介绍

RocketMQ在4.4.0版本开始支持ACL。ACL是access control list的简称，俗称访问控制列表。访问控制，基本上会涉及到用户、资源、权限、角色等概念；

1> 用户

• 用户是访问控制的基础要素，包括：用户名、秘钥；用来区分一个请求可以使用哪些权限；

2> 资源

• RocketMQ中需要被保护的对象，主要包括：Topic、ConsumerGroup。

3> 权限

• 针对资源可以做的操作，包括：DENY（拒绝）、PUB（发送）、SUB（订阅）

4> 角色

• RocketMQ仅有两种角色：管理员和非管理员。

此外，RocketMQ中还支持客户端IP的白名单机制。

ACL原理

像这种做权限的，基本就是

• 在某一个地方存储用户信息，包括：用户名、秘钥、用户角色、用户拥有的权限。其实很像Spring-Security。只是RocketMQ的ACL比较轻量级。
• 每个请求访问Broker时，都要在执行请求之前做一个拦截，判断用户是否有权限操作，没权限就直接报错返回。这里的拦截其实就可以理解为SpringMVC的Interceptor、Dubbo的Filter。

思路嘎嘎简单，具体原理、源码，下一篇聊。

二、ACL使用

1、Broker端配置

想要使用ACL权限，首先必须要在Broker端加上相应的配置（broker.conf中增加aclEnable=true）开启ACL；

aclEnable=true

然后设置具体的鉴权方式；参考RocketMQ源码中distribution/conf/plain_acl.yml文件（https://github.com/apache/rocketmq/blob/4.9.x/distribution/conf/plain_acl.yml），将其复制到${ROCKETMQ_HOME}/conf目录下。

globalWhiteRemoteAddresses:
- 10.10.103.*
- 192.168.0.*

accounts:
- accessKey: mockmock
  secretKey: 123456
  whiteRemoteAddress:
  admin: false
  defaultTopicPerm: DENY
  defaultGroupPerm: SUB
  topicPerms:
  - topicA=DENY
  - topicB=PUB|SUB
  - topicC=SUB
  groupPerms:
  # the group should convert to retry topic
  - groupA=DENY
  - groupB=PUB|SUB
  - groupC=SUB

- accessKey: mockmock2
  secretKey: 123456
  whiteRemoteAddress: 192.168.1.*
  admin: true

这里的配置表示：

• 10.10.103.*、192.168.0.*网段下的Client可以直接通过ACL权限认证；
• mockmock用户角色为非管理员，针对主题topicB、消费组groupB可以做PUB、SUB；针对主题topicC、消费组groupC只能做SUB；而针对主题topicA、消费组groupA啥操作也做不了。
• mockmock2用户为管理员，用户有到任意资源的访问权限。

ACL配置详解

1> globalWhiteRemoteAddresses

• 全局IP白名单，类型为数组，支持配置多个IP、多种类型IP地址；

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

2> accounts

• 配置用户信息，类型为数组类型。包含多个子元素，比如：accessKey、secretKey、whiteRemote
  Address、admin、defaultTopicPerm、defaultGroupPerm、topicPerms、groupPe
  rms。
  • accessKey --> 登录用户名，长度必须大于 6 个字符。
  • secretKey --> 登录密码。长度必须大于 6 个字符。
  • whiteRemoteAddress --> 用户级别的 IP 地址白名单。类型为字符串，配置规则与 globalWhiteRemot
    eAddresses一样，但只能配置一条规则。
  • admin --> boolean 类型，设置是否是 admin。
    下列权限只有 admin=true 时才有权限执行:
    • UPDATE_AND_CREATE_TOPIC --> 更新或创建主题
    • UPDATE_BROKER_CONFIG --> 更新 Broker 配置
    • DELETE_TOPIC_IN_BROKER --> 删除主题
    • UPDATE_AND_CREATE_SUBSCRIPTIONGROUP --> 更新或创建订阅组信息
    • DELETE_SUBSCRIPTIONGROUP --> 删除订阅组信息。
  • defaultTopicPerm --> 默认 topic 权限。该值默认为 DENY(拒绝)。
  • defaultGroupPerm --> 默认消费组权限，该值默认为 DENY(拒绝)，建议值为 SUB。
  • topicPerms --> 设置 topic 的权限，类型为数组，可选值：DENY、PUB、SUB
  • groupPerms --> 设置消费组的权限，类型为数组，可选值：DENY、PUB、SUB

2、Producer

创建DefaultMQProducer时，指定RPCHook；

public class AclProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("saint-test", getAclRPCHook());
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.setMaxMessageSize(1024 * 1024 * 10);
        producer.start();

        // topic 和body
        Message msg = new Message("test-topic", "study002".getBytes(StandardCharsets.UTF_8));
        SendResult send = producer.send(msg);

        // 关闭生产者
        producer.shutdown();
    }

    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("username", "password"));
    }
}

3、Consumer

同样在创建DefaultMQPushConsumer时，指定RPCHook；

public class AclConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("study-consumer", getAclRPCHook(), new AllocateMessageQueueAveragely());
        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.subscribe("test-topic", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer start。。。。。。");
    }

    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("username", "password"));
    }
}

4、RocketMQ-Console

在application.properties文件中设置用户名和密码（建议设置管理员用户）：

rocketmq.config.accessKey=administrators
rocketmq.config.secretKey=administrators