分析 Dubbo SPI 源码，扩展 Dubbo Validation (groups)

Java Dubbo
目前架构是网关直接通过泛化调用Dubbo服务，不同于web Controller使用SpringMVC模块来做到参数注解校验。不过不用担心Dubbo也考虑到了这一点，基于SPI机制提供了ValidationFilter。来看看他是如何实现的。

Dubbo 源码实现

Dubbo SPI定义

什么是Dubbo SPI？简单说是通过文件配置对应class路径后会被执行class里的invoke函数。其中的实现原理大家顺着Dubbo的ExtensionLoader去看下源码就能知道。

ValidationFilter说明

//在哪种服务类型激活
//这里的VALIDATION_KEY=“validation” 也就是我们在SPI中需要把key按这个规定定义
@Activate(group = {CONSUMER, PROVIDER}, value = VALIDATION_KEY, order = 10000)
public class ValidationFilter implements Filter {
    private Validation validation;
    public void setValidation(Validation validation) {
        this.validation = validation;
    }
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        //如果SPI中定义了validation 那么就进行校验
        if (validation != null && !invocation.getMethodName().startsWith("$")
            && ConfigUtils.isNotEmpty(invoker.getUrl().getMethodParameter(invocation.getMethodName(), VALIDATION_KEY))) {
            try {
                //执行参数校验
                Validator validator = validation.getValidator(invoker.getUrl());
                if (validator != null) {
                    validator.validate(invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());
                }
            } catch (RpcException e) {
                throw e;
            } catch (ValidationException e) {
                //抛出异常 这里的ValidationException需要深挖一下，后面会说
                // only use exception's message to avoid potential serialization issue
                return AsyncRpcResult.newDefaultAsyncResult(new ValidationException(e.getMessage()), invocation);
            } catch (Throwable t) {
                return AsyncRpcResult.newDefaultAsyncResult(t, invocation);
            }
        }
        return invoker.invoke(invocation);
    }
}

基础使用

Maven依赖

SpringBoot项目推荐使用

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
</dependency>

手动依赖

<dependency>
  <groupId>javax.el</groupId>
  <artifactId>javax.el-api</artifactId>
  <version>3.0.0</version>
</dependency>
<dependency>
  <groupId>org.glassfish</groupId>
  <artifactId>javax.el</artifactId>
</dependency>
<dependency>
  <groupId>javax.validation</groupId>
  <artifactId>validation-api</artifactId>
</dependency>
<dependency>
  <groupId>org.hibernate</groupId>
  <artifactId>hibernate-validator</artifactId>
</dependency>

DTO添加validation定义

import lombok.Data;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.io.Serializable;
@Data
public class PracticeParam implements Serializable {
    @NotNull(message = "periodId不能为空")
    private Long periodId;
}

服务者interface

public interface IPracticeService {
    boolean practiceAdd(PracticeParam practiceParam);
}

Dubbo RPC单元测试

@SpringBootTest(classes = ClientApplication.class)
@RunWith(SpringRunner.class)
@Slf4j
public class PrecticeTest {
    @DubboReference(group = "user")
    private IPracticeService practiceLogicService;
    @Test
    public void add(){
        PracticeParam practiceParam=new PracticeParam();
        log.info(String.valueOf(practiceLogicService.practiceAdd(practiceParam)));
    }
}

测试结果

javax.validation.ValidationException: Failed to validate service: com.xx.contract.IPracticeService, method: practiceAdd, cause: [ConstraintViolationImpl{interpolatedMessage='periodId不能为空', propertyPath=periodId, rootBeanClass=class com.xx.request.PracticeParam, messageTemplate='periodId不能为空'}]

看结果是生效了，不过这和实际项目中还有些距离。不能把这个异常抛给网关，根据项目需要适配上全局异常试试。

源码分析与扩展

Dubbo异常处理

再回到前面SPI文件

ExceotionFilter源码分析

