源码分析Dubbo异步调用与事件回调机制

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 源码分析Dubbo异步调用与事件回调机制

微信公众号:**[中间件兴趣圈]** 作者简介:《RocketMQ技术内幕》作者

本文将详细分析Dubbo服务异步调用与事件回调机制。

异步调用与事件回调机制

TTY异步回调

事件回调

异步调用与事件回调机制

在Dubbo中,引入特定的过滤器FutureFilter来处理异步调用相关的逻辑,其定义如下:


1@Activate(group = Constants.CONSUMER)
2public class FutureFilter implements Filter {
3}

group=CONSUMER说明该过滤器属于消费端过滤器。
接下来从从invoke方法详细分析其实现逻辑。


 1public Result invoke(final Invoker? invoker, final Invocation invocation) throws RpcException {
 2        final boolean isAsync = RpcUtils.isAsync(invoker.getUrl(), invocation);     // @1
 3        fireInvokeCallback(invoker, invocation);                                                     // @2
 4        // need to configure if there's return value before the invocation in order to help invoker to judge if it's
 5        // necessary to return future.
 6        Result result = invoker.invoke(invocation);                                                 // @3
 7        if (isAsync) {
 8            asyncCallback(invoker, invocation);                                                       // @4
 9        } else {
10            syncCallback(invoker, invocation, result);                                              // @5
11        }
12        return result;
13}

代码@1:首先从URL中获取是否是异步调用标志,其配置属性为 dubbo:service async=””/获取其子标签 dubbo:method async=””/。

代码@2:同步调用oninvoke事件,执行invoke方法之前的事件。

代码@3:继续沿着调用链调用,最终会到具体的协议Invoker,例如DubboInvoker,发生具体的服务调用,跟踪一下同步、异步调用的实现细节。

代码@4:如果调用方式是异步模式,则异步调用onreturn或onthrow事件。

代码@5:如果调用方式是同步模式,则同步调用onreturn或onthrow事件。

FutureFilter#fireInvokeCallback


 1private void fireInvokeCallback(final Invoker? invoker, final Invocation invocation) {
 2        final Method onInvokeMethod = (Method) StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), 
 3Constants.ON_INVOKE_METHOD_KEY));      // @1
 4        final Object onInvokeInst = StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), 
 5Constants.ON_INVOKE_INSTANCE_KEY));   // @2
 6        if (onInvokeMethod == null && onInvokeInst == null) {    // @3
 7            return;
 8        }
 9        if (onInvokeMethod == null || onInvokeInst == null) {    // @4
10            throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onreturn callback config , but no such " + (onInvokeMethod == null ? "method" : 
11                         "instance") + " found. url:" + invoker.getUrl());
12        }
13        if (!onInvokeMethod.isAccessible()) {
14            onInvokeMethod.setAccessible(true);
15        }
16        Object[] params = invocation.getArguments();
17        try {
18            onInvokeMethod.invoke(onInvokeInst, params);        // @5
19        } catch (InvocationTargetException e) {
20            fireThrowCallback(invoker, invocation, e.getTargetException());    // @6
21        } catch (Throwable e) {
22            fireThrowCallback(invoker, invocation, e);                         // @7
23        }
24    }

代码@1:StaticContext.getSystemContext()中根据key:serviceKey + “.” + method + “.” + “oninvoke.method” 获取配置的oninvoke.method方法名。其中serviceKey为[group]/interface:[version],其中group与version可能为空,忽略。

代码@2:同样根据key:serviceKey + “.” + method + “.” + “oninvoke.instance” 从StaticContext.getSystemContext()获取oninvoke.method方法所在的实例名对象,也就是说该调用哪个对象的oninvoke.method指定的方法。这里就有一个疑问,这些数据是在什么时候存入StaticContext中的呢?下文会详细分析。

代码@3、@4:主要检测 dubbo:method oninvoke=””/配置的正确性,其正确的配置方式如下:“实例名.方法名”,例如:

代码@5:根据发射机制,调用oninvoke中指定的实例的指定方法,注意,这里传入的参数为调用远程RPC服务的参数。

注意:从这里可以看出,如果要实现事件通知,也即在调用远程RPC服务之前,之后、抛出异常时执行回调函数,该回调事件的方法的参数列表需要与被调用服务的参数列表一致。

代码@6、@7,如果在执行调用前方法(oninvoke)事件方法失败,则会同步调用onthrow中定义的方法(如有定义)。关于dubbo:method oninvoke属性的解析以及在什么时候会向StaticContext.getSystemContext()中添加信息,将在下文统一介绍。

源码分析DubboInvoker关于同步异步调用处理

