深度解析dubbo过滤器之泛化调用
程序员文章站
2024-02-21 12:12:16
...
本文基于dubbo v2.6.x
1. 泛化调用
什么是泛化调用？我个人是这样理解的，服务调用者不需要引用服务接口就能够调用服务提供者，我们用过dubbo的都知道，在开发的时候服务双方需要引用相同的接口，这种其实依赖性还是比较强，但是有某些场景使用共同接口的开发方式并不方便，dubbo提供了泛化调用的功能，我们服务调用端不需要定义共同的情况下就可以调用服务提供者。
2.泛化调用实例
我们这里简单演示下泛化调用的使用，这里使用的spring的方式，如果想体验下api的方式可以访问dubbo官网，官网提供了spring与api的实例。
2.1 服务提供者
服务提供者 定义接口，然后书写实现类 实现接口，并把进行服务暴露。服务提供者端就按照正常的开发流程开发就行。
接口：
public interface IUserInfoService {
public String getUserName(Integer id);
}
实现类：
@org.springframework.stereotype.Service
@Service
public class IUserInfoServiceImpl implements IUserInfoService {
@Override
public String getUserName(Integer id) {
return "0";
}
}
2.2 服务调用者
之前开发服务调用者，需要引用服务提供者共同的接口，咱们这里泛化调用就不需要了，dubbo给我们提供了一个泛化接口GenericService，我们在使用的时候直接注入就可以了。
咱们这里先配置一下：
<dubbo:reference interface="com.xuzhaocai.dubbo.provider.IUserInfoService" id="iUserInfoService" generic="true" check="false" retries="0">
</dubbo:reference>
这里我们虽然没有IUserInfoService 接口，但是interface需要写上，同时generic="true"表示使用泛化。
使用：我这里直接使用controller 注入iUserInfoService，我们要关心的是注入的类型GenericService。
在使用的时候，调用GenericService的$invoke 方法，参数： 第一参数是 调用的方法名， 第二个是参数的类型们，第三个就是参数值。
2.3 调用测试
我们来使用postman测试下
我们可以看到调用成功了，那dubbo是怎样实现的呢，我们接下来看下它的实现源码。
3.泛化调用源码分析
这个要从服务调用端的服务引用说起，首先在ReferenceConfig的init方法中，有这么一段代码：
如果是泛化调用，就将interfaceClass设置成GenericService.class,这时候interfaceName 属性还是咱们那个配置的那个接口，在上面案例中就是com.xuzhaocai.dubbo.provider.IUserInfoService，接着在一个收集信息的map 添加interface=com.xuzhaocai.dubbo.provider.IUserInfoService 键值对。
再往后走将map拼成一个长串，然后添加refer参数中。
同时我们生成的代理类，实现了GenericService 接口，重写$invoke方法。下面截图是我使用arthas获取的代理类。
接着当注册中心变动的时候会通知RegistryDirectory进行toInvokers，在new DubboInvoker 的构造中，有这么一段代码，
我们在看下super的处理：
这个convertAttachment 其实就是从url中获取这个属性，然后将值塞到AbstractInvoker的attachment成员中。接下来我们再来看下服务调用者调用过程。
我们在服务调用端 invoke的时候会调用到一个GenericImplFilter ，我们看下源码实现
@Activate(group = Constants.CONSUMER, value = Constants.GENERIC_KEY, order = 20000)
public class GenericImplFilter implements Filter {
private static final Logger logger = LoggerFactory.getLogger(GenericImplFilter.class);
private static final Class<?>[] GENERIC_PARAMETER_TYPES = new Class<?>[]{String.class, String[].class, Object[].class};
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
String generic = invoker.getUrl().getParameter(Constants.GENERIC_KEY);// 是否是泛化调用
if (ProtocolUtils.isGeneric(generic) // 是泛化调用
&& !Constants.$INVOKE.equals(invocation.getMethodName())// 这里这个方法名不是$invoke
&& invocation instanceof RpcInvocation) {
RpcInvocation invocation2 = (RpcInvocation) invocation;
String methodName = invocation2.getMethodName();//方法名
Class<?>[] parameterTypes = invocation2.getParameterTypes();// 参数类型们
Object[] arguments = invocation2.getArguments();// 具体参数
// 将参数类型class 们转成 string类型
String[] types = new String[parameterTypes.length];
for (int i = 0; i < parameterTypes.length; i++) {
types[i] = ReflectUtils.getName(parameterTypes[i]);
}
Object[] args;
if (ProtocolUtils.isBeanGenericSerialization(generic)) {// bean
args = new Object[arguments.length];
for (int i = 0; i < arguments.length; i++) {// 进行序列化
args[i] = JavaBeanSerializeUtil.serialize(arguments[i], JavaBeanAccessor.METHOD);
}
} else {
args = PojoUtils.generalize(arguments);
}
// 这里封装成 $invoke的 形式，就是封装成 跟调用 GenericService接的 $invoke 方法一样
invocation2.setMethodName(Constants.$INVOKE);
invocation2.setParameterTypes(GENERIC_PARAMETER_TYPES);
invocation2.setArguments(new Object[]{methodName, types, args});
Result result = invoker.invoke(invocation2);//交给下一个invoker
if (!result.hasException()) {// 没有异常
Object value = result.getValue();// 获取结果值
try {// 获取原先调用的那个method
Method method = invoker.getInterface().getMethod(methodName, parameterTypes);
if (ProtocolUtils.isBeanGenericSerialization(generic)) {// bean
if (value == null) {
return new RpcResult(value);
} else if (value instanceof JavaBeanDescriptor) {// 需要反序列化
return new RpcResult(JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) value));
} else {
throw new RpcException(
"The type of result value is " +
value.getClass().getName() +
" other than " +
JavaBeanDescriptor.class.getName() +
", and the result is " +
value);
}
} else {// 使用Pojo的反序列化
return new RpcResult(PojoUtils.realize(value, method.getReturnType(), method.getGenericReturnType()));
}
} catch (NoSuchMethodException e) {
throw new RpcException(e.getMessage(), e);
}
} else if (result.getException() instanceof GenericException) {
GenericException exception = (GenericException) result.getException();
try {
String className = exception.getExceptionClass();
Class<?> clazz = ReflectUtils.forName(className);
Throwable targetException = null;
Throwable lastException = null;
try {
targetException = (Throwable) clazz.newInstance();
} catch (Throwable e) {
lastException = e;
for (Constructor<?> constructor : clazz.getConstructors()) {
try {
targetException = (Throwable) constructor.newInstance(new Object[constructor.getParameterTypes().length]);
break;
} catch (Throwable e1) {
lastException = e1;
}
}
}
if (targetException != null) {
try {
Field field = Throwable.class.getDeclaredField("detailMessage");
if (!field.isAccessible()) {
field.setAccessible(true);
}
field.set(targetException, exception.getExceptionMessage());
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
result = new RpcResult(targetException);
} else if (lastException != null) {
throw lastException;
}
} catch (Throwable e) {
throw new RpcException("Can not deserialize exception " + exception.getExceptionClass() + ", message: " + exception.getExceptionMessage(), e);
}
}
return result;
}
///-----------------------------------------------------------------
// 方法名= $invoke
if (invocation.getMethodName().equals(Constants.$INVOKE)
&& invocation.getArguments() != null
&& invocation.getArguments().length == 3
&& ProtocolUtils.isGeneric(generic)) {
Object[] args = (Object[]) invocation.getArguments()[2];
if (ProtocolUtils.isJavaGenericSerialization(generic)) {
for (Object arg : args) {
if (!(byte[].class == arg.getClass())) {
error(generic, byte[].class.getName(), arg.getClass().getName());
}
}
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
for (Object arg : args) {
if (!(arg instanceof JavaBeanDescriptor)) {
error(generic, JavaBeanDescriptor.class.getName(), arg.getClass().getName());
}
}
}
// 设置generic=true
((RpcInvocation) invocation).setAttachment(
Constants.GENERIC_KEY, invoker.getUrl().getParameter(Constants.GENERIC_KEY));
}
return invoker.invoke(invocation);
}
private void error(String generic, String expected, String actual) throws RpcException {
throw new RpcException(
"Generic serialization [" +
generic +
"] only support message type " +
expected +
" and your message type is " +
actual);
}
}
代码中隔离线下面的就是处理$invoke这种的，我们看到最后其实就是往invocation 添加了个generic=true的属性。其他两种需要 进行序列化，nativejava与bean。
我们再来看下服务提供者端的GenericFilter
@Activate(group = Constants.PROVIDER, order = -20000)
public class GenericFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
if (inv.getMethodName().equals(Constants.$INVOKE)// 判断是不是泛化调用
&& inv.getArguments() != null
&& inv.getArguments().length == 3
&& !invoker.getInterface().equals(GenericService.class)) {// interface !=GenericService
String name = ((String) inv.getArguments()[0]).trim();// 获取第一个参数，也就是 方法名
String[] types = (String[]) inv.getArguments()[1];//类型参数
Object[] args = (Object[]) inv.getArguments()[2];// 参数列表
try {
// 通过 interface ， 方法名 ，方法参数类型 获得method
Method method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), name, types);
Class<?>[] params = method.getParameterTypes();
if (args == null) {// 如果是null
args = new Object[params.length];
}
String generic = inv.getAttachment(Constants.GENERIC_KEY);//generic
if (StringUtils.isBlank(generic)) {// 从rpccontext 中获取 generic
generic = RpcContext.getContext().getAttachment(Constants.GENERIC_KEY);
}
if (StringUtils.isEmpty(generic)
|| ProtocolUtils.isDefaultGenericSerialization(generic)) {// generic不是null，然后=true
args = PojoUtils.realize(args, params, method.getGenericParameterTypes());
} else if (ProtocolUtils.isJavaGenericSerialization(generic)) {//nativejava
for (int i = 0; i < args.length; i++) {
if (byte[].class == args[i].getClass()) {
try {
UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream((byte[]) args[i]);
args[i] = ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
.deserialize(null, is).readObject();
} catch (Exception e) {
throw new RpcException("Deserialize argument [" + (i + 1) + "] failed.", e);
}
} else {
throw new RpcException(
"Generic serialization [" +
Constants.GENERIC_SERIALIZATION_NATIVE_JAVA +
"] only support message type " +
byte[].class +
" and your message type is " +
args[i].getClass());
}
}
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {//bean
for (int i = 0; i < args.length; i++) {
if (args[i] instanceof JavaBeanDescriptor) {
args[i] = JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) args[i]);
} else {
throw new RpcException(
"Generic serialization [" +
Constants.GENERIC_SERIALIZATION_BEAN +
"] only support message type " +
JavaBeanDescriptor.class.getName() +
" and your message type is " +
args[i].getClass().getName());
}
}
}
//进行执行
Result result = invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
if (result.hasException()
&& !(result.getException() instanceof GenericException)) {// 有异常，且不是GenericException
return new RpcResult(new GenericException(result.getException()));
}
if (ProtocolUtils.isJavaGenericSerialization(generic)) {
try {
UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
.serialize(null, os).writeObject(result.getValue());
return new RpcResult(os.toByteArray());
} catch (IOException e) {
throw new RpcException("Serialize result failed.", e);
}
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
return new RpcResult(JavaBeanSerializeUtil.serialize(result.getValue(), JavaBeanAccessor.METHOD));
} else {
return new RpcResult(PojoUtils.generalize(result.getValue()));
}
} catch (NoSuchMethodException e) {
throw new RpcException(e.getMessage(), e);
} catch (ClassNotFoundException e) {
throw new RpcException(e.getMessage(), e);
}
}
return invoker.invoke(inv);
}
}
我们可以看到首先判断inv的方法名是不是$invoke,参数个数，interface不是GenericService。然后获取generic，进行反序列化，因为在服务调用端的filter中有对参数值序列化。接着就是重新封装invocation，替换成真正要调用的方法名，方法参数类型，方法参数，然后进行调用用，封装结果或者处理异常返回给调用端。
上一篇： LVS理论学习