增加ACL权限

1.在broker-*.properties文件中增加aclEnable=true的配置

修改权限控制文件 /home/rocketmq-4.5.2/conf/plain_acl.yml如下图所示

#全局白名单，其类型为数组，即支持多个配置。其支持的配置格式如下：
#空：表示不设置白名单，该条规则默认返回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段使用 "" 或"-"表示范围。

globalWhiteRemoteAddresses:
- 10.10.103.*
- 192.168.0.*

accounts:
- accessKey: RocketMQ
  secretKey: 12345678
  #用户级别的IP地址白名单。其类型为一个字符串，其配置规则与globalWhiteRemoteAddresses，但只能配置一条规则。
  whiteRemoteAddress: 192.168.1.*
  #如下权限只有admin=true时才有权限执行。UPDATE_AND_CREATE_TOPIC,UPDATE_BROKER_CONFIG,DELETE_TOPIC_IN_BROKER,UPDATE_AND_CREATE_SUBSCRIPTIONGROUP,DELETE_SUBSCRIPTIONGROUP
  admin: false
  #默认topic权限。 该值默认为DENY(拒绝)。
  defaultTopicPerm: DENY
  #默认消费组权限，该值默认为DENY(拒绝)，建议值为SUB。
  defaultGroupPerm: SUB
  #设置topic的权限。其类型为数组，其可选择值在下节介绍。
  topicPerms:
  - topicA=DENY
  - topicB=PUB|SUB
  - topicC=SUB
  - TopicTest=PUB
  #设置消费组的权限。其类型为数组，其可选择值在下节介绍。可以为每一消费组配置不一样的权限。
  groupPerms:
  #the group should convert to retry topic
  - groupA=DENY
  - groupB=PUB|SUB
  - groupC=SUB
  - acl_test_consumer_group_DENY=DENY
  
- accessKey: rocketmq2
  secretKey: 12345678
  whiteRemoteAddress: 192.168.1.*
  # if it is admin, it could access all resources
  admin: true

Maven 依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-acl</artifactId>
    <version>4.5.2</version>
</dependency>

生产者代码

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;

public class AclProducer {
    public static void main(String[] args)
        throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("rexel_notice_p1", getAclRPCHook());
        producer.setNamesrvAddr("192.168.29.100:9876;192.168.29.101:9876");
        producer.start();
        Message msg = new Message("rexel_notice" ,"*" , ("Hello RocketMQ ").getBytes());
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
        producer.shutdown();
    }

    static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("rexel_developer","19@ljWo2iUow"));
    }
}

消费者代码

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

public class AclConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
            "rexel_notice_g1", getAclRPCHook(), new AllocateMessageQueueAveragely());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("rexel_notice", "*");
        consumer.setNamesrvAddr("192.168.29.100:9876;192.168.29.101: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("rexel_developer","19@ljWo2iUow"));
    }
}

总结

ACL的执行判断逻辑可以简单的总结下面所示

################总结###############################
#### -> 符合帐号                            
####           -> 符合白名单   -> 检查规则
####           -> 不符合白名单 -> AclException
#### -> 不符合帐号 -> AclException

本文转载自：
配置RocketMQ ACL权限