dubbo 服务调用源码分析
下面是调用方法栈和核心代码分析
InvokerInvocationHandler.invoke(Object, Method, Object[]) line: 38
com.alibaba.dubbo.rpc.cluster.support.wrapper.MockClusterInvoker.invoke(Invocation)
com.alibaba.dubbo.rpc.cluster.support.AbstractClusterInvoker.invoke(Invocation)
public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
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);
}
//幂等操作:异步操作默认添加invocation id
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
com.alibaba.dubbo.rpc.cluster.support.FailoverClusterInvoker.doInvoke(Invocation, List<Invoker<T>>, LoadBalance)
@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);
//获取重试次数,dubbo有容错机制
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) {
checkWheatherDestoried();
copyinvokers = list(invocation);
//重新检查一下
checkInvokers(copyinvokers, invocation);
}
//使用loadbalance选择invoker.
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);
}
com.alibaba.dubbo.rpc.protocol.InvokerWrapper.invoke(Invocation)
com.alibaba.dubbo.rpc.filter.ConsumerContextFilter.invoke(Invoker<?>, Invocation)
com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter.invoke(Invoker<?>, Invocation)
com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker.doInvoke(Invocation)
@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);
}
}
com.alibaba.dubbo.remoting.transport.AbstractClient.send(Object, boolean)
public void send(Object message, boolean sent) throws RemotingException {
if (send_reconnect && !isConnected()){
connect();
}
//使用com.alibaba.dubbo.remoting.transport.netty.NettyChannel传递数据
Channel channel = getChannel();
//TODO getChannel返回的状态是否包含null需要改进
if (channel == null || ! channel.isConnected()) {
throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
}
channel.send(message, sent);
}
//最后一下
com.alibaba.dubbo.remoting.transport.netty.NettyChannel.send(Object, boolean)
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
//调用netty org.jboss.netty.channel.Channel.write(Object)
ChannelFuture future = channel.write(message);
if (sent) {
timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
//如果是异步调用
success = future.await(timeout);
}
Throwable cause = future.getCause();
if (cause != null) {
throw cause;
}
} catch (Throwable e) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
}
if(! success) {
throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
+ "in timeout(" + timeout + "ms) limit");
}
}
dubbo 服务调用源码分析相关推荐
- Dubbo 服务订阅源码分析
Dubbo 服务引用的时机有两个: 第一个是在 Spring 容器调用 ReferenceBean 的 afterPropertiesSet 方法时引用服务 第二个是在 ReferenceBean 对 ...
- Dubbo服务注册源码分析
本代码版本基于Dubbo2.7.8版本进行源码分析 注册概览 扫描所有@DubboService注解, 加载配置文件, 装载注解中的所有属性, 把每个服务都封装成一个ServiceBean, 注入到S ...
- 开发 web 程序服务 之 源码分析
文章目录 开发 web 程序服务 之 源码分析 前言 http 包源码 路由部分 监听和服务部分 mux 库源码 源码分析 创建路由 路由匹配 总结 开发 web 程序服务 之 源码分析 前言 本文的 ...
- 分布式定时任务—xxl-job学习(四)——调度中心web页面端api调用源码分析
分布式定时任务-xxl-job学习(四)--调度中心web页面端api调用源码分析 前言 一.controller目录下非controller类 1.1 PermissionLimit自定义注解 1. ...
- Eureka服务注册源码分析
本文来说下Eureka服务注册源码 文章目录 Eureka-Client注册服务 啥时候会注册 定时器注册 自动注册 DiscoveryClient.register() Eureka-Server接 ...
- 深入 Eureka 服务注册 源码分析(二)
说一下自己对Eureka注册的理解 Eureka注册的流程是 1.客户端在初始化或者在注册信息发生变化的时候,发送注册信息.其实在初始化的时候也是通过改变注册信息的. 客户端会开启一个定时任务,每40 ...
- dubbo的Extension源码分析
2019独角兽企业重金招聘Python工程师标准>>> 我们基于ExtensionLoader.getExtensionLoader().getAdaptiveExtension() ...
- Android服务函数远程调用源码分析
在Android服务查询完整过程源码分析中介绍了客户进程向ServiceManager进程查询服务的完整过程,ServiceManager进程根据服务名称在自身维护的服务链表中查找ServiceMan ...
- 阿里巴巴商城源码JAVA_阿里巴巴Dubbo实现的源码分析
1. Dubbo概述 Dubbo是阿里巴巴开源出来的一个分布式服务框架,致力于提供高性能和透明化的RPC远程服务调用方案,以及作为SOA服务治理的方案.它的核心功能包括: #remoting ...
最新文章
- 英特尔将进行重大业务重组
- sql server 连接工具_SQL on file 工具
- 最小公倍数(Least_Common_Multiple)
- java ios压缩_iOS与Java服务器GZip压缩问题【转】
- winform插入时间类型数据到oracle数据库,winform操作访问Oracle 10g数据库,并自动填充到DataGridView...
- 前端开发攻城狮必须知道的开发环境和插件
- php 一键登录插件,FastAdmin一键管理插件
- Thinkphp3.2版本Controller和Action的访问方法
- 单链表的合并算法_图解算法:单链表两两反转 | 眼睛会了手就会系列
- [Python] L1-013. 计算阶乘和-PAT团体程序设计天梯赛GPLT
- 窗口操作-关闭,最小化
- 转:什么是Node.js?
- 分享240道有意思的逻辑思维题
- matlab堆积式玫瑰图,用SAS实现堆积式南丁格尔玫瑰图Nightingale Rose Diagram (上)...
- noob_臭代码-Java Noob的自白
- 离散数学 (上)小结
- 7周入门数据分析:(2)分析界的No.1——Excel
- html5波浪效果,html5 canvas粒子波浪动画特效
- 微信公众号平搜索排名,如何让公众号搜索排名靠前,公众号文章关键词排名规则
- java二维数组两种初始化方法
热门文章
- java总是标点符号报错_[javamail]AUTH LOGIN failed;Invalid username or password报错
- 【论文笔记】Towards Privacy-Preserving Affect Recognition: A Two-Level Deep Learning Architecture
- 2020年综合评价备考全知道(附31所综合评价院校名单)
- 【广东开放大学(广东理工职业学院)主办】第二届计算机图形学、人工智能与数据处理国际学术会议(ICCAID 2022)
- docker容器Linux环境下二维码图片中文字体乱码处理办法
- Java人才,请往这里看过来!
- 求1到n的素数个数C语言,求 1~n 之间素数的个数
- 我和世界杯的‘恩怨情仇’
- 【文献阅读】Silhouette based View embeddings for Gait Recognit
- 梦想实现_实现梦想的软件工程工作需要什么