从源码全面解析 dubbo 消费端服务调用的来龙去脉

news/2024/11/27 2:29:21/
  • 👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
  • 📕系列专栏：Java设计模式、Spring源码系列、Netty源码系列、Kafka源码系列、JUC源码系列、duubo源码系列
  • 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
  • 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
  • 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀

在这里插入图片描述

文章目录

    • 一、引言
    • 二、服务调用流程
      • 1、消费端
        • 1.1 动态代理的回调
        • 1.2 过滤器
        • 1.3 路由逻辑
        • 1.4 重试次数
        • 1.5 负载均衡
          • 1.4.1 自定义负载均衡
        • 1.6 调用服务
          • 1.6.1 配置 RPCinvocation
          • 1.6.2 调用 RPC 同步返回结果
          • 1.6.3 等待返回结果
    • 三、流程
    • 四、总结

一、引言

对于 Java 开发者而言，关于 dubbo ，我们一般当做黑盒来进行使用，不需要去打开这个黑盒。

但随着目前程序员行业的发展，我们有必要打开这个黑盒，去探索其中的奥妙。

本期 dubbo 源码解析系列文章，将带你领略 dubbo 源码的奥秘

本期源码文章吸收了之前 Spring、Kakfa、JUC源码文章的教训，将不再一行一行的带大家分析源码，我们将一些不重要的部分当做黑盒处理，以便我们更快、更有效的阅读源码。

虽然现在是互联网寒冬，但乾坤未定，你我皆是黑马！

废话不多说，发车！

二、服务调用流程

1、消费端

上一篇文章，讲解了我们的消费端如何订阅我们服务端注册到 Zookeeper 的服务接口：从源码全面解析 dubbo 服务订阅的来龙去脉

既然消费端已经知道了我们的服务信息，那么下一步就要开始正式调用了

我们先从消费端聊聊服务调用的流程

1.1 动态代理的回调

我们聊到消费端订阅服务时，最终创建的代码如下：

public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}

相信看过 动态代理 的小伙伴应该知道，当我们调用 代理 的接口时，实际上走的是 InvokerInvocationHandler 该类的 invoke 方法

public Object invoke(Object proxy, Method method, Object[] args){// 获取方法名=getUserByIdString methodName = method.getName();// 获取参数Class<?>[] parameterTypes = method.getParameterTypes();// 组装成 RpcInvocation 进行调用RpcInvocation rpcInvocation = new RpcInvocation(serviceModel, method.getName(), invoker.getInterface().getName(), protocolServiceKey, method.getParameterTypes(), args);// 执行调用方法return InvocationUtil.invoke(invoker, rpcInvocation);
}

这里我们重点介绍下 RpcInvocation 的几个参数：

  • serviceModel(Consumer)：决定了服务的调用方式，包括使用哪种协议、注册中心获取服务列表、负载均衡和容错策略等。
  • method.getName：getUserById
  • invoker.getInterface().getName：com.common.service.IUserService
  • protocolServiceKey：com.common.service.IUserService:dubbo
  • method.getParameterTypes：方法的入参类型（Long）
  • args：方法的入参值（2）

我们继续往下看 InvocationUtil.invoke 做了什么

public static Object invoke(Invoker<?> invoker, RpcInvocation rpcInvocation) throws Throwable {URL url = invoker.getUrl();String serviceKey = url.getServiceKey();rpcInvocation.setTargetServiceUniqueName(serviceKey);return invoker.invoke(rpcInvocation).recreate();
}// 判断当前的是应用注册还是接口注册
public Result invoke(Invocation invocation) throws RpcException {if (currentAvailableInvoker != null) {if (step == APPLICATION_FIRST) {if (promotion < 100 && ThreadLocalRandom.current().nextDouble(100) > promotion) {return invoker.invoke(invocation);}return decideInvoker().invoke(invocation);}return currentAvailableInvoker.invoke(invocation);}
}

