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()) {//异常情况触发onthrow
fireThrowCallback(invoker, invocation, result.getException());
} else {//否则触发onreturn
fireReturnCallback(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
