本文主要是介绍Dubbo源码分析----过滤器之FutureFilter,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
FutureFilter主要是用来处理事件通知的过滤器,这么讲可能不太清楚,先看下下面的demo
先定一个是事件通知的类:
public class Notify {public void oninvoke(String msg){System.out.println("oninvoke:" + msg);}public void onreturn(String msg) {System.out.println("onreturn:" + msg);}public void onthrow(Throwable e) {System.out.println("onthrow:" + e);}
}
然后xml配置如下:
<bean id="notify" class="com.alibaba.dubbo.demo.consumer.Notify"/><dubbo:reference id="demoService" interface="com.alibaba.dubbo.demo.DemoService" check="false" timeout="5000"><dubbo:method name="sayHello" onreturn="notify.onreturn"/></dubbo:reference>
注意dubbo:method的配置,有个onreturn属性,意思是在return结果的时候调用Notify的onreturn方法,onthrow和oninvoke同理,实现这个处理的逻辑就在FutureFilter中,看下实现
@Activate(group = Constants.CONSUMER)
public class FutureFilter implements Filter {
//....public Result invoke(final Invoker<?> invoker, final Invocation invocation) throws RpcException {final boolean isAsync = RpcUtils.isAsync(invoker.getUrl(), invocation);// 判断dubbo:method中async属性的值,即是否异步fireInvokeCallback(invoker, invocation);// oninvoke方法的处理//需要在调用前配置好是否有返回值,已供invoker判断是否需要返回future.Result result = invoker.invoke(invocation);if (isAsync) {asyncCallback(invoker, invocation);// 异步回调oninvoke和onthrow} else {syncCallback(invoker, invocation, result);// oninvoke和onthrow的处理}return result;}
//....
}
fireInvokeCallback方法如下:
private void fireInvokeCallback(final Invoker<?> invoker, final Invocation invocation) {//onInvokeMethod 即为java的一个Method对象,代表Notify的onInvoke方法,这个很好理解final Method onInvokeMethod = (Method)StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_INVOKE_METHOD_KEY));// Notify对象final Object onInvokeInst = StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_INVOKE_INSTANCE_KEY));// 没有设置的话就直接返回if (onInvokeMethod == null && onInvokeInst == null ){return ;}if (onInvokeMethod == null || onInvokeInst == null ){throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() +" has a onreturn callback config , but no such "+(onInvokeMethod == null ? "method" : "instance")+" found. url:"+invoker.getUrl());}if (onInvokeMethod != null && ! onInvokeMethod.isAccessible()) {onInvokeMethod.setAccessible(true);}Object[] params = invocation.getArguments();try {onInvokeMethod.invoke(onInvokeInst, params);//反射调用} catch (InvocationTargetException e) {// 异常情况调用onthrow配置的方法fireThrowCallback(invoker, invocation, e.getTargetException());} catch (Throwable e) {fireThrowCallback(invoker, invocation, e);}}
syncCallback方法如下:
private void syncCallback(final Invoker<?> invoker, final Invocation invocation, final Result result) {if (result.hasException()) {//异常情况触发onthrowfireThrowCallback(invoker, invocation, result.getException());} else {//否则触发onreturnfireReturnCallback(invoker, invocation, result.getValue());}}
fireReturnCallback方法如下:
private void fireReturnCallback(final Invoker<?> invoker, final Invocation invocation, final Object result) {//....和oninvoke处理一样Object[] args = invocation.getArguments();Object[] params ;Class<?>[] rParaTypes = onReturnMethod.getParameterTypes() ;if (rParaTypes.length >1 ) {// 如果Notify方法的参数有多个// 有两个参数,且第二个参数为Object或者Object的数组if (rParaTypes.length == 2 && rParaTypes[1].isAssignableFrom(Object[].class)){// 构造两个参数的数组,一个为结果result,一个为请求入参params = new Object[2];params[0] = result;params[1] = args ;}else {// 这种情况,假设Notify有3个参数,如果本来方法入参有2个// 那么第一个为结果,后面为入参,如果入参只有1个,那么会导致异常,因为参数不匹配params = new Object[args.length + 1];params[0] = result;System.arraycopy(args, 0, params, 1, args.length);}} else {params = new Object[] { result };}try {onReturnMethod.invoke(onReturnInst, params);} catch (InvocationTargetException e) {fireThrowCallback(invoker, invocation, e.getTargetException());} catch (Throwable e) {fireThrowCallback(invoker, invocation, e);}}
这种情况下和onvoke多了参数的一些判断
asyncCallback方法如下:
private void asyncCallback(final Invoker<?> invoker, final Invocation invocation) {Future<?> f = RpcContext.getContext().getFuture();if (f instanceof FutureAdapter) {ResponseFuture future = ((FutureAdapter<?>)f).getFuture();future.setCallback(new ResponseCallback() {public void done(Object rpcResult) {//....Result result = (Result) rpcResult;if (result.hasException()) {fireThrowCallback(invoker, invocation, result.getException());} else {fireReturnCallback(invoker, invocation, result.getValue());}}public void caught(Throwable exception) {fireThrowCallback(invoker, invocation, exception);}});}}
如果是异步的方法,那么返回的就是一个future了,这时候在future上注册一个回调, 在future已经完成的情况下触发配置好的回调
注意:低版本的有个BUG,在使用oninvoke的时候会报找不到bean的错误,这是因为在解析method标签的时候,没有处理oninvoke这个节点,导致失败,具体代码在com.alibaba.dubbo.config.spring.schema.DubboBeanDefinitionParser#parse中,其中处理了onthrow和onreturn,但是少了oninvoke
这篇关于Dubbo源码分析----过滤器之FutureFilter的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!