学习笔记---DUBBO的服务调用（消费端）
dubbo的服务调用的逻辑可以拆分为两部分来处理：消费端的调用；提供端的响应；
消费端的调用：
代码分析：
从： String hello = demoService.sayHello(“world”) 开始，由上次dubbod的服务引入分析可以知道，这个：demoService就是Invoker包装类的代理对象，下一步执行的逻辑：
@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();
这里进行了aop的拦截过滤，直接看：
invoker.invoke(new RpcInvocation(method, args)).recreate();
这里进行了：RpcInvocation对象的封装，可以看到这里封装了请求的方法，参数实体关键元素，也就是说：RpcInvocation代表了dubbo消费端的一个请求实体；接着跑到被代理的对象：MockClusterInvoker的invoke方法：
@Override
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")) {
//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
result = doMockInvoke(invocation, null);
} else {
//fail-mock
try {
result = this.invoker.invoke(invocation);
} catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
if (logger.isWarnEnabled()) {
logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
}
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
可以看到这里进行了是否降级的处理，我们可以看到如果跑降级的处理的有两个渠道，一个是正常调用失败后降级，一个是直接降级逻辑；直接进入到飞降级逻辑：AbstractClusterInvoker的invoke方法，这里进行了RpcContext的参数的绑定，这个可以在消费端调用的时候一直存储着，之前的公司就是利用这个以及责任链来开发了分布式处理框架；接着根据配置的来找到对应的负载均衡策略，默认是随机选择策略：RandomLoadBalance，接着跑到doInvoke里面，个人分析这里进行了负载均衡以及根据配置来如何重试，比如：如果选择快速失败，则如果调用失败则直接失败，如果失败重试的话，则失败后用集合来过滤失败的调用invoker,重新选择invoker,在配置的次数允许下重新调用：
@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);
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++) {
//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 {
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);
}
其中循环体中的：
Invoker invoker = select(loadbalance, invocation, copyinvokers, invoked) 就是用来调用负载均衡的方法，并且避免上次调用的invoker,
接着经过：
InvokerWrapper-》ProtocolFilterWrapper 后开始进入消费端的filter责任链：ConsumerContextFilter-》FutureFilter-》MonitorFilter，其中ConsumerContextFilter是绑定IP以及port基本信息到RpcContext上下文上，FutureFilter是返回结果的回调，MonitorFilter是收集统计调用结果；接着，我们直接到：DubboInvoker的invoke方法，这里就是非常关键的逻辑了：
@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);
logger.info("ResponseFuture responseFuture = currentClient.request(inv, timeout);");
ResponseFuture responseFuture = currentClient.request(inv, timeout);
logger.info(" Object o = responseFuture.get();");
Object o = responseFuture.get();
return (Result) o;
}
} 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);
}
}
关键的地方：
ResponseFuture responseFuture = currentClient.request(inv, timeout);
最后跑到：
HeaderExchangeChannel，分装了Qequest请求实体，进行IO发送了；
然后回到：
DubboInvoker的：Object o = responseFuture.get() 此处开始进行异步转同步，进入到：DefaultFuture
@Override
public Object get(int timeout) throws RemotingException {
if (timeout <= 0) {
timeout = Constants.DEFAULT_TIMEOUT;
}
if (!isDone()) {
long start = System.currentTimeMillis();
lock.lock();
try {
while (!isDone()) {
done.await(timeout, TimeUnit.MILLISECONDS);
if (isDone() || System.currentTimeMillis() - start > timeout) {
break;
}
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
lock.unlock();
}
if (!isDone()) {
throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
}
}
return returnFromResponse();
}
可以看到这里利用了AQS来处理，而且这个：lock对应了每一个ResponseFuture，意思就是锁的力度已经是精细化到对象级别，只会影响到这个对象，而每次执行调用都会新的ResponseFuture对象，这样就避免了不同的调用之间的阻塞影响，在后面的逻辑中， 加完这个AQS乐观锁以及awate等待配置的超时时间，等待isDone()，而isDone() 由：Response 这个属性是否为空来决定的，问题来了，如何确定这个Response 呢？可以看到这个：Response 是：volatile修饰的共享可变变量，并且还有：
private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>()； 并且这个key值是在：
HeaderExchangeChannel里面：生成Request对象时通过：
private static final AtomicLong INVOKE_ID = new AtomicLong(0);
这个：INVOKE_ID 这个原子递增而得，意味着请求Request与DefaultFuture是可以做到一一对应的，读到这里，可以大胆的猜想，这个全局静态变量，是不是存在某逻辑来从FUTURES 这全局静态常量的Map中，获取对应的DefaultFuture对象，并且将结果赋值到：Response 这个属性，从而完成结果的返回？？尝试着查询用这个：getFuture方法的逻辑，发现：
发现这个方法在处理IO编码解码的方法是调用，断点到：ExchangeCodec
查看调用链后，看到这里：
解码得到结果后通过：
Response res = new Response(id) 以及 data = result 来完成返回结果的封装，可以肯定在调用的请实体信息中必定存在请求的唯一id，那么在什么时候将这个：Response 如何赋值？可以猜想，netty里面处理发送以及返回的结果实在：ChannelHandler 的，查找：
DefaultFuture.received(channel, response)，
public static void received(Channel channel, Response response) {
try {
DefaultFuture future = FUTURES.remove(response.getId());
if (future != null) {
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response " + response
+ (channel == null ? "" : ", channel: " + channel.getLocalAddress()
+ " -> " + channel.getRemoteAddress()));
}
} finally {
CHANNELS.remove(response.getId());
}
}
发现断点：在此完成了：Response的赋值，问题来了，如果返回的结果在调用超时时间之前完成，前面已经 awate 等待了，如何唤醒呢？
private void doReceived(Response res) {
lock.lock();
try {
response = res;
if (done != null) {
done.signal();
}
} finally {
lock.unlock();
}
if (callback != null) {
invokeCallback(callback);
}
}
代码很明显了；
那么如果超时的话，如何处理 ,在：FailoverClusterInvoker 重试后如果依旧的话，则：
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);
抛给调用者；
回到正常调用的情况，我们关注到责任链上面去：
ConsumerContextFilter-》FutureFilter-》MonitorFilter，
首先：MonitorFilter 并没有处理回来的结果，只是处理消费端调用而已，用来给管理台用的，
接着：FutureFilter 处理返回，这个根据配置分情况处理异常以及非异常的结果返回，
最后：ConsumerContextFilter ，清除上下文的 ：clearAttachments ；
最后就一路返回给调用者结果；
下一篇： 编写程序，求1-100之间所有奇数的和。
推荐阅读
-
学习笔记---DUBBO的服务调用（消费端）
-
Javaweb学习笔记之Servlet(二)：HttpServletRequest获取客户端发送给服务器的数据
-
ROS学习笔记：用C++编写一个简单的服务和客户端
-
荐 【dubbo源码解析】--- dubbo的服务暴露+服务消费（RPC调用）底层原理深入探析
-
多线程编程学习笔记——编写一个异步的HTTP服务器和客户端
-
Socket学习笔记(一):python通过socket实现客户端到服务器端的文件传输
-
ROS学习笔记：用C++编写一个简单的服务和客户端
-
荐 【dubbo源码解析】--- dubbo的服务暴露+服务消费（RPC调用）底层原理深入探析
-
多线程编程学习笔记——编写一个异步的HTTP服务器和客户端