Dubbo源码分析----过滤器之FutureFilter

本文主要是介绍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的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---