dubbo引用
registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
和導(dǎo)出一樣，引用標(biāo)簽，最后會解析成ReferenceBean。
看實(shí)現(xiàn)接口
FactoryBean：所以是工廠方法，最終注入spring的對象是調(diào)用FactoryBean的getObject方法獲得的對象。
ApplicationContextAware：注入ApplicationContext對象。
InitializingBean：初始化了spring之后調(diào)用afterPropertiesSet方法。
DisposableBean：spring容器銷毀會調(diào)用destory方法。這里什么沒做。
public class ReferenceBean<T> extends ReferenceConfig<T> implements FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
@Override
@SuppressWarnings({"unchecked"})
public void afterPropertiesSet() throws Exception {
if (applicationContext != null) {
// 從applicationContext中獲取ConfigCenterBean類型的實(shí)例列表。這里什么也沒做
BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConfigCenterBean.class, false, false);
}
...省略代碼。從spring容器獲取各種配置類set保存到成員方法中
// 是否初始化。dubbo:reference可以配置init屬性。
if (shouldInit()) {
// 真正獲取對象
getObject();
}
}
@Override
public Object getObject() {
return get();
}
public synchronized T get() {
// 檢查和更新一下配置屬性，導(dǎo)出服務(wù)也有這個(gè)方法，不看
checkAndUpdateSubConfigs();
if (destroyed) {
throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}
private void init() {
...省略代碼。各種拼裝配置參數(shù)到map
ref = createProxy(map);
String serviceKey = URL.buildKey(interfaceName, group, version);
ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes));
initialized = true;
}
private T createProxy(Map<String, String> map) {
// 是否jvm內(nèi)部就可以引用。根據(jù)url判定。false
if (shouldJvmRefer(map)) {
URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
invoker = REF_PROTOCOL.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else {
urls.clear(); // reference retry init will add url to urls, lead to OOM
// url不為空，則說明是dubbo:reference直接指定了url調(diào)用
if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (StringUtils.isEmpty(url.getPath())) {
url = url.setPath(interfaceName);
}
// 是否是注冊中心
if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else { // assemble URL from register center's configuration
// if protocols not injvm checkRegistry
// 不是本地協(xié)議引用，通過注冊中心獲取url，常用方式
if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
checkRegistry();
// 加載注冊中心url
List<URL> us = loadRegistries(false);
// 只有一個(gè)注冊中心。us = registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?...
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.");
}
}
}
// urls添加的注冊中心url：registry://。。。這里只有一個(gè)。
if (urls.size() == 1) {
// REF_PROTOCOL自然是protocol擴(kuò)展適配器類，因?yàn)閡rl是registry協(xié)議，所以自然調(diào)用的是RegistryProtocol的refer方法
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
} else {
// 如果urls是多個(gè)注冊中心，則便利各個(gè)url生成多個(gè)invoker，然后用cluster把多個(gè)invoker合并成一個(gè)invoker
。。。暫時(shí)不看
}
// create service proxy
return (T) PROXY_FACTORY.getProxy(invoker);
}
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 組成成zookeeper://..
url = URLBuilder.from(url)
.setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
.removeParameter(REGISTRY_KEY)
.build();
Registry registry = registryFactory.getRegistry(url);
// type是引用的接口
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// 如有有g(shù)roup分組，就從url參數(shù)中獲取匹配。這里沒有
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
String group = qs.get(GROUP_KEY);
if (group != null && group.length() > 0) {
if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
return doRefer(cluster, registry, type, url);
}
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// RegistryDirectory是比較重要的類，有訂閱方法，訂閱從注冊中心獲取遠(yuǎn)程服務(wù)地址就是這個(gè)類
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
// all attributes of REFER_KEY
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
// 構(gòu)建訂閱url。consumer://。。。
URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
// 不是通用接口（interface=*)，切需要注冊。這里符合
if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
// 往Zookeeper中注冊consumer://，就是創(chuàng)建consumer節(jié)點(diǎn)。
registry.register(directory.getRegisteredConsumerUrl());
}
directory.buildRouterChain(subscribeUrl);
// 訂閱consumer://?？梢钥吹絚ategory有三個(gè)：providers、configurators、routes。后面會針對這三個(gè)節(jié)點(diǎn)添加訂閱
directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;
}
public void subscribe(URL url) {
setConsumerUrl(url);
CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this);
serviceConfigurationListener = new ReferenceConfigurationListener(this, url);
// registry是zookeeperRegistry。這里真真添加訂閱
registry.subscribe(url, this);
}
ZookeeperRegistry繼承于FailbackRegistry。跟服務(wù)導(dǎo)出，快速失敗的邏輯是一樣的。
@Override
public void doSubscribe(final URL url, final NotifyListener listener) {
try {
// 是否是通用接口。不看
if (ANY_VALUE.equals(url.getServiceInterface())) {
...省略
} else {
List<URL> urls = new ArrayList<>();
// 前面看到category有三個(gè)。providers、configurators、routes。所以這里path有三個(gè):/dubbo/**接口名/providers，/dubbo/**接口名/configurators、/dubbo/**接口名/routes。
for (String path : toCategoriesPath(url)) {
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
if (listeners == null) {
zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
listeners = zkListeners.get(url);
}
ChildListener zkListener = listeners.get(listener);
if (zkListener == null) {
// 監(jiān)聽器是內(nèi)部類，最終監(jiān)聽調(diào)用的是ZookeeperRegistry.this.notify方法
listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
zkListener = listeners.get(listener);
}
// zkClient內(nèi)部邏輯會判定，如果沒有這個(gè)節(jié)點(diǎn)就創(chuàng)建這個(gè)節(jié)點(diǎn)。
zkClient.create(path, false);
// zkClient跟服務(wù)導(dǎo)出一樣是：CuratorZookeeperClient。
// 添加監(jiān)聽器，并且返回節(jié)點(diǎn)下的子節(jié)點(diǎn)。如果子節(jié)點(diǎn)為空，則通過consumer協(xié)議構(gòu)建empty://協(xié)議返回。
// 所以providers節(jié)點(diǎn)下自然是服務(wù)地址方地址（可以登錄zookeeper查看）
// configurators和routes沒有子節(jié)點(diǎn)。所以返回empty://。
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 添加完之后，手動調(diào)用喚醒方法。前面添加了三個(gè)節(jié)點(diǎn)的監(jiān)聽器。并返回了三個(gè)節(jié)點(diǎn)下的子節(jié)點(diǎn)。
// 一個(gè)是dubbo://...真正的服務(wù)地址地址。另外兩個(gè)是empty://地址
notify(url, listener, urls);
}
} catch (Throwable e) {
throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
// keep every provider's category.
Map<String, List<URL>> result = new HashMap<>();
for (URL u : urls) {
if (UrlUtils.isMatch(url, u)) {
// 獲取url的category參數(shù)，有：providers、configurators、routes。
String category = u.getParameter(CATEGORY_KEY, DEFAULT_CATEGORY);
// 以category分組，保存在result
List<URL> categoryList = result.computeIfAbsent(category, k -> new ArrayList<>());
categoryList.add(u);
}
}
if (result.size() == 0) {
return;
}
Map<String, List<URL>> categoryNotified = notified.computeIfAbsent(url, u -> new ConcurrentHashMap<>());
for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
String category = entry.getKey();
List<URL> categoryList = entry.getValue();
categoryNotified.put(category, categoryList);
// listener是之前傳入的RegistryDirectory。categoryList是result遍歷：有dubbo://...，empty://,empty://
listener.notify(categoryList);
// We will update our cache file after each notification.
// When our Registry has a subscribe failure due to network jitter, we can return at least the existing cache URL.
saveProperties(url);
}
}
@Override
public synchronized void notify(List<URL> urls) {
// 過濾，并按vonfigurator、route、provider分三組。
Map<String, List<URL>> categoryUrls = urls.stream()
.filter(Objects::nonNull)
.filter(this::isValidCategory)
.filter(this::isNotCompatibleFor26x)
.collect(Collectors.groupingBy(url -> {
if (UrlUtils.isConfigurator(url)) {
return CONFIGURATORS_CATEGORY;
} else if (UrlUtils.isRoute(url)) {
return ROUTERS_CATEGORY;
} else if (UrlUtils.isProvider(url)) {
return PROVIDERS_CATEGORY;
}
return "";
}));
// 如果是configurator協(xié)議則toConfigurators處理，否則不處理。toConfigurators方法內(nèi)部會忽略empty://。
List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);
// 如果是route協(xié)議則toRoute處理，否則不處理。toRoute方法內(nèi)部會忽略empty://。
List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
toRouters(routerURLs).ifPresent(this::addRouters);
// providers協(xié)議則refreshOverrideAndInvoker處理。
List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());
refreshOverrideAndInvoker(providerURLs);
}
private void refreshInvoker(List<URL> invokerUrls) {
Assert.notNull(invokerUrls, "invokerUrls should not be null");
// 如果providers節(jié)點(diǎn)下已經(jīng)空了，則說明服務(wù)提供沒了，則這里invokerUrls則是empty：//，這里就會調(diào)用destroyAllInvokers方法銷毀所有invoke。
if (invokerUrls.size() == 1
&& invokerUrls.get(0) != null
&& EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
this.forbidden = true; // Forbid to access
this.invokers = Collections.emptyList();
routerChain.setInvokers(this.invokers);
destroyAllInvokers(); // Close all invokers
} else {
this.forbidden = false; // Allow to access
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
if (invokerUrls == Collections.<URL>emptyList()) {
invokerUrls = new ArrayList<>();
}
if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
this.cachedInvokerUrls = new HashSet<>();
this.cachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
}
if (invokerUrls.isEmpty()) {
return;
}
// 真正通過provider://，構(gòu)建invoke類。key手機(jī)provider url。v
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);// Translate url list to Invoker map
。。。省略代碼
try {
// 新舊invokder對比，銷毀不可用的舊的invoker。
destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
。。。省略代碼
// url是dubbo://,所以這里自然調(diào)用的是DubboProtocol.refer方法
invoker = new InvokerDelegate<>(protocol.refer(serviceType, url), url, providerUrl);
return newUrlInvokerMap;
}
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
}
@Override
public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
// 如果url有optimize參數(shù)，就會調(diào)用對應(yīng)的接口方法，注冊。這里沒有不管
optimizeSerialization(url);
// create rpc invoker.
//真正構(gòu)建invoker。getClient猜都能猜到是建立遠(yuǎn)程服務(wù)提供者的服務(wù)連接
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
private ExchangeClient[] getClients(URL url) {
// 是否使用共享連接。不同的遠(yuǎn)程接口，相同的遠(yuǎn)程應(yīng)用應(yīng)該會使用同一個(gè)連接
boolean useShareConnect = false;
int connections = url.getParameter(CONNECTIONS_KEY, 0);
List<ReferenceCountExchangeClient> shareClients = null;
// 下文的英文注釋，也可以看出來，如果沒有手動配置連接，則是共享連接，一個(gè)服務(wù)只有一個(gè)連接
// if not configured, connection is shared, otherwise, one connection for one service
if (connections == 0) {
useShareConnect = true;
/**
* The xml configuration should have a higher priority than properties.
*/
String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY,
DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
//
shareClients = getSharedClient(url, connections);
}
ExchangeClient[] clients = new ExchangeClient[connections];
for (int i = 0; i < clients.length; i++) {
if (useShareConnect) {
clients[i] = shareClients.get(i);
} else {
clients[i] = initClient(url);
}
}
return clients;
}
private List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
// 獲得ip和端口：192.168.106.1:20880
String key = url.getAddress();
List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key);
if (checkClientCanUse(clients)) {
batchClientRefIncr(clients);
return clients;
}
// 這個(gè)鎖的方式可以借鑒一下。我之前一直synchronized("".internal())。
locks.putIfAbsent(key, new Object());
synchronized (locks.get(key)) {
clients = referenceClientMap.get(key);
// dubbo check
if (checkClientCanUse(clients)) {
batchClientRefIncr(clients);
return clients;
}
connectNum = Math.max(connectNum, 1);
if (CollectionUtils.isEmpty(clients)) {
// 這里建立連接
clients = buildReferenceCountExchangeClientList(url, connectNum);
referenceClientMap.put(key, clients);
}
。。。省略代碼
return clients;
}
}
private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) {
ExchangeClient exchangeClient = initClient(url);
return new ReferenceCountExchangeClient(exchangeClient);
}
private ExchangeClient initClient(URL url) {
// 獲取連接類型，默認(rèn)是netty
String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
// 添加編碼方式
url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
//設(shè)置心跳時(shí)間
url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT));
// BIO is not allowed since it has severe performance issue.
if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported client type: " + str + "," +
" supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
}
ExchangeClient client;
try {
// connection should be lazy
if (url.getParameter(LAZY_CONNECT_KEY, false)) {
client = new LazyConnectExchangeClient(url, requestHandler);
} else {
// 沒有顯示制定lasy連接，就進(jìn)入這里。requestHandler是DubboProtol的內(nèi)部類，之后處理消息就是這個(gè)類
// 接下來的邏輯和服務(wù)導(dǎo)出差不多，不看了
client = Exchangers.connect(url, requestHandler);
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
}
return client;
}
至此，從注冊中心獲取服務(wù)地址，構(gòu)造invoker看完了。
// // 訂閱完之后，directory內(nèi)部就維護(hù)了以遠(yuǎn)程服務(wù)有連接的invoker
directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));
//cluster是spi set注入的。spi注入的肯定是可適配擴(kuò)展類。因?yàn)閐irectory的url參數(shù)中沒有指明cluster，所以默認(rèn)的是FailoverCluster。外面還有一層MockClusterWrapper，這個(gè)不管。所以這個(gè)invoker可以看成是一個(gè)FailoverClusterInvoker。
Invoker invoker = cluster.join(directory);
接著看怎么進(jìn)行服務(wù)調(diào)用的。注入到spring中的引用對象到底是怎樣的
回到createProxy方法。
// ProxyFactory默認(rèn)是javassist。
return (T) PROXY_FACTORY.getProxy(invoker);
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
// Proxy.getProxy(interfaces)會用javassist動態(tài)構(gòu)建一個(gè)代理。看newInstance傳參就知道，這個(gè)代理的構(gòu)造方法接受一個(gè)InvokerInvocationHandler
// 代理類的邏輯就是。代理類invoke，會調(diào)用構(gòu)造方法傳參過來的InvokerInvocationHandler的invoke方法
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
至此，可以看到，最后返回給spring容器的是一個(gè)代理類。這個(gè)代理類調(diào)用的時(shí)候，其實(shí)最終調(diào)用的就是InvokerInvocationHandler的invoke方法。
所以看一下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]);
}
// 最終調(diào)用的是構(gòu)造方法里傳過來的invoker的invoke方法。之前傳過來的是MockClusterWrapper包裝的FailoverClusterInvoker
// 這里把method和args封裝在RpcInvocation進(jìn)行傳輸
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
看FailoverClusterInvoker的invoke方法
@Override
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
// binding attachments into invocation.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
// 之前cluster維護(hù)了構(gòu)造好的RegistryDirectly。而RegistryDirectly維護(hù)了已經(jīng)和遠(yuǎn)程服務(wù)建立好鏈接的invoker。所以這里不用看邏輯都知道，是把那些invoker取出來。為什么是數(shù)組，因?yàn)榭梢源嬖诙鄠€(gè)連接調(diào)用。
List<Invoker<T>> invokers = list(invocation);
// 獲取負(fù)載均衡策略
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 調(diào)用
return doInvoke(invocation, invokers, loadbalance);
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyInvokers = invokers;
checkInvokers(copyInvokers, invocation);
String methodName = RpcUtils.getMethodName(invocation);
// 獲取重試次數(shù)
int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
// retry loop.
RpcException le = null; // last exception.
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size()); // invoked invokers.
Set<String> providers = new HashSet<String>(len);
// for循環(huán)重試
for (int i = 0; i < len; i++) {
//Reselect before retry to avoid a change of candidate `invokers`.
//NOTE: if `invokers` changed, then `invoked` also lose accuracy.
if (i > 0) {
checkWhetherDestroyed();
copyInvokers = list(invocation);
// check again
checkInvokers(copyInvokers, invocation);
}
...省略代碼
Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
// 真正調(diào)用。這個(gè)invoker就是之前訂閱zookeeper構(gòu)建的RegistryDirectly的包裝過的DubboInvoker
Result result = invoker.invoke(invocation);
return result;
} catch (RpcException e) {
if (e.isBiz()) { // biz exception.
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
throw new RpcException(le.getCode(), "Failed to invoke the method "
+ methodName + " in the service " + getInterface().getName()
+ ". Tried " + len + " times of the providers " + providers
+ " (" + providers.size() + "/" + copyInvokers.size()
+ ") from the registry " + directory.getUrl().getAddress()
+ " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
+ Version.getVersion() + ". Last error is: "
+ le.getMessage(), le.getCause() != null ? le.getCause() : le);
}
dubboInvoker外面包裝的Wrapper等不看
@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(PATH_KEY, getUrl().getPath());
inv.setAttachment(VERSION_KEY, version);
// dubboInvoker內(nèi)部已經(jīng)維護(hù)了跟遠(yuǎn)程服務(wù)建立了連接的client
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
// 方法是否有返回值
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
// 沒有返回值
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
// 發(fā)送接口調(diào)用
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
return AsyncRpcResult.newDefaultAsyncResult(invocation);
} else {
AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
// 發(fā)送接口調(diào)用
CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
responseFuture.whenComplete((obj, t) -> {
if (t != null) {
asyncRpcResult.completeExceptionally(t);
} else {
asyncRpcResult.complete((AppResponse) obj);
}
});
RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
return asyncRpcResult;
}
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
關(guān)于序列化和反序列化，就是netty的handler處理的（反序列化DecodeHandler），通過spi動態(tài)擴(kuò)展選擇哪種序列化方式。具體源碼就不看了。