//在服务提供者端生效
@Activate(group = CommonConstants.PROVIDER)
public class ExceptionFilter implements Filter, Filter.Listener {
    private Logger logger = LoggerFactory.getLogger(ExceptionFilter.class);
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        return invoker.invoke(invocation);
    }
    @Override
    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        //异常处理逻辑
        if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
            try {
                Throwable exception = appResponse.getException();
                // directly throw if it's checked exception
                if (!(exception instanceof RuntimeException) && (exception instanceof Exception)) {
                    return;
                }
                // directly throw if the exception appears in the signature
                try {
                    Method method = invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes());
                    Class<?>[] exceptionClassses = method.getExceptionTypes();
                    for (Class<?> exceptionClass : exceptionClassses) {
                        if (exception.getClass().equals(exceptionClass)) {
                            return;
                        }
                    }
                } catch (NoSuchMethodException e) {
                    return;
                }
                // for the exception not found in method's signature, print ERROR message in server's log.
                logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + exception.getClass().getName() + ": " + exception.getMessage(), exception);
                // directly throw if exception class and interface class are in the same jar file.
                String serviceFile = ReflectUtils.getCodeBase(invoker.getInterface());
                String exceptionFile = ReflectUtils.getCodeBase(exception.getClass());
                if (serviceFile == null || exceptionFile == null || serviceFile.equals(exceptionFile)) {
                    return;
                }
                // directly throw if it's JDK exception
                String className = exception.getClass().getName();
                if (className.startsWith("java.") || className.startsWith("javax.")) {
                    return;
                }
                // directly throw if it's dubbo exception
                if (exception instanceof RpcException) {
                    return;
                }
                // otherwise, wrap with RuntimeException and throw back to the client
                //重点时这句，替换异常信息
                appResponse.setException(new RuntimeException(StringUtils.toString(exception)));
            } catch (Throwable e) {
                logger.warn("Fail to ExceptionFilter when called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
            }
        }
    }
    @Override
    public void onError(Throwable e, Invoker<?> invoker, Invocation invocation) {
        logger.error("Got unchecked and undeclared exception which called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
    }
    // For test purpose
    public void setLogger(Logger logger) {
        this.logger = logger;
    }
}

基础使用

定义SPI

项目结构说明：

文件定义（Dubbo SPI需要严格按如下路径和文件名）

src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter

文件内容

validation=com.xx.xx.config.DubboValidationFilter
exception=com.xx.xx.config.DubboExceptionFilter

自定义DubboValidationFilter

import com.xx.exception.ParamException;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.rpc.*;
import org.apache.dubbo.validation.Validation;
import org.apache.dubbo.validation.Validator;
import javax.validation.ConstraintViolation;
import javax.validation.ConstraintViolationException;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
import static org.apache.dubbo.common.constants.FilterConstants.VALIDATION_KEY;
@Activate(group = {PROVIDER}, value = VALIDATION_KEY, order = -1)
public class DubboValidationFilter implements Filter {
    private Validation validation;
    public void setValidation(Validation validation) {
        this.validation = validation;
    }
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        if (validation != null && !invocation.getMethodName().startsWith("$")
            && ConfigUtils.isNotEmpty(invoker.getUrl().getMethodParameter(invocation.getMethodName(), VALIDATION_KEY))) {
            try {
                Validator validator = validation.getValidator(invoker.getUrl());
                if (validator != null) {
                    //挖掘点 validate函数的源码
                    validator.validate(invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());
                }
            }
            //Dubbo源码里捕获的是ValidationException这个异常,原始信息变成了字符串，所以接下来
            //通过JValidator源码分析进行如下扩展
            catch (ConstraintViolationException e) {
                //获取我们的异常，这里的异常时集合的，因为我们参数可能多个都不通过
                StringBuilder message = new StringBuilder();
                Set<ConstraintViolation<?>> violations = e.getConstraintViolations();
                for (ConstraintViolation<?> violation : violations) {
                    //这里只获取第一个不通过的原因
                    message.append(violation.getMessage().concat(";"));
                    break;
                }
                //项目自定义异常类型，网关可以捕获到该异常
                throw new ParamException(message.toString());
            } catch (RpcException e) {
                throw e;
            } catch (Throwable t) {
                return AsyncRpcResult.newDefaultAsyncResult(t, invocation);
            }
        }
        return invoker.invoke(invocation);
    }
}