我们继续往下追源码

1.2 过滤器

// 过滤器责任链模式
// 依次遍历，执行顺序：
public interface FilterChainBuilder {public Result invoke(Invocation invocation) throws RpcException {Result asyncResult;InvocationProfilerUtils.enterDetailProfiler(invocation, () -> "Filter " + filter.getClass().getName() + " invoke.");asyncResult = filter.invoke(nextNode, invocation);}
}

这里会依次遍历所有的 filter：

  • ConsumerContextFilter：将消费者端的信息（远程地址、应用名、服务名）传递给服务提供者端
  • ConsumerClassLoaderFilter：将消费者端的ClassLoader传递给服务提供者端，以便服务提供者端可以在调用时使用相同的ClassLoader加载类。
  • FutureFilter：异步调用
  • MonitorFilter：统计服务调用信息（调用次数、平均响应时间、失败次数）
  • RouterSnapshotFilter：动态路由，它可以根据路由规则选择服务提供者，并缓存路由结果，以提高性能。

具体每个过滤器怎么实现的，这里就不展开讲了，后面有机会单独出一章

1.3 路由逻辑

当我们的责任链完成之后，下一步会经过我们的 路由 逻辑

public Result invoke(final Invocation invocation) throws RpcException {// List<Invoker<T>> invokers = list(invocation);InvocationProfilerUtils.releaseDetailProfiler(invocation);LoadBalance loadbalance = initLoadBalance(invokers, invocation);RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);return doInvoke(invocation, invokers, loadbalance);      
}

其中 List<Invoker<T>> invokers = list(invocation) 这里就是我们的路由逻辑：

List<Invoker<T>> invokers = list(invocation);public List<Invoker<T>> list(Invocation invocation) throws RpcException {List<Invoker<T>> routedResult = doList(availableInvokers, invocation);
}public List<Invoker<T>> doList(BitList<Invoker<T>> invokers, Invocation invocation) {// 这里就是我们的路由策略！！！List<Invoker<T>> result = routerChain.route(getConsumerUrl(), invokers, invocation);return result == null ? BitList.emptyList() : result;
}

这里的路由策略比较多，我举两个比较经典的：

  • simpleRoute（简单路由策略）：默认的路由策略

  • routeAndPrint（自定义路由策略）：我们可以自定义其路由逻辑

而对于整体路由的流程：

  • 获取可用的服务提供者列表
  • 过滤出符合条件的服务提供者
  • 对过滤后的服务提供者列表进行排序
  • 得到符合规定的服务提供者信息

到这里，我们路由会把符合要求的 服务端 给筛选出来，接下来就进入我们的负载均衡环节了

1.4 重试次数

这里我们设置 retries 为 5

@DubboReference(protocol = "dubbo", timeout = 100, retries = 5)
private IUserService iUserService;

我们看下源码里面有几次调用：根据源码来看，我们会有 5+1 次调用

int len = calculateInvokeTimes(methodName);
for (int i = 0; i < len; i++) {}private int calculateInvokeTimes(String methodName) {// 获取当前的重试次数+1int len = getUrl().getMethodParameter(methodName, RETRIES_KEY, DEFAULT_RETRIES) + 1;RpcContext rpcContext = RpcContext.getClientAttachment();Object retry = rpcContext.getObjectAttachment(RETRIES_KEY);if (retry instanceof Number) {len = ((Number) retry).intValue() + 1;rpcContext.removeAttachment(RETRIES_KEY);}if (len <= 0) {len = 1;}return len;
}

我们直接 Debug 一下看看：

image-20230612235945732

1.5 负载均衡

这一行 LoadBalance loadbalance = initLoadBalance(invokers, invocation) 得到我们的负载均衡策略，默认情况下如下：

image-20230612223930762

我们可以看到，默认情况下是 RandomLoadBalance 随机负载。

