189 8069 5689
本篇内容介绍了“RocketMQ ACL实现机制是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
创新互联从2013年成立，公司以成都网站建设、成都网站制作、系统开发、网络推广、文化传媒、企业宣传、平面广告设计等为主要业务，适用行业近百种。服务企业客户超过千家，涉及国内多个省份客户。拥有多年网站建设开发经验。为企业提供专业的网站建设、创意设计、宣传推广等服务。 通过专业的设计、独特的风格，为不同客户提供各种风格的特色服务。
根据RocketMQ ACL使用手册，我们应该首先看一下Broker服务器在开启ACL机制时如何加载配置文件，并如何工作的。
Broker端ACL的入口代码为：BrokerController#initialAcl
private void initialAcl() {
if (!this.brokerConfig.isAclEnable()) { // [@1](https://my.oschina.net/u/1198)
log.info("The broker dose not enable acl");
return;
}
List accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class); // @2
if (accessValidators == null || accessValidators.isEmpty()) {
log.info("The broker dose not load the AccessValidator");
return;
}
for (AccessValidator accessValidator: accessValidators) { // [@3](https://my.oschina.net/u/2648711)
final AccessValidator validator = accessValidator;
this.registerServerRPCHook(new RPCHook() {
[@Override](https://my.oschina.net/u/1162528)
public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
//Do not catch the exception
validator.validate(validator.parse(request, remoteAddr)); // @4
}
@Override
public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
}
});
}
} 本方法的实现共4个关键点。 代码@1：首先判断Broker是否开启了acl，通过配置参数aclEnable指定，默认为false。
代码@2：使用类似SPI机制，加载配置的AccessValidator,该方法返回一个列表，其实现逻辑时读取META-INF/service/org.apache.rocketmq.acl.AccessValidator文件中配置的访问验证器，默认配置内容如下： 
代码@3：遍历配置的访问验证器(AccessValidator),并向Broker处理服务器注册钩子函数，RPCHook的doBeforeRequest方法会在服务端接收到请求，将其请求解码后，执行处理请求之前被调用;RPCHook的doAfterResponse方法会在处理完请求后，将结果返回之前被调用，其调用如图所示：
代码@4：在RPCHook#doBeforeRequest方法中调用AccessValidator#validate, 在真实处理命令之前，先执行ACL的验证逻辑，如果拥有该操作的执行权限，则放行，否则抛出AclException。
接下来，我们将重点放到Broker默认实现的访问验证器：PlainAccessValidator。
AccessValidator 访问验证器接口，主要定义两个接口。 1）AccessResource parse(RemotingCommand request, String remoteAddr) 从请求头中解析本次请求对应的访问资源，即本次请求需要的访问权限。 2）void validate(AccessResource accessResource) 根据本次需要访问的权限，与请求用户拥有的权限进行对比验证，判断是拥有权限，如果没有访问该操作的权限，则抛出异常，否则放行。
PlainAccessValidator RocketMQ默认提供的基于yml配置格式的访问验证器。
接下来我们重点看一下PlainAccessValidator的parse方法与validate方法的实现细节。在讲解该方法之前，我们首先认识一下RocketMQ封装访问资源的PlainAccessResource。
我们对其属性一一做个介绍：
private String accessKey 访问Key，用户名。
private String secretKey 用户密码。
private String whiteRemoteAddress 远程IP地址白名单。
private boolean admin 是否是管理员角色。
private byte defaultTopicPerm = 1 默认topic访问权限，即如果没有配置topic的权限，则Topic默认的访问权限为1，表示为DENY。
private byte defaultGroupPerm = 1 默认的消费组访问权限，默认为DENY。
private Map
private RemoteAddressStrategy remoteAddressStrategy 远程IP地址验证策略。
private int requestCode 当前请求的requestCode。
private byte[] content 请求头与请求体的内容。
private String signature 签名字符串，这是通常的套路，在客户端时，首先将请求参数排序，然后使用secretKey生成签名字符串，服务端重复这个步骤，然后对比签名字符串，如果相同，则认为登录成功，否则失败。
private String secretToken 密钥token。
private String recognition 目前作用未知，代码中目前未被使用。
public PlainAccessValidator() {
aclPlugEngine = new PlainPermissionLoader();
}构造函数，直接创建PlainPermissionLoader对象，从命名上来看，应该是触发acl规则的加载，即解析plain_acl.yml，接下来会重点探讨，即acl启动流程之配置文件的解析。
该方法的作用就是从请求命令中解析出本次访问所需要的访问权限，最终构建AccessResource对象，为后续的校验权限做准备。
PlainAccessResource accessResource = new PlainAccessResource();
if (remoteAddr != null && remoteAddr.contains(":")) {
accessResource.setWhiteRemoteAddress(remoteAddr.split(":")[0]);
} else {
accessResource.setWhiteRemoteAddress(remoteAddr);
}Step1：首先创建PlainAccessResource，从远程地址中提取出远程访问IP地址。
if (request.getExtFields() == null) {
throw new AclException("request's extFields value is null");
}
accessResource.setRequestCode(request.getCode());
accessResource.setAccessKey(request.getExtFields().get(SessionCredentials.ACCESS_KEY));
accessResource.setSignature(request.getExtFields().get(SessionCredentials.SIGNATURE));
accessResource.setSecretToken(request.getExtFields().get(SessionCredentials.SECURITY_TOKEN));Step2：如果请求头中的扩展字段为空，则抛出异常，如果不为空，则从请求头中读取requestCode、accessKey(请求用户名)、签名字符串(signature)、secretToken。
try {
switch (request.getCode()) {
case RequestCode.SEND_MESSAGE:
accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.PUB);
break;
case RequestCode.SEND_MESSAGE_V2:
accessResource.addResourceAndPerm(request.getExtFields().get("b"), Permission.PUB);
break;
case RequestCode.CONSUMER_SEND_MSG_BACK:
accessResource.addResourceAndPerm(request.getExtFields().get("originTopic"), Permission.PUB);
accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("group")), Permission.SUB);
break;
case RequestCode.PULL_MESSAGE:
accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("consumerGroup")), Permission.SUB);
break;
case RequestCode.QUERY_MESSAGE:
accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
break;
case RequestCode.HEART_BEAT:
HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
accessResource.addResourceAndPerm(getRetryTopic(data.getGroupName()), Permission.SUB);
for (SubscriptionData subscriptionData : data.getSubscriptionDataSet()) {
accessResource.addResourceAndPerm(subscriptionData.getTopic(), Permission.SUB);
}
}
break;
case RequestCode.UNREGISTER_CLIENT:
final UnregisterClientRequestHeader unregisterClientRequestHeader =
(UnregisterClientRequestHeader) request
.decodeCommandCustomHeader(UnregisterClientRequestHeader.class);
accessResource.addResourceAndPerm(getRetryTopic(unregisterClientRequestHeader.getConsumerGroup()), Permission.SUB);
break;
case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
final GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader =
(GetConsumerListByGroupRequestHeader) request
.decodeCommandCustomHeader(GetConsumerListByGroupRequestHeader.class);
accessResource.addResourceAndPerm(getRetryTopic(getConsumerListByGroupRequestHeader.getConsumerGroup()), Permission.SUB);
break;
case RequestCode.UPDATE_CONSUMER_OFFSET:
final UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader =
(UpdateConsumerOffsetRequestHeader) request
.decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);
accessResource.addResourceAndPerm(getRetryTopic(updateConsumerOffsetRequestHeader.getConsumerGroup()), Permission.SUB);
accessResource.addResourceAndPerm(updateConsumerOffsetRequestHeader.getTopic(), Permission.SUB);
break;
default:
break;
}
} catch (Throwable t) {
throw new AclException(t.getMessage(), t);
}Step3：根据请求命令，设置本次请求需要拥有的权限，上述代码比较简单，就是从请求中得出本次操作的Topic、消息组名称，为了方便区分topic与消费组，消费组使用消费者对应的重试主题，当成资源的Key，从这里也可以看出，当前版本需要进行ACL权限验证的请求命令如下：
SEND_MESSAGE
SEND_MESSAGE_V2
CONSUMER_SEND_MSG_BACK
PULL_MESSAGE
QUERY_MESSAGE
HEART_BEAT
UNREGISTER_CLIENT
GET_CONSUMER_LIST_BY_GROUP
UPDATE_CONSUMER_OFFSET
