dubbo服務引用之創(chuàng)建Invoker流程詳解
1、創(chuàng)建Invoker流程
1.1、收集引用參數
ReferenceConfig#init方法的結尾處，調用createProxy方法，采集的參數集合作為入參傳遞到該方法中。
1.2、創(chuàng)建Invoker
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
private T createProxy(Map<String, String> map) {
...
// 創(chuàng)建服務代理
return (T) proxyFactory.getProxy(invoker);
}方法主要邏輯就是
- 1、默認情況下如果本地有服務暴露，則引用本地服務.
- 2、用戶寫死了引用的URL，指定的URL可能是對點對直連地址，也可能是注冊中心URL
- 3、通過注冊中心配置拼裝URL，List<URL> us = loadRegistries(false); 用戶配置了幾個注冊中心，就會產生幾個URL
不管走哪種引用類型，都會執(zhí)行下面的核心代碼
invoker = refprotocol.refer(interfaceClass, url);
refprotocol是一個Protocol接口，getAdaptiveExtension返回的是一個Protocol$Adpative，在該類的源碼及其生產方法。
private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
又看到了Protocol這個接口
@SPI("dubbo")
public interface Protocol {
int getDefaultPort();
@Adaptive
<T> Exporter<T> export(Invoker<T> var1) throws RpcException;
@Adaptive
<T> Invoker<T> refer(Class<T> var1, URL var2) throws RpcException;
void destroy();
}主要任務是，暴露遠程服務、引用遠程服務、釋放協議【釋放暴露于引用服務時占用的資源】，dubbo支持多種協議，http,thrift,RMI等，真是通過dubbo的SPI機制，才可以靈活的在這些協議來回切換。
我們第二種為例講一下核心邏輯，同服務暴露時的一樣，因為Dubbo的AOP機制，在獲RegistryProtocol時，會經過兩個Wrapper類的包裝
這個地方也不例外，但是兩個Wrapper類的ProtocolFilterWrapper，ProtocolListenerWrapper并無實際的業(yè)務邏輯，我們直接跳過。
執(zhí)行refprotocol.refer(interfaceClass, url)即執(zhí)行RegistryProtocol#refer代碼。
@SuppressWarnings("unchecked")
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
String group = qs.get(Constants.GROUP_KEY);
if (group != null && group.length() > 0) {
if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
|| "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
return doRefer(cluster, registry, type, url);
}1.2.1、 連接zookeeper
Registry registry = registryFactory.getRegistry(url);
在服務發(fā)布的時候，已經講過了，其主要核心作用就是連接zookeeper服務器，并返回一個ZookeeperRegistry實例。
在RegistryProtocol#doRefer方法中，通過ZookeeperRegistry#register的執(zhí)行，創(chuàng)建引用服務的consumers節(jié)點。創(chuàng)建如下節(jié)點：
/dubbo/com.alibaba.dubbo.demo.DemoService/consumers/consumer%3A%2F%2F192.168.43.156%2Fcom.alibaba.dubbo.demo.DemoService%3Fapplication%3Ddemo-consumer%26category%3Dconsumers%26check%3Dfalse%26dubbo%3D2.0.0%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D15775%26side%3Dconsumer%26timestamp%3D1525073802234
1.2.2、 監(jiān)聽節(jié)點
1.2.3、 加入集群
cluster.join(directory)
cluster也是一個帶有Adaptive注解的擴展類，默認實現時FailoverCluster
@SPI(FailoverCluster.NAME)
public interface Cluster {
/**
* Merge the directory invokers to a virtual invoker.
*
* @param <T>
* @param directory
* @return cluster invoker
* @throws RpcException
*/
@Adaptive
<T> Invoker<T> join(Directory<T> directory) throws RpcException;
}進入FailoverCluster#join,返回FailoverClusterInvoker，一個可以失敗轉移的Invoker，
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailoverClusterInvoker<T>(directory);
}
}FailoverClusterInvoker源碼中，它實現了父類中的一個模板子方法doInvoke。
父類AbstractClusterInvoker的invoke方法，
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}方法list(invocation)返回了List<Invoker<T>> invokers，這些Invoker就是實際與服務交互的對象，
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}我們在構造FailoverClusterInvoker時，傳入的Directory實現類是RegistryDirectory，即AbstractDirectory#list方法。
1.2.4、 核心類DubboInvoker
DubboInvoker 最終Invoker執(zhí)行的方法是：
@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
inv.setAttachment(Constants.VERSION_KEY, version);
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
return new RpcResult();
} else if (isAsync) {
ResponseFuture future = currentClient.request(inv, timeout);
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
} 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);
}
}調用invoker，白話描述就是：
將通過遠程通信將Invocation信息傳遞給服務器端，服務器端接收到該Invocation信息后，找到對應的本地Invoker，然后通過反射執(zhí)行相應的方法，將方法的返回值再通過遠程通信將結果傳遞給客戶端。
這里分3種情況：
- 執(zhí)行方法不需要返回值：直接調用ExchangeClient.send()方法
- 執(zhí)行方法的結果需要異步返回：使用ExchangeClient.request()方法返回一個ResponseFuture對象，通過RpcContext中的ThreadLocal使ResponseFuture和當前線程綁定，未等服務端響應結果就直接返回，然后服務端通過ProtocolFilterWrapper.buildInvokerChain()方法會調用Filter.invoke()方法，即FutureFilter.invoker()->asyncCallback()，會獲取RpcContext的ResponseFuture對象，異步返回結果
- 執(zhí)行方法的結果需要同步返回：使用ExchangeClient.request()方法，返回一個ResponseFuture，一直阻塞到服務端返回響應結果
返回FailoverClusterInvoker
以上就是dubbo服務引用二之創(chuàng)建Invoker的詳細內容，更多關于dubbo服務引用創(chuàng)建Invoker的資料請關注腳本之家其它相關文章！
相關文章
使用mybatis-plus的insert方法遇到的問題及解決方法（添加時id值不存在異常）
這篇文章主要介紹了使用mybatis-plus的insert方法遇到的問題及解決方法（添加時id值不存在異常）,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2020-08-08
跨域問題是瀏覽器為了保護用戶的信息安全,實施了同源策略（Same-Origin Policy）,即只允許頁面請求同源（相同協議、域名和端口）的資源,當 JavaScript 發(fā)起的請求跨越了同源策略,即請求的目標與當前頁面的域名、端口、協議不一致時,瀏覽器會阻止請求的發(fā)送或接收2025-03-03
Mybatis分頁插件Pagehelper的PageInfo字段屬性使用及解釋
這篇文章主要介紹了Mybatis分頁插件Pagehelper的PageInfo字段屬性使用及解釋,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-05-05
這篇文章主要為大家詳細介紹了java基于GUI實現簡單畫筆小畫板，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-06-06
這篇文章主要介紹了springmvc參數為對象,數組的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08
這篇文章主要介紹了java關于Date日期類型的大小比較,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-08-08
Java?guava框架LoadingCache及CacheBuilder本地小容量緩存框架總結
Guava?Cache本地緩存框架主要是一種將本地數據緩存到內存中,但數據量并不能太大,否則將會占用過多的內存,本文給大家介紹Java?guava框架?LoadingCache及CacheBuilder?本地小容量緩存框架總結,感興趣的朋友一起看看吧2023-12-12
泛型在java中有很重要的地位，在面向對象編程及各種設計模式中有非常廣泛的應用。泛型是參數化類型的應用，操作的數據類型不限定于特定類型，可以根據實際需要設置不同的數據類型，以實現代碼復用。下面小編來簡單講一講泛型2019-05-05
這篇文章主要介紹了淺談Java向下轉型的意義，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-02-02
作者是一個來自河源的大三在校生，以下筆記都是作者自學之路的一些淺薄經驗，如有錯誤請指正，將來會不斷的完善筆記，幫助更多的Java愛好者入門2022-03-03