我们继续往下追源码：

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) {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,防止服务端发生了变化if (i > 0) {checkWhetherDestroyed();copyInvokers = list(invocation);// 再次校验checkInvokers(copyInvokers, invocation);}// 负载均衡逻辑！！！Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);invoked.add(invoker);RpcContext.getServiceContext().setInvokers((List) invoked);boolean success = false;try {Result result = invokeWithContext(invoker, invocation);success = true;return result;} }
}

这里我简单将下负载均衡的逻辑：

Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected){// 如果只有一个服务端，那还负载均衡个屁// 直接校验下OK不OK直接返回就好if (invokers.size() == 1) {Invoker<T> tInvoker = invokers.get(0);checkShouldInvalidateInvoker(tInvoker);return tInvoker;}// 如果多个服务端,需要执行负载均衡算法Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);return invoker;
}

Dubbo 里面的负载均衡算法如下：

image-20230612225319739

这里也就不一介绍了，正常情况下，我们采用的都是 RandomLoadBalance 负载均衡

当然这里博主介绍另外一个写法，也是我们业务中使用的

1.4.1 自定义负载均衡

上面我们看到，通过 LoadBalance loadbalance = initLoadBalance(invokers, invocation) ，我们可以得到一个负载均衡的实现类

在我们的生产场景中，不同的集群上含有不同的合作方，我们需要根据合作方去分发不同集群的调用

这个时候，我们可以重写我们的 LoadBalance ，在里面重写我们 doSelect 的逻辑，而这里的 集群A 也就是我们的 group

image-20230612232552377

1.6 调用服务

当我们完成下面的流程：过滤器 —> 路由 —> 重试 —> 负载均衡，就到了下面这行：

Result result = invokeWithContext(invoker, invocation)

我们继续往下追：

public Result invoke(Invocation invocation) throws RpcException {try {// 加读写锁lock.readLock().lock();return invoker.invoke(invocation);} finally {lock.readLock().unlock();}
}

我们直接追到 AbstractInvoker 的 invoke 方法

public Result invoke(Invocation inv) throws RpcException {RpcInvocation invocation = (RpcInvocation) inv;// 配置RPCinvocationprepareInvocation(invocation);// 调用RPC同时同步返回结果AsyncRpcResult asyncResult = doInvokeAndReturn(invocation);// 等待返回结果waitForResultIfSync(asyncResult, invocation);return asyncResult;
}

我们可以看到，对于调用服务来说，一共分为一下三步：

  • 配置 RPCinvocation
  • 调用 RPC 同步返回结果
  • 等待返回结果
1.6.1 配置 RPCinvocation

这里主要将 Invocation 转变成 RPCInvocation

  • 设置 RpcInvocation 的 Invoker 属性，指明该调用是由哪个 Invoker 发起的
  • 当前线程的一些状态信息
  • 同步调用、异步调用
  • 异步调用生成一个唯一的调用 ID
  • 选择序列化的类型
private void prepareInvocation(RpcInvocation inv) {// 设置 RpcInvocation 的 Invoker 属性，指明该调用是由哪个 Invoker 发起的inv.setInvoker(this);// 当前线程的一些状态信息addInvocationAttachments(inv);// 同步调用、异步调用inv.setInvokeMode(RpcUtils.getInvokeMode(url, inv));// 异步调用生成一个唯一的调用 IDRpcUtils.attachInvocationIdIfAsync(getUrl(), inv);// 选择序列化的类型Byte serializationId = CodecSupport.getIDByName(getUrl().getParameter(SERIALIZATION_KEY, DefaultSerializationSelector.getDefaultRemotingSerialization()));if (serializationId != null) {inv.put(SERIALIZATION_ID_KEY, serializationId);}
}
1.6.2 调用 RPC 同步返回结果
private AsyncRpcResult doInvokeAndReturn(RpcInvocation invocation) {asyncResult = (AsyncRpcResult) doInvoke(invocation);
}protected Result doInvoke(final Invocation invocation){// 获取超时时间int timeout = RpcUtils.calculateTimeout(getUrl(), invocation, methodName, DEFAULT_TIMEOUT);// 设置超时时间invocation.setAttachment(TIMEOUT_KEY, String.valueOf(timeout));// 从dubbo线程池中拿出一个线程ExecutorService executor = getCallbackExecutor(getUrl(), inv);// request：进行调用CompletableFuture<AppResponse> appResponseFuture = currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);FutureContext.getContext().setCompatibleFuture(appResponseFuture);AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);result.setExecutor(executor);return result;
}

这里的 currentClient.request 进行请求的发送：

public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor){return client.request(request, timeout, executor);
}public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor){Request req = new Request();req.setVersion(Version.getProtocolVersion());req.setTwoWay(true);req.setData(request);DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout, executor);channel.send(req);return future;
}

