开篇
在Dubbo官方文档中关于泛化调用和泛化实现的说明，这里针对文档的案例做一些简单的说明和解释。
例子
// 引用远程服务
// 该实例很重量，里面封装了所有与注册中心及服务提供方连接，请缓存
ReferenceConfig<GenericService> reference = new ReferenceConfig<GenericService>();
reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
// 弱类型接口名
reference.setInterface("org.apache.dubbo.demo.DemoService");
// 声明为泛化接口
reference.setGeneric("true");
// 用org.apache.dubbo.rpc.service.GenericService可以替代所有接口引用
GenericService genericService = reference.get();
// 基本类型以及Date,List,Map等不需要转换，直接调用
Object result = genericService.$invoke("sayHello",
new String[] {"java.lang.String"}, new Object[] {"12345678"});
System.out.println(result.toString());
- 例子在注释中已经注明了泛化引用的一般步骤，这里不再赘述。
- 核心的本质在于reference.get()的流程获取了consumer的对象，等同于xml通过reference初始化consumer的bean对象。
- 本质上解析<reference>生成consumer的bean对象的过程就是泛化过程中reference.get()的过程。
- 泛化调用唯一需要注意的是对参数的一些限制，基本上参数不需要转换，Class对象等参数需要把保存Class对象的完整类名。
序列化说明
public class Person {
public Person() {
}
public Person(String name, Integer age) {
this.name = name;
this.age = age;
}
private String name;
private Integer age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
}
代码例子-------------
public class Demo {
public static void main(String[] args) {
Person person = new Person();
person.setName("person");
person.setAge(123);
System.out.println("序列化后对象：");
System.out.println(JSON.toJSONString(PojoUtils.generalize(person)));
System.out.println("反序列化后对象：");
System.out.println(JSON.toJSONString(PojoUtils.realize(PojoUtils.generalize(person), Person.class)));
}
}
运行结果-----------------
序列化后对象：
{"name":"person","class":"org.apache.dubbo.demo.consumer.Person","age":123}
反序列化后对象：
{"age":123,"name":"person"}
- 一般的泛化调用都是普通的reference引用，在泛化调用过程中通过PojoUtils.generalize/realize进行序列化和反序列化。
- 在序列化过程中Person对象通过PojoUtils.generalize()序列化的结果当中增加了class字段，然后对序列化的结果进行反序列才能正常反序列化。
- 在dubbo官网有下列说明，指明在泛化调用参数为POJO对象的时候通过使用Map需要带上class类型。
泛化调用序列化补充
源码分析
- Dubbo泛化调用实际是在filter过滤链上执行的序列化和反序列化操作。
- genericimpl对应consumer侧的泛化调用操作。
- generic对应的为provider侧的返回调用操作。
com.alibaba.dubbo.rpc.Filter文件内容
generic=com.alibaba.dubbo.rpc.filter.GenericFilter
genericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFilter
泛化引用
@Activate(group = CommonConstants.CONSUMER, value = GENERIC_KEY, order = 20000)
public class GenericImplFilter extends ListenableFilter {
private static final Class<?>[] GENERIC_PARAMETER_TYPES =
new Class<?>[]{String.class, String[].class, Object[].class};
public GenericImplFilter() {
super.listener = new GenericImplListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
String generic = invoker.getUrl().getParameter(GENERIC_KEY);
//服务端是泛化暴露，客户端不是使用泛化调用场景
if (ProtocolUtils.isGeneric(generic)
&& (!$INVOKE.equals(invocation.getMethodName()) && !$INVOKE_ASYNC.equals(invocation.getMethodName()))
&& invocation instanceof RpcInvocation) {
RpcInvocation invocation2 = new RpcInvocation(invocation);
String methodName = invocation2.getMethodName();
Class<?>[] parameterTypes = invocation2.getParameterTypes();
Object[] arguments = invocation2.getArguments();
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)) {
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);
}
if (RpcUtils.isReturnTypeFuture(invocation)) {
invocation2.setMethodName($INVOKE_ASYNC);
} else {
invocation2.setMethodName($INVOKE);
}
invocation2.setParameterTypes(GENERIC_PARAMETER_TYPES);
invocation2.setArguments(new Object[]{methodName, types, args});
// 客户端调用转换为服务端的泛化调用
return invoker.invoke(invocation2);
// 服务端非泛化暴露，消费使用泛化调用
} else if ((invocation.getMethodName().equals($INVOKE) || invocation.getMethodName().equals($INVOKE_ASYNC))
&& 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());
}
}
}
invocation.setAttachment(
GENERIC_KEY, invoker.getUrl().getParameter(GENERIC_KEY));
}
return invoker.invoke(invocation);
}
}
Dubbo泛化引用过程 - consumer
泛化实现
@Activate(group = CommonConstants.PROVIDER, order = -20000)
public class GenericFilter extends ListenableFilter {
public GenericFilter() {
super.listener = new GenericListener();
}
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
// 如果方法名为$invoker，并且只有3个参数，
// 并且服务端暴露的invoker不是GenericService的相关类
// 则认为本次服务调用时客户端泛化引用服务端，客户端的泛化调用，
// 需要将请求参数反序列化为该接口真实的pojo对象。
if ((inv.getMethodName().equals($INVOKE) || inv.getMethodName().equals($INVOKE_ASYNC))
&& inv.getArguments() != null
&& inv.getArguments().length == 3
&& !GenericService.class.isAssignableFrom(invoker.getInterface())) {
String name = ((String) inv.getArguments()[0]).trim();
String[] types = (String[]) inv.getArguments()[1];
Object[] args = (Object[]) inv.getArguments()[2];
try {
// 根据接口名（API类）、方法名、方法参数类型列表，根据反射机制获取对应的方法。
Method method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), name, types);
Class<?>[] params = method.getParameterTypes();
if (args == null) {
args = new Object[params.length];
}
String generic = inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = RpcContext.getContext().getAttachment(GENERIC_KEY);
}
// 处理普通的泛化引用调用，即处理<dubbo:referecnce generic=“true” …/>,
// 只需要将参数列表Object[]反序列化为pojo即可，
// 具体的反序列化为PojoUtils#realize,
// 其实现原理如下：在JAVA的世界中，pojo通常用map来表示，
// 也就是一个Map可以用来表示一个对象的值，那从一个Map如果序列化一个对象呢？
// 其关键的要素是要在Map中保留该对象的类路径名，
// 也就是通过class来标识该Map需要反序列化的pojo类型。
if (StringUtils.isEmpty(generic)
|| ProtocolUtils.isDefaultGenericSerialization(generic)
|| ProtocolUtils.isGenericReturnRawResult(generic)) {
args = PojoUtils.realize(args, params, method.getGenericParameterTypes());
// 处理< dubbo:reference generic=“nativejava” /> 启用泛化引用，
// 并使用nativejava序列化参数，在服务端这边通过nativejava反序列化参数成pojo对象。
} else if (ProtocolUtils.isJavaGenericSerialization(generic)) {
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(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 [" +
GENERIC_SERIALIZATION_NATIVE_JAVA +
"] only support message type " +
byte[].class +
" and your message type is " +
args[i].getClass());
}
}
// 处理< dubbo:reference generic=“bean” /> 启用泛化引用，
// 并使用javabean序列化参数，在服务端这边通过javabean反序列化参数成pojo对象。
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
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 [" +
GENERIC_SERIALIZATION_BEAN +
"] only support message type " +
JavaBeanDescriptor.class.getName() +
" and your message type is " +
args[i].getClass().getName());
}
}
} else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
// as proto3 only accept one protobuf parameter
if (args.length == 1 && args[0] instanceof String) {
try (UnsafeByteArrayInputStream is =
new UnsafeByteArrayInputStream(((String) args[0]).getBytes())) {
args[0] = ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension("" + GENERIC_SERIALIZATION_PROTOBUF)
.deserialize(null, is).readObject(method.getParameterTypes()[0]);
} catch (Exception e) {
throw new RpcException("Deserialize argument failed.", e);
}
} else {
throw new RpcException(
"Generic serialization [" +
GENERIC_SERIALIZATION_PROTOBUF +
"] only support one" + String.class.getName() +
" argument and your message size is " +
args.length + " and type is" +
args[0].getClass().getName());
}
}
// 序列化API方法中声明的类型，构建new RpcInvocation(method, args, inv.getAttachments())调用环境，继续调用后续过滤器。
return invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
} catch (NoSuchMethodException e) {
throw new RpcException(e.getMessage(), e);
} catch (ClassNotFoundException e) {
throw new RpcException(e.getMessage(), e);
}
}
return invoker.invoke(inv);
}
}
Dubbo泛化引用过程 - provider
网友评论