JValidator源码分析

//Validated 校验过程
public void validate(String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Exception {
  List<Class<?>> groups = new ArrayList<>();
  Class<?> methodClass = methodClass(methodName);
  if (methodClass != null) {
    groups.add(methodClass);
  }
  //异常返回信息 violations
  Set<ConstraintViolation<?>> violations = new HashSet<>();
  Method method = clazz.getMethod(methodName, parameterTypes);
  Class<?>[] methodClasses;
  if (method.isAnnotationPresent(MethodValidated.class)){
    methodClasses = method.getAnnotation(MethodValidated.class).value();
    groups.addAll(Arrays.asList(methodClasses));
  }
  // add into default group
  groups.add(0, Default.class);
  groups.add(1, clazz);
  // convert list to array
  Class<?>[] classgroups = groups.toArray(new Class[groups.size()]);
  Object parameterBean = getMethodParameterBean(clazz, method, arguments);
  if (parameterBean != null) {
    violations.addAll(validator.validate(parameterBean, classgroups ));
  }
  for (Object arg : arguments) {
    validate(violations, arg, classgroups);
  }
  if (!violations.isEmpty()) {
    logger.error("Failed to validate service: " + clazz.getName() + ", method: " + methodName + ", cause: " + violations);
    //这里原始异常是它，所以我们需要捕获它，得到violations
    throw new ConstraintViolationException("Failed to validate service: " + clazz.getName() + ", method: " + methodName + ", cause: " + violations, violations);
  }
}

自定义DubboExceptionFilter

@Slf4j
@Activate(group = CommonConstants.PROVIDER)
public class DubboExceptionFilter extends ExceptionFilter {
   @Override
   public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
       return invoker.invoke(invocation);
  }
   @Override
   public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
       log.error("dubbo global exception ---------->{}", appResponse.getException());
       if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
           try {
               Throwable exception = appResponse.getException();
               // 自定义异常处理
               if (exception instanceof ParamException) {
              //按项目log收集规范输出
                   log.error("dubbo service exception ---------->{}", exception);
                   return;
              }
              ......
          } catch (Throwable e) {
               log.warn("Fail to ExceptionFilter when called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
          }
      }
  }
}

PRC单元测试

org.apache.dubbo.remoting.RemotingException: com.xx.exception.ParamException: periodId不能为空;
com.xx.exception.ParamException: periodId不能为空;

高级进阶

在业务里一个DTO对象会用于新增或更新，在新增时不需要主键ID，在更新时需要主键ID。
就需要引入分组的概念了。

定义validation groups，在API模块中定义两个分组。

//用于新增
public interface InsertValidation {
}
//用于更新
public interface UpdateValidation {
}

定义 DTO

@Data
public class PracticeParam implements Serializable {
    //只用于更新
    @NotNull(groups={UpdateValidation.class},message = "id不能为空")
    private Integer id;
    // 如果两组校验都需要可以省去group的定义，完整的如下
    // @NotBlank(groups={InsertValidation.class, UpdateValidation.class},message = "名称不能为空")
    @NotNull(message = "periodId不能为空")
    private Long periodId;
}

服务提供者interface

public interface IPracticeService {
    //因为periodId参数是默认不区分组的，所以这里省去了Validated注解
    boolean practiceAdd(PracticeParam practiceParam);
    boolean practiceEdit(@Validated(value = {UpdateValidation.class}) PracticeParam practiceParam);
}

看看真实运行情况。

生产环境验证