原 荐 Dubbo服务调用——流程分析
我们根据官网的案例来分析Dubbo的调用过程是什么样的
1.首先粘贴下调用源头
public static void main(String[] args) throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});
context.start();
DemoService demoService = (DemoService) context.getBean("demoService"); // 获取远程服务代理
String hello = demoService.sayHello("world"); // 执行远程方法
System.out.println(hello); // 显示调用结果
System.in.read();
}
断点根据调用堆栈得知，实际上调用了InvokerInvocationHandler的代理对象(动态代理)
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();
}
- 将入方法和入参封装为一个远程调用的实体：RpcInvocation
- 调用MokeClusterInvoker 对象的 invoke方法
public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
if (value.length() == 0 || value.equalsIgnoreCase("false")) {
//1. no mock
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
if (logger.isWarnEnabled()) {
logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
}
//force:direct mock 强制走moke逻辑
result = doMockInvoke(invocation, null);
} else {
//fail-mock 异常moke逻辑
try {
result = this.invoker.invoke(invocation);
} catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
if (logger.isWarnEnabled()) {
logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
}
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
MokeClusterInvoker moke 为正常业务错误时的"备胎"，上面代码执行分为三种逻辑
- 正常逻辑不走moke
- 强制 moke
- 出错 moke
moke 实现方式我们以后分析。 接下来调用 FailoverClusterInvoker 对象 invoke 方法
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance;
// 从 directory 获得所有服务提供者 Invoker 集合
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);
}
//异步的话，需要添加id
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
FailoverClusterInvoker 只是ClusterInvoker 的一种扩展实现。（ 失败自动切换，当出现失败，重试其它服务器）
- 检测是否禁用
- 从Directory 中获取可用的Invoker数组
- 获取负载策略 默认为RandomLoadBalance
我们查看 FailoverClusterInvoker 主业务方法 （失败自动切换，当出现失败，重试其它服务器，通常用于读操作）
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
checkInvokers(copyinvokers, invocation);
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.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 (int i = 0; i < len; i++) {
//重试时，进行重新选择，避免重试时invoker列表已发生变化.
//注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
if (i > 0) {
checkWhetherDestroyed();
copyinvokers = list(invocation);
//重新检查一下
checkInvokers(copyinvokers, invocation);
}
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + invocation.getMethodName()
+ " in the service " + getInterface().getName()
+ " was successful by the provider " + invoker.getUrl().getAddress()
+ ", but there have been failed 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);
}
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 != null ? le.getCode() : 0, "Failed to invoke the method "
+ invocation.getMethodName() + " 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 != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}
- 检测copyinvokers合法性 (size > 0)
- 获取调用方法重试次数 len 默认为3次
- 声明 List<Invoker<T>> invoked 大小为invokers的大小 ， 存放失败调用的invoke对象
- 声明 Set<String> providers 记录调用失败的远程连接地址 大小默认为len 大小
- 针对每次循环调用，都会重新获取一遍invoker列表, 防止服务提供端发生改变。
- 调用select方法从invokers列表中 按照指定策略获取一个invoker对象。并在invoked中添加 invoker对象，并设置到本线程的调用上下文中。
- 进行服务调用(经过过滤器链)
- 调用失败则进入循环体，重新循环调用。
接下来看 6过程中 Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
/**
* 使用 loadbalance 选择 invoker.
*
* @param loadbalance Loadbalance 对象，提供负责均衡策略
* @param invocation Invocation 对象
* @param invokers 候选的 Invoker 集合
* @param selected 已选过的 Invoker 集合. 注意：输入保证不重复
* @return 最终的 Invoker 对象
* @throws RpcException RpcException 当发生 RpcException 时
*/
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
// 获得 sticky(粘滞)配置项，方法级
String methodName = invocation == null ? "" : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
//ignore overloaded method // 若 stickyInvoker 不存在于 invokers 中，说明不在候选中，需要置空，重新选择
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//ignore cucurrent problem 若开启粘滞连接的特性，且 stickyInvoker 不存在于 selected 中，则返回 stickyInvoker 这个 Invoker 对象
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
// 若开启排除非可用的 Invoker 的特性，则校验 stickyInvoker 是否可用。若可用，则进行返回
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
// 执行选择
Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
// 若开启粘滞连接的特性，记录最终选择的 Invoker 到 stickyInvoker
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}
该方法作用是获取 invokers 中指定的调用对象，调用过程如下
- 如果服务提供端列表为空，则直接返回null
- 获取调用方法名称 ， 并判断该方法是否支持粘滞 ，如果粘滞则默认从 stickyInvoker 中获取对象
- 如果stickyInvoker 不存在于invokers中说明该服务已经卸载， 则置空并重新选取并调用。
- 调用 doSelect 执行选择
粘滞连接用于有状态服务，尽可能让客户端总是向同一提供者发起调用，除非该提供者挂了，再连另一台。粘滞连接将自动开启延迟连接，以减少长连接数。
<dubbo:protocol name="dubbo" sticky="true" />
接下来看4 中的 doselect 方法
private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
// 一、如果只有一个 Invoker ，直接选择
if (invokers.size() == 1)
return invokers.get(0);
// 二、如果只有两个invoker，退化成轮循
// If we only have two invokers, use round-robin instead.
if (invokers.size() == 2 && selected != null && selected.size() > 0) {
return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
}
// 三、使用 Loadbalance ，选择一个 Invoker 对象。
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
// If the `invoker` is in the `selected` or invoker is unavailable && availablecheck is true, reselect.
// 如果 selected中包含（优先判断） 或者 不可用&&availablecheck=true 则重试.
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
//四、重选一个 Invoker 对象
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
invoker = rinvoker;
} else {
// Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
// 五、看下第一次选的位置，如果不是最后，选+1位置.
int index = invokers.indexOf(invoker);
try {
// Avoid collision
//最后在避免碰撞
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invoker;
} catch (Exception e) {
logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
}
}
} catch (Throwable t) {
logger.error("clustor relselect fail reason is :" + t.getMessage() + " if can not slove ,you can set cluster.availablecheck=false in url", t);
}
}
return invoker;
}
方法的作用是 ： 从候选的 Invoker 集合，选择一个最终调用的 Invoker 对象。我们看下它的调用过程
- 如果候选集合中只有一个invoker ， 则直接返回该invoker
- 如果有两个invoker并且不是第一次调用，则退化为轮询。(round-robin)
- 否则利用loadbalance.select 获取一个invoker对象。
- 如果selected中包含(优先判断) 或者 不可用&&检测 则重新选择invoker reselect(loadbalance, invocation, invokers, selected, availablecheck);
- 如果重选为空，获取第一次选择下一个或当前选择。
private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck)
重选逻辑 ： 先从非selected的列表中选择，没有在从selected列表中选择.
最后进行Invoker过滤器链调用进行远程方法调用。
loadbalance 结构图如下
这几种扩展种类：
1.Random LoadBalance
随机，按权重设置随机概率
在一个截面上碰撞的概率高，但调用量越大，分布越均匀，而且按概率使用权重后也比较均匀，有利于动态调整提供者权重
2.RoundRobin LoadBalance
轮询，按公约后的权重设置轮询比例
存在慢的提供者累计请求的问题。
3.LeastActive LoadBalance
最少活跃调用数，相同活跃数的随机，活跃数指调用前后计数差
使慢的提供者受到更少的请求，因为越慢的提供者的调用前后计数差会越大。
4.ConsistentHash LoadBalance
一致性Hash,相同参数请求总数发到同一提供者。
当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其他提供者，不会引起剧烈变动。
配置参考如下：
<dubbo:service interface="..."loadbalance="roundrobin"/>
<dubbo:reference interface="..."loadbalance="roundrobin"/>
<dubbo:service interface="...">
<dubbo:method name="..."loadbalance="roundrobin"/>
</dubbo:service>
<dubbo:reference interface="...">
<dubbo:method name="..."loadbalance="roundrobin"/>
</dubbo:reference>
<dubbo:reference interface="..." loadbalance="consistenthash" />
缺省只对第一个参数Hash，如果要修改，请配置
<dubbo:parameter key="hash.arguments" value="0,1" />
缺省用160份虚拟节点，如果要修改，请配置
<dubbo:parameter key="hash.nodes" value="320" />
正文到此结束
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
