ITPub博客

首页 > 数据库 > 数据库开发技术 > 深度分析 Dubbo SPI 源码,扩展 Dubbo Validation (groups)

深度分析 Dubbo SPI 源码,扩展 Dubbo Validation (groups)

原创 数据库开发技术 作者:玄姐聊AGI 时间:2021-12-03 21:00:00 0 删除 编辑

目前架构是网关直接通过泛化调用Dubbo服务,不同于web Controller使用spring mvc模块来做到参数注解校验。不过不用担心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);
}

完工,这样我们就齐活了,截个图看看真实运行情况吧。



-     生产环境验证    -



感谢您的耐心看到这里,学海无涯继续奋斗!

原文作者:玄姐聊AGI;如有侵权,请联系管理员删除。

请登录后发表评论 登录
全部评论

注册时间:2020-11-03

  • 博文量
    314
  • 访问量
    143487