在上文提到FutureFilter#invoke中的第三步调用invoker.invoker方法时,我们应该会有兴趣了解一下真实的invoker是如何处理同步、异步请求的。
我们以dubbo协议DubboInvoker来重点分析一下其实现原理:

DubboInvoker#doInvoke


 1try {
 2            boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);            // @1
 3            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
 4            int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
 5            if (isOneway) {
 6                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
 7                currentClient.send(inv, isSent);                                             // @2
 8                RpcContext.getContext().setFuture(null);
 9                return new RpcResult();
10            } else if (isAsync) {
11                ResponseFuture future = currentClient.request(inv, timeout);                 // @3
12                RpcContext.getContext().setFuture(new FutureAdapterObject(future));
13                return new RpcResult();
14            } else {
15                RpcContext.getContext().setFuture(null);     // @4
16                return (Result) currentClient.request(inv, timeout).get();
17            }
18        } catch (TimeoutException e) {
19            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
20        } catch (RemotingException e) {
21            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
22        }

代码@1:首先获取async属性,如果为true表示异步请求,如果配置了return=”false”表示调用模式为oneway,只发调用,不关注其调用结果。

代码@2:处理oneway的情况。如果设置了sent=true,表示等待网络数据发出才返回,如果sent=false,只是将待发送数据发到IO写缓存区就返回。

代码@3:处理异步的情况,代码@4处理同步调用的情况,细看其实都是通过调用网络客户端client的request,最终调用HeaderExchangeChannel#request方法:

这里是通过Future模式来实现异步调用的,同步调用也是通过异步调用来实现,只是同步调用发起后,直接调用future#get的方法来同步等待结果的返回,而异步调用只返回Future Response,在用户需要关心其结果时才调用get方法。

asyncCallback与syncCallback

前面介绍了方法执行之前oninvoker事件的调用分析,接下来分析RPC服务调用完成后,onreturn和onthrow方法的调用逻辑。

异步回调与同步回调的区别就是调用onreturn(fireReturnCallback)和onthrow(fireThrowCallback)调用的地方不同,如果是同步调用,也就是在完成RPC服务调用后,立即调用相关的回调方法,如果是异步调用的话,RPC服务完成后,通过Future模式异步执行。其实关于onreturn、onthrow属性的解析,执行与oninvoker属性的解析完全一样,再这里也就不重复介绍了。

广告:作者新书《RocketMQ技术内幕》已上市

源码分析Dubbo异步调用与事件回调机制

《RocketMQ技术内幕》已出版上市,目前可在主流购物平台(京东、天猫等)购买,本书从源码角度深度分析了RocketMQ NameServer、消息发送、消息存储、消息消费、消息过滤、主从同步HA、事务消息;在实战篇重点介绍了RocketMQ运维管理界面与当前支持的39个运维命令;并在附录部分罗列了RocketMQ几乎所有的配置参数。本书得到了RocketMQ创始人、阿里巴巴Messaging开源技术负责人、Linux OpenMessaging 主席的高度认可并作序推荐。目前是国内第一本成体系剖析RocketMQ的书籍。
新书7折优惠!7折优惠!7折优惠!

更多文章请关注微信公众号:

源码分析Dubbo异步调用与事件回调机制

推荐关注微信公众号:RocketMQ官方微信公众号

源码分析Dubbo异步调用与事件回调机制

原文始发于微信公众号(中间件兴趣圈):

本人花费半年的时间总结的《Java面试指南》已拿腾讯等大厂offer,已开源在github ,欢迎star!

转载声明:转载请注明出处,本技术博客是本人原创文章

本文GitHub https://github.com/OUYANGSIHAI/JavaInterview 已收录,这是我花了6个月总结的一线大厂Java面试总结,本人已拿大厂offer,欢迎star

原文链接:blog.ouyangsihai.cn >> 源码分析Dubbo异步调用与事件回调机制


 上一篇
源码分析Dubbo服务调用日志(accesslog参数)实现原理 源码分析Dubbo服务调用日志(accesslog参数)实现原理
微信公众号:**[中间件兴趣圈]** 作者简介:《RocketMQ技术内幕》 谈到服务调用日志,大家恐怕第一想到就是如果开启了这个参数,会影响性能。那真实的情况是怎么样了?性能损耗到底有多大呢?在实践中我们如何使用该功能呢?本文将详
2021-04-05
下一篇 
源码分析Dubbo监控中心实现原理 源码分析Dubbo监控中心实现原理
微信公众号:**[中间件兴趣圈]** 作者简介:《RocketMQ技术内幕》作者 Dubbo监控的基本实现原理就是在服务调用时收集服务调用并发度、服务响应时间,然后以一定频率向监控中心汇报统计数据。 MonitorFilter过滤器
2021-04-05