这里的 channel.send(req) 是 dubbo 自己包装的 channel，我们去看看其实现

当然，我们这里如果看过博主 Netty 源码文章的话，实际可以猜到，肯定是封装了 Netty 的 channel

public void send(Object message, boolean sent) throws RemotingException {// 校验当前的Channel是否关闭super.send(message, sent);boolean success = true;int timeout = 0;try {// channel 写入并刷新// channel:io.netty.channel.ChannelChannelFuture future = channel.writeAndFlush(message);if (sent) {// 等待超时的时间// 超过时间会报错timeout = getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);success = future.await(timeout);}// 这里如果报错了，就会走重试的逻辑Throwable cause = future.cause();}
}
1.6.3 等待返回结果
waitForResultIfSync(asyncResult, invocation);private void waitForResultIfSync(AsyncRpcResult asyncResult, RpcInvocation invocation) {// 判断当前的调用是不是同步调用// 异步调用直接返回即可if (InvokeMode.SYNC != invocation.getInvokeMode()) {return;}// 获取超时时间 Object timeoutKey = invocation.getObjectAttachmentWithoutConvert(TIMEOUT_KEY);long timeout = RpcUtils.convertToNumber(timeoutKey, Integer.MAX_VALUE);// 等待timeout时间// 获取失败-直接抛出异常asyncResult.get(timeout, TimeUnit.MILLISECONDS);
}public Result get(long timeout, TimeUnit unit){// 获取响应返回的数据-等待timeout时间return responseFuture.get(timeout, unit);
}

如果没有异常，如下图所示：

image-20230618160337104

到这里我们的消费端调用服务的整个流程源码剖析就完毕了~

三、流程

高清图片可私聊博主

在这里插入图片描述

四、总结

鲁迅先生曾说：独行难，众行易，和志同道合的人一起进步。彼此毫无保留的分享经验，才是对抗互联网寒冬的最佳选择。

其实很多时候，并不是我们不够努力，很可能就是自己努力的方向不对，如果有一个人能稍微指点你一下，你真的可能会少走几年弯路。

如果你也对 后端架构和中间件源码 有兴趣，欢迎添加博主微信：hls1793929520，一起学习，一起成长

我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，喜欢后端架构和中间件源码。

我们下期再见。

我从清晨走过，也拥抱夜晚的星辰，人生没有捷径，你我皆平凡，你好，陌生人，一起共勉。

往期文章推荐：

  • 美团二面：聊聊ConcurrentHashMap的存储流程
  • 从源码全面解析Java 线程池的来龙去脉
  • 从源码全面解析LinkedBlockingQueue的来龙去脉
  • 从源码全面解析 ArrayBlockingQueue 的来龙去脉
  • 从源码全面解析ReentrantLock的来龙去脉
  • 阅读完synchronized和ReentrantLock的源码后，我竟发现其完全相似
  • 从源码全面解析 ThreadLocal 关键字的来龙去脉
  • 从源码全面解析 synchronized 关键字的来龙去脉
  • 阿里面试官让我讲讲volatile，我直接从HotSpot开始讲起，一套组合拳拿下面试

