配置RocketMQ ACL權限

==環境==

系統：Linux Centos7.2

RocketMQ版本：4.6.1

 

==集群形態==

 

==修改前配置文件==

broker-a.properties

brokerClusterName=rexel
brokerName=broker-a
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10921
brokerIP1=192.168.29.100
namesrvAddr=192.168.29.100:9876;192.168.29.101:9876
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/data/rocketmq/rootdir-a-m
storePathCommitLog=/home/data/rocketmq/commitlog-a-m
storePathConsumerQueue=/home/data/rocketmq/consumequeue-a-m
storePathIndex=/home/data/rocketmq/index-a-m
storeCheckpoint=/home/data/rocketmq/checkpoint-a-m

broker-a-s.properties

brokerClusterName=rexel
brokerName=broker-a
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10931
brokerIP1=192.168.29.101
namesrvAddr=192.168.29.100:9876;192.168.29.101:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/data/rocketmq/rootdir-a-s
storePathCommitLog=/home/data/rocketmq/commitlog-a-s
storePathConsumerQueue=/home/data/rocketmq/consumequeue-a-s
storePathIndex=/home/data/rocketmq/index-a-s
storeCheckpoint=/home/data/rocketmq/checkpoint-a-s

broker-b.properties

brokerClusterName=rexel
brokerName=broker-b
brokerId=0
deleteWhen=04
fileReservedTime=48
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
listenPort=10921
brokerIP1=192.168.29.101
namesrvAddr=192.168.29.100:9876;192.168.29.101:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/data/rocketmq/rootdir-b-m
storePathCommitLog=/home/data/rocketmq/commitlog-b-m
storePathConsumerQueue=/home/data/rocketmq/consumequeue-b-m
storePathIndex=/home/data/rocketmq/index-b-m
storeCheckpoint=/home/data/rocketmq/checkpoint-b-m

broker-b-s.properties

brokerClusterName=rexel
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
listenPort=10931
brokerIP1=192.168.29.102
namesrvAddr=192.168.29.100:9876;192.168.29.101:9876
defaultTopicQueueNums=4
autoCreateTopicEnable=false
autoCreateSubscriptionGroup=true
storePathRootDir=/home/data/rocketmq/rootdir-b-s
storePathCommitLog=/home/data/rocketmq/commitlog-b-s
storePathConsumerQueue=/home/data/rocketmq/consumequeue-b-s
storePathIndex=/home/data/rocketmq/index-b-s
storeCheckpoint=/home/data/rocketmq/checkpoint-b-s

 

==增加ACL權限==

1. 在Broker文件中增加aclEnable=true的配置

如下圖

 

2.重新啟動集群

[1][2]nohup sh /home/rocketmq-4.6.1/bin/mqnamesrv >/home/rocketmq-4.6.1/logs/mqnamesrv.log &
[1]nohup sh /home/rocketmq-4.6.1/bin/mqbroker -c /home/rocketmq-4.6.1/conf/rexel/broker-a.properties >/home/rocketmq-4.6.1/logs/broker-a-m.log 2>&1 &
[2]nohup sh /home/rocketmq-4.6.1/bin/mqbroker -c /home/rocketmq-4.6.1/conf/rexel/broker-a-s.properties >/home/rocketmq-4.6.1/logs/broker-a-s.log 2>&1 &
[2]nohup sh /home/rocketmq-4.6.1/bin/mqbroker -c /home/rocketmq-4.6.1/conf/rexel/broker-b.properties >/home/rocketmq-4.6.1/logs/broker-b-m.log 2>&1 &
[3]nohup sh /home/rocketmq-4.6.1/bin/mqbroker -c /home/rocketmq-4.6.1/conf/rexel/broker-b-s.properties >/home/rocketmq-4.6.1/logs/broker-b-s.log 2>&1 &

 

這個時候通過集群查看命令已經會提示acl權限錯誤了，證明權限已經生效。

命令：sh /home/rocketmq-4.6.1/bin/mqadmin clusterList -n "192.168.29.100:9876;192.168.29.101:9876"

錯誤日志：

org.apache.rocketmq.client.exception.MQBrokerException: CODE: 1  DESC: org.apache.rocketmq.acl.common.AclException: Check signature failed for accessKey=rocketmq2, org.apache.rocketmq.acl.plain.PlainPermissionManager.validate(PlainPermissionManager.java:410)
For more information, please visit the url, http://rocketmq.apache.org/docs/faq/
    at org.apache.rocketmq.client.impl.MQClientAPIImpl.getBrokerRuntimeInfo(MQClientAPIImpl.java:1288)
    at org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl.fetchBrokerRuntimeStats(DefaultMQAdminExtImpl.java:266)
    at org.apache.rocketmq.tools.admin.DefaultMQAdminExt.fetchBrokerRuntimeStats(DefaultMQAdminExt.java:231)
    at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.printClusterBaseInfo(ClusterListSubCommand.java:212)
    at org.apache.rocketmq.tools.command.cluster.ClusterListSubCommand.execute(ClusterListSubCommand.java:88)
    at org.apache.rocketmq.tools.command.MQAdminStartup.main0(MQAdminStartup.java:139)
    at org.apache.rocketmq.tools.command.MQAdminStartup.main(MQAdminStartup.java:90)

 

3.配置權限

文件路徑：/home/rocketmq-4.6.1/conf/plain_acl.yml

網上給的例子：

globalWhiteRemoteAddresses:
- 10.10.15.*
- 192.168.0.*
 
accounts:
- accessKey: RocketMQ
  secretKey: 12345678
  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: rocketmq2
  secretKey: 12345678
  whiteRemoteAddress: 192.168.1.*
  # if it is admin, it could access all resources
  admin: true

 

plain_acl.yml文件中相關的參數含義及使用

字段	取值	含義
globalWhiteRemoteAddresses	*;192.168.*.*;192.168.0.1	全局IP白名單
accessKey	字符串	Access Key 用戶名
secretKey	字符串	Secret Key 密碼
whiteRemoteAddress	*;192.168.*.*;192.168.0.1	用戶IP白名單
admin	true;false	是否管理員賬戶
defaultTopicPerm	DENY;PUB;SUB;PUB|SUB	默認的Topic權限
defaultGroupPerm	DENY;PUB;SUB;PUB|SUB	默認的ConsumerGroup權限
topicPerms	topic=權限	各個Topic的權限
groupPerms	group=權限	各個ConsumerGroup的權限

權限標識符的含義

權限	含義
DENY	拒絕
ANY	PUB 或者 SUB 權限
PUB	發送權限
SUB	訂閱權限

 

我的配置。配置完成之后不需要重啟。

globalWhiteRemoteAddresses:
- 192.168.29.100
- 192.168.29.101
- 192.168.29.102

accounts:
- accessKey: rexel_developer
  secretKey: 19@ljWo2iUow
  whiteRemoteAddress:
  admin: false
  defaultTopicPerm: DENY
  defaultGroupPerm: SUB
  topicPerms:
  - rexel_notice=PUB|SUB
  groupPerms:
  - rexel_notice_g1=SUB
  - rexel_notice_p1=PUB

- accessKey: rexel_admin
  secretKey: 98&UIwowu@9o
  whiteRemoteAddress:
  admin: true

 

==編寫驗證程序==

增加acl的maven依賴：

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

 

生產者代碼：

package acl;

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"));
    }
}

 

消費者代碼：

package acl;

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"));
    }
}

 

最后可以看到消費者可以正常消費的日志：

 

==補充==

增加了權限之后，貌似就沒有辦法通過控制台命令上創建topic了。

我目前是通過rocketmq-console來進行Topic及ConsumerGroup管理的。