簡介
dubbo服務(wù)引用有兩個(gè)時(shí)機(jī)：
- Spring 容器調(diào)用 ReferenceBean 的 afterPropertiesSet 方法時(shí)引用服務(wù)
- 第二個(gè)是在 ReferenceBean 對應(yīng)的服務(wù)被注入到其他類中時(shí)引用
第一個(gè)引用時(shí)機(jī)是餓漢式的，第二個(gè)是懶漢式的佃却。默認(rèn)使用懶漢式的。如果需要使用餓漢式否彩，可通過配置 <dubbo:reference> 的 init 屬性開啟螟蝙。init=true。
服務(wù)用的方式，有三種肌索，第一種是引用本地 (JVM)服務(wù)，第二是通過直連方式引用遠(yuǎn)程服務(wù)特碳，第三是通過注冊中心引用遠(yuǎn)程服務(wù)诚亚。不管是哪種引用方式，最后都會(huì)得到一個(gè) Invoker 實(shí)例午乓。如果有多個(gè)注冊中心站宗，多個(gè)服務(wù)提供者，這個(gè)時(shí)候會(huì)得到一組 Invoker 實(shí)例益愈，此時(shí)需要通過集群管理類 Cluster 將多個(gè) Invoker 合并成一個(gè)實(shí)例梢灭。合并后的 Invoker實(shí)例已經(jīng)具備調(diào)用本地或遠(yuǎn)程服務(wù)的能力了，但并不能將此實(shí)例暴露給用戶使用蒸其，這會(huì)對用戶業(yè)務(wù)代碼造成侵入敏释。此時(shí)框架還需要通過代理工廠類 (ProxyFactory) 為服務(wù)接口生成代理類，并讓代理類去調(diào)用 Invoker 邏輯摸袁。避免了 Dubbo 框架代碼對業(yè)務(wù)代碼的侵入钥顽，同時(shí)也讓框架更容易使用。
ReferenceBean#getObject
服務(wù)引用的入口方法為 ReferenceBean 的 getObject 方法靠汁，該方法定義在 Spring 的 FactoryBean 接口中蜂大，ReferenceBean 實(shí)現(xiàn)了這個(gè)方法
@Override
public Object getObject() {
return get();
}
public synchronized T get() {
checkAndUpdateSubConfigs();
if (destroyed) {
throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}
private void init() {
if (initialized) {
return;
}
checkStubAndLocal(interfaceClass);
checkMock(interfaceClass);
Map<String, String> map = new HashMap<String, String>();
map.put(SIDE_KEY, CONSUMER_SIDE);
appendRuntimeParameters(map);
if (!isGeneric()) {
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put(REVISION_KEY, revision);
}
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if (methods.length == 0) {
logger.warn("No method found in service interface " + interfaceClass.getName());
map.put(METHODS_KEY, ANY_VALUE);
} else {
map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR));
}
}
map.put(INTERFACE_KEY, interfaceName);
appendParameters(map, metrics);
appendParameters(map, application);
appendParameters(map, module);
// remove 'default.' prefix for configs from ConsumerConfig
// appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, consumer);
appendParameters(map, this);
Map<String, Object> attributes = null;
if (CollectionUtils.isNotEmpty(methods)) {
attributes = new HashMap<String, Object>();
for (MethodConfig methodConfig : methods) {
appendParameters(map, methodConfig, methodConfig.getName());
String retryKey = methodConfig.getName() + ".retry";
if (map.containsKey(retryKey)) {
String retryValue = map.remove(retryKey);
if ("false".equals(retryValue)) {
map.put(methodConfig.getName() + ".retries", "0");
}
}
attributes.put(methodConfig.getName(), convertMethodConfig2AyncInfo(methodConfig));
}
}
String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
if (StringUtils.isEmpty(hostToRegistry)) {
hostToRegistry = NetUtils.getLocalHost();
} else if (isInvalidLocalHost(hostToRegistry)) {
throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
}
map.put(REGISTER_IP_KEY, hostToRegistry);
ref = createProxy(map);
String serviceKey = URL.buildKey(interfaceName, group, version);
ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes));
initialized = true;
}
以上代碼就是客戶端服務(wù)引用的核心代碼湿蛔，主要步驟如下：
- 各種校驗(yàn)
- 加載類
- 添加各個(gè)屬性
- 獲取消費(fèi)端ip
- 創(chuàng)建代理類
- 完成創(chuàng)建
創(chuàng)建服務(wù)并引用 createProxy
服務(wù)引用包含本地調(diào)用、遠(yuǎn)程點(diǎn)對點(diǎn)調(diào)用县爬、遠(yuǎn)程注冊中心調(diào)用阳啥。主要看下遠(yuǎn)程注冊中心的調(diào)用代碼
// if protocols not injvm checkRegistry
if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
checkRegistry();
List<URL> us = loadRegistries(false);
if (CollectionUtils.isNotEmpty(us)) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
}
}
if (urls.isEmpty()) {
throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
}
}
多個(gè)服務(wù)提供者參考
if (urls.size() == 1) {
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
} else {
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // use last registry url
}
}
if (registryURL != null) { // registry url is available
// use RegistryAwareCluster only when register's CLUSTER is available
URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
// The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
invoker = CLUSTER.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = CLUSTER.join(new StaticDirectory(invokers));
}
}
獲取invoker后，一系列檢查
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
*/
MetadataReportService metadataReportService = null;
if ((metadataReportService = getMetadataReportService()) != null) {
URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
metadataReportService.publishConsumer(consumerURL);
}
// create service proxy
return (T) PROXY_FACTORY.getProxy(invoker);
創(chuàng)建invoker
Invoker 是 Dubbo 的核心模型财喳，代表一個(gè)可執(zhí)行體察迟。在服務(wù)提供方，Invoker 用于調(diào)用服務(wù)提供類耳高。在服務(wù)消費(fèi)方扎瓶，Invoker 用于執(zhí)行遠(yuǎn)程調(diào)用。Invoker 是由 Protocol 實(shí)現(xiàn)類構(gòu)建而來泌枪。Protocol 實(shí)現(xiàn)類有很多概荷，本節(jié)會(huì)分析最常用的兩個(gè)，分別是 RegistryProtocol 和 DubboProtocol碌燕，其他的大家自行分析误证。下面先來分析 DubboProtocol 的 refer 方法源碼。
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
URL獲取的參數(shù)為
registry://zookeeper.local:2181/org.apache.dubbo.registry.RegistryService?application=springboot-dubbo-consumer&dubbo=2.0.2&pid=76522&qos-accept-foreign-ip=false&qos-enable=true&qos-port=33333&refer=application%3Dspringboot-dubbo-consumer%26dubbo%3D2.0.2%26init%3Dtrue%26interface%3Dcom.mergades.dubboprovider.api.ISayHello%26lazy%3Dfalse%26methods%3DsayHello%26pid%3D76522%26qos-accept-foreign-ip%3Dfalse%26qos-enable%3Dtrue%26qos-port%3D33333%26register.ip%3D192.168.59.83%26release%3D2.7.2%26revision%3D0.0.1-SNAPSHOT%26side%3Dconsumer%26sticky%3Dfalse%26timestamp%3D1565091394680®istry=zookeeper&release=2.7.2&timeout=6000×tamp=1565091399875
根據(jù)SPI機(jī)制修壕，我們可以知道是根據(jù)RegistryProtocol#refer方法實(shí)現(xiàn)的愈捅。
跟蹤代碼，最終看到實(shí)現(xiàn)
注冊中心注冊
if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
registry.register(directory.getRegisteredConsumerUrl());
}
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
生成的invoker對象為
invoker :interface com.mergades.dubboprovider.api.ISayHello -> zookeeper://zookeeper.local:2181/org.apache.dubbo.registry.RegistryService?anyhost=true&application=springboot-dubbo-consumer&bean.name=ServiceBean:com.mergades.dubboprovider.api.ISayHello&check=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&init=true&interface=com.mergades.dubboprovider.api.ISayHello&lazy=false&methods=sayHello&pid=76533&qos-accept-foreign-ip=false&qos-enable=true&qos-port=33333®ister=true®ister.ip=192.168.59.83&release=2.7.2&remote.application=springboot-dubbo-provider&revision=0.0.1-SNAPSHOT&side=consumer&sticky=false×tamp=1565091725210,directory: org.apache.dubbo.registry.integration.RegistryDirectory@6f3f0ae
創(chuàng)建代理對象
PROXY_FACTORY.getProxy(invoker)
private static final ProxyFactory PROXY_FACTORY = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
同樣的慈鸠，根據(jù)SPI機(jī)制查看默認(rèn)實(shí)現(xiàn) JavassistProxyFactory
@Override
public <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException {
Class<?>[] interfaces = null;
String config = invoker.getUrl().getParameter(INTERFACES);
if (config != null && config.length() > 0) {
String[] types = COMMA_SPLIT_PATTERN.split(config);
if (types != null && types.length > 0) {
interfaces = new Class<?>[types.length + 2];
interfaces[0] = invoker.getInterface();
interfaces[1] = EchoService.class;
for (int i = 0; i < types.length; i++) {
// TODO can we load successfully for a different classloader?.
interfaces[i + 2] = ReflectUtils.forName(types[i]);
}
}
}
if (interfaces == null) {
interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class};
}
if (!GenericService.class.isAssignableFrom(invoker.getInterface()) && generic) {
int len = interfaces.length;
Class<?>[] temp = interfaces;
interfaces = new Class<?>[len + 1];
System.arraycopy(temp, 0, interfaces, 0, len);
interfaces[len] = com.alibaba.dubbo.rpc.service.GenericService.class;
}
return getProxy(invoker, interfaces);
}
最終根據(jù)子類的實(shí)現(xiàn)
@Override
@SuppressWarnings("unchecked")
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
最終實(shí)現(xiàn)自己定義類文件蓝谨，下面以 org.apache.dubbo.demo.DemoService 這個(gè)接口為例，來看一下該接口代理類代碼大致是怎樣的
package org.apache.dubbo.common.bytecode;
public class proxy0 implements org.apache.dubbo.demo.DemoService {
public static java.lang.reflect.Method[] methods;
private java.lang.reflect.InvocationHandler handler;
public proxy0() {
}
public proxy0(java.lang.reflect.InvocationHandler arg0) {
handler = $1;
}
public java.lang.String sayHello(java.lang.String arg0) {
Object[] args = new Object[1];
args[0] = ($w) $1;
Object ret = handler.invoke(this, methods[0], args);
return (java.lang.String) ret;
}
}
最終生成對應(yīng)的Proxy對象完成服務(wù)應(yīng)用青团。
服務(wù)調(diào)用
服務(wù)代用首先調(diào)用ReferenceAnnotationBeanPostProcessor#invoke
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Object result;
try {
if (bean == null) { // If the bean is not initialized, invoke init()
// issue: https://github.com/apache/dubbo/issues/3429
init();
}
result = method.invoke(bean, args);
} catch (InvocationTargetException e) {
// re-throws the actual Exception.
throw e.getTargetException();
}
return result;
}
此處校驗(yàn)服務(wù)是否為空譬巫，如果為空則初始化。
代理對象最終執(zhí)行InvokerInvocationHandler#invoke
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
參考
http://dubbo.apache.org/zh-cn/docs/source_code_guide/refer-service.html
