我们在日常使用dubbo实现rpc时，都会通过引用提供者提供的jar包中的接口，在我们需要调用的类中，注入该接口的一个实现bean，通过调用该bean的方法来完成一次rpc调用，使用方式非常简单，但是大家有没有想过dubbo是如何实现这个功能的呢？
@Component("demoServiceComponent")
public class DemoServiceComponent implements DemoService {
@DubboReference
private DemoService demoService;
@Override
public String sayHello(String name) {
return demoService.sayHello(name);
}
@Override
public CompletableFuture sayHelloAsync(String name) {
return null;
}
} 其实，实现这个功能和dubbo中的一个类referenceBean有关！
在dubbo中，我们所使用rpc接口实际会在spring容器中生成两个bean，一个是该接口对应的referenceBean，而另一个则是该接口的一个代理实现类，我们进行rpc调用时，实际是这个代理实现类帮我们完成的远程通信，返回处理等 *** 作。但是，这个代理实现类又是通过referencce来生成的，接下来我们就会讲到这个是如何生成的。
首先，这一切的入口都在ReferenceAnnotationBeanPostProcessor中(beanPostProssor是spring的bean后置处理器，可以在bean生命周期的某些阶段对bean做一些加工 *** 作，后期有时间可以和大家分享一下这块的知识)，这个beanPostProcessor在处理bean的属性注入时(DemoServiceComponent )，生成了referenceBean.
@Override
protected Object doGetInjectedBean(AnnotationAttributes attributes, Object bean, String beanName, Class injectedType,
InjectionMetadata.InjectedElement injectedElement) throws Exception {
/**
* The name of bean that annotated Dubbo's {@link Service @Service} in local Spring {@link ApplicationContext}
*/
String referencedBeanName = buildReferencedBeanName(attributes, injectedType);
/**
* The name of bean that is declared by {@link Reference @Reference} annotation injection
*/
String referenceBeanName = getReferenceBeanName(attributes, injectedType);
ReferenceBean referenceBean = buildReferenceBeanIfAbsent(referenceBeanName, attributes, injectedType);
registerReferenceBean(referencedBeanName, referenceBean, attributes, injectedType);
cacheInjectedReferenceBean(referenceBean, injectedElement);
return getOrCreateProxy(referencedBeanName, referenceBeanName, referenceBean, injectedType);
}通过buildReferencedBeanName(attributes, injectedType)方法创建了一个referenceBean.再通过referenceBean的get()获取了代理实现类
private Object getOrCreateProxy(String referencedBeanName, String referenceBeanName, ReferenceBean referenceBean, Class serviceInterfaceType) {
if (existsServiceBean(referencedBeanName)) { // If the local @Service Bean exists, build a proxy of ReferenceBean
return newProxyInstance(getClassLoader(), new Class[]{serviceInterfaceType},
wrapInvocationHandler(referenceBeanName, referenceBean));
} else { // ReferenceBean should be initialized and get immediately
/**
* TODO, if we can make sure this happens after {@link DubboLifecycleComponentApplicationListener},
* TODO, then we can avoid starting bootstrap in here, because bootstrap should has been started.
*/
return referenceBean.get();
}
}那么这个referenceBean.get()方法到底做了什么呢？
public synchronized T get() {
if (destroyed) {
throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}点进去可以发现，这个方法就是获取referenceBean中的一个属性，如果该属性不存在就先执行初始化init()方法，我们第一次执行当然是不存在，所以我们继续看init()方法。
protected synchronized void init() {
...................
/*获取代理实现类*/
ref = createProxy(referenceParameters);
.................
} @SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
private T createProxy(Map map) {
if (shouldJvmRefer(map)) {
.............
} else {
.................
if (urls.size() == 1) {
/*单体*/
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
} else {
/*集群*/
List> invokers = new ArrayList>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
if (UrlUtils.isRegistry(url)) {
registryURL = url; // use last registry url
}
}
if (registryURL != null) { // registry url is available
// for multi-subscription scenario, use 'zone-aware' policy by default
URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
// The invoker wrap relation would be like: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
invoker = CLUSTER.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = CLUSTER.join(new StaticDirectory(invokers));
}
}
}
if (shouldCheck() && !invoker.isAvailable()) {
throw new IllegalStateException("Failed to check the status of the service "
+ interfaceName
+ ". No provider available for the service "
+ (group == null ? "" : group + "/")
+ interfaceName +
(version == null ? "" : ":" + version)
+ " from the url "
+ invoker.getUrl()
+ " to the consumer "
+ NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
/**
* @since 2.7.0
* ServiceData Store
*/
String metadata = map.get(METADATA_KEY);
WritableMetadataService metadataService = WritableMetadataService.getExtension(metadata == null ? DEFAULT_METADATA_STORAGE_TYPE : metadata);
if (metadataService != null) {
URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
metadataService.publishServiceDefinition(consumerURL);
}
// 生成代理类
return (T) PROXY_FACTORY.getProxy(invoker);
} 最终是 PROXY_FACTORY.getProxy(invoker)方法完成的代理类生成。那么这个方法做了什么呢？
public class JavassistProxyFactory extends AbstractProxyFactory {
@Override
@SuppressWarnings("unchecked")
public T getProxy(Invoker invoker, Class[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
} 其实重点还是在Proxy.getProxy(interfaces)，interface明显是我们需要远程调用的接口类，通过调用这个方法返回对象的newInstance方法来返回我们所需要的代理类。那么这个方法返回了什么呢？
public class proxy0 implements ClassGenerator.DC, EchoService, DemoService{
public static Method[] methods;
private InvocationHandler handler;
public String sayHello(String paramString){
Object[] arrayOfObject = new Object[1];
arrayOfObject[0] = paramString;
Object localObject = this.handler.invoke(this, methods[0], arrayOfObject);
return ((String)localObject);
}
public Object $echo(Object paramObject)
{
Object[] arrayOfObject = new Object[1];
arrayOfObject[0] = paramObject;
Object localObject = this.handler.invoke(this, methods[1], arrayOfObject);
return ((Object)localObject);
}
public proxy0(){
}
public proxy0(InvocationHandler paramInvocationHandler){
this.handler = paramInvocationHandler;
}
}其中sayHello方法是我们需要调用的业务方法，从这个例子中我们可以看到当我们进行rpc时实际上是调用的this.handler.invoke()方法，handler就是我们之前看到的
Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker))的入参
继续往下看就可以发现实际是调用的invoker类的方法
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
String methodName = method.getName();
Class[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 0) {
if ("toString".equals(methodName)) {
return invoker.toString();
} else if ("$destroy".equals(methodName)) {
invoker.destroy();
return null;
} else if ("hashCode".equals(methodName)) {
return invoker.hashCode();
}
} else if (parameterTypes.length == 1 && "equals".equals(methodName)) {
return invoker.equals(args[0]);
}
RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface().getName(), args);
rpcInvocation.setTargetServiceUniqueName(invoker.getUrl().getServiceKey());
return invoker.invoke(rpcInvocation).recreate();
}而invoker是在之前通过 invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0))生成的。大家看一下源码就可以知道最终实际上是生成了一个DubboInovker的对象。DubboInovker实际上做的事呢就是帮我们与另一台服务器通信.
@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {
.............................
try {
..................
if (isOneway)
..................
} else {
ExecutorService executor = getCallbackExecutor(getUrl(), inv);
/*远程通信*/
CompletableFuture appResponseFuture =
currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);
FutureContext.getContext().setCompatibleFuture(appResponseFuture);
AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);
result.setExecutor(executor);
return result;
}
}
}
}所以，当我们调用接口进行rpc时，实际上是调用的dubboInvoker的doInvoke方法，由它帮我们完成rpc.而dubboInvoker以及这个代理实现类又是通过referenceBean来创建的。
最后，本此分享还有很多东西没有给大家详解，dubbo是如何通过注册中心完成注册发现的，dubbo的spi机制，dubbo的filter机制、负载均衡以及集群容错机制等，后续会继续给大家分享。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）