http://www.ppmy.cn/news/453438.html

相关文章

使用dropzone实现文件拖拽上传功能

使用dropzone实现文件拖拽上传功能

前端代码&#xff0c;关于dropzone的配置我写在注释里了 <html> <head><meta charset"utf-8" /><%include file"/webmana/meta.jsp" %><title></title><link rel"stylesheet" type"text/css"…
阅读更多...
sdcard里边的东西能删除吗_手机内存中有一个sdcard文件夹删除不了

sdcard里边的东西能删除吗_手机内存中有一个sdcard文件夹删除不了

问&#xff1a;我的g14手机里原本内存空间共一个多G呢&#xff0c;现在我打开一看还剩84m&#xff0c;我用拉峰文件管理看到手机内存中有一个sdcard文件夹&#xff0c;这个文件占了一个多G&#xff0c;而且删还删不掉&#xff0c;不知道是什么原因&#xff1f; 回答&#xff1a…
阅读更多...
Android中的清除数据和清除缓存有什么区别，它们分别清除了哪些文件

Android中的清除数据和清除缓存有什么区别，它们分别清除了哪些文件

清除数据和清除缓存的区别 反射调用系统隐藏接口&#xff0c;需要准备的东西&#xff1a; 一、在AndroidManifest.xml中的manifest标签下添加声明 android:sharedUserId"android.uid.system" 二、添加如下权限 <uses-permission android:name"android.pe…
阅读更多...
使用adb命令删除Android系统data目录下的文件及文件夹

使用adb命令删除Android系统data目录下的文件及文件夹

使用命令删除data目录下的文件夹&#xff0c;和删除一般文件不同&#xff0c; data目录下的文件需要777权限&#xff0c;所以需要的命令不同。 data目录&#xff1a; C:\Users\aw>adb shell rootandroid:/ # su su rootandroid:/ # cd data/data cd data/data rootandroid…
阅读更多...
使用adb命令删除Android系统data目录下文件及文件夹

使用adb命令删除Android系统data目录下文件及文件夹

使用命令删除data目录下的文件夹&#xff0c;和删除一般文件不同。 data目录下的文件需要777权限&#xff0c;所以需要的命令不同 data目录&#xff1a; C:\Users\aw>adb shell rootandroid:/ # su su rootandroid:/ # cd data/data cd data/data rootandroid:/data/data…
阅读更多...
adb清除缓存数据

adb清除缓存数据

ADB Shell 应用 查看目录结构 &#xff1a;adb shell ls 查看系统当前日期 &#xff1a;adb shell date 查看系统 CPU 使用情况 &#xff1a;adb shell cat /proc/cpuinfo 查看系统内存使用情况 &#xff1a;adb shell cat /proc/meminfo 显示所有应用 &#xff1a;adb shell p…
阅读更多...
文件管理Android可以删,文件管理里的Android系统文件可以删吗

文件管理Android可以删,文件管理里的Android系统文件可以删吗

Windows Registry Editor Version 5.00 [HKEY_CLASSES_ROOT\*\shell\runas] "获取系统权限" "NoWorkingDirectory""" [HKEY_CLASSES_ROOT\*\shell\runas\command] "cmd.exe /c takeown /f \"%1\" && icacls \"%1\&q…
阅读更多...
Android系统tencent文件夹下哪些文件可以删除

Android系统tencent文件夹下哪些文件可以删除

手机越用内存越少&#xff0c;特别是安卓手机。当安装了qq、微信、qq空间&#xff0c;太多垃圾文件占用内存&#xff0c;请问哪些可以删除呢&#xff1f;又如何删除&#xff1f; 1.微信自带清理功能。 点开微信--我--设置-通用-清理微信存储空间。右上角有个3个点的按钮可以清理…
阅读更多...
最新文章

Copyright @ 2022~2023