dubbo事件通知机制 （2）

此文已由作者赵计刚授权网易云社区发布。

欢迎访问网易云社区，了解更多网易技术产品运营经验。

142      * 反射执行xxxService.onthrow方法：至少要有一个入参且第一个入参类型为Throwable或其子类，接收返回结果
143      */
144     private void fireThrowCallback(final Invoker<?> invoker, final Invocation invocation, final Throwable exception) {
145         final Method onthrowMethod = (Method) StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_THROW_METHOD_KEY));
146         final Object onthrowInst = StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_THROW_INSTANCE_KEY));
147 
148         //onthrow callback not configured
149         if (onthrowMethod == null && onthrowInst == null) {
150             return;
151         }
152         if (onthrowMethod == null || onthrowInst == null) {
153             throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onthrow callback config , but no such " + (onthrowMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
154         }
155         if (onthrowMethod != null && !onthrowMethod.isAccessible()) {
156             onthrowMethod.setAccessible(true);
157         }
158         Class<?>[] rParaTypes = onthrowMethod.getParameterTypes();
159         if (rParaTypes[0].isAssignableFrom(exception.getClass())) {
160             try {
161                 Object[] args = invocation.getArguments();
162                 Object[] params;
163 
164                 if (rParaTypes.length > 1) {
165                     // onthrow(xx, Object[]) 两个参数：第一个参数接收exception，第二个接收所有的真实请求参数
166                     if (rParaTypes.length == 2 && rParaTypes[1].isAssignableFrom(Object[].class)) {
167                         params = new Object[2];
168                         params[0] = exception;
169                         params[1] = args;
170                     // onthrow(xx, Object... args) 多个参数：第一个参数接收exception，后边几个接收所有的真实请求参数
171                     } else {
172                         params = new Object[args.length + 1];
173                         params[0] = exception;
174                         System.arraycopy(args, 0, params, 1, args.length);
175                     }
176                 } else {
177                     // onthrow(xx) 只有一个参数：接收exception
178                     params = new Object[]{exception};
179                 }
180                 onthrowMethod.invoke(onthrowInst, params);
181             } catch (Throwable e) {
182                 logger.error(invocation.getMethodName() + ".call back method invoke error . callback method :" + onthrowMethod + ", url:" + invoker.getUrl(), e);
183             }
184         } else {
185             logger.error(invocation.getMethodName() + ".call back method invoke error . callback method :" + onthrowMethod + ", url:" + invoker.getUrl(), exception);
186         }
187     }
188 }

从@Activate(group = Constants.CONSUMER)来看FutureFilter只用在consumer端；不管是同步调用还是异步调用，都会走FutureFilter。

原理：

• 首先走oninvoke(String name)方法；

• 然后走sayHello(String name)

• 最后根据同步还是异步分别走不同的逻辑。 

其中同步很简单，看sayHello(String name)的返回结果RpcResult中是否有exception对象，如果有，执行onthrow(Throwable ex, String name)；如果没有执行onreturnWithoutParam(String result)。

异步的操作：由于不知道provider什么时候回执行完毕，所以要添加回调等待provider端返回结果后，再执行onthrow(Throwable ex, String name)或者onreturnWithoutParam(String result)，这种模式很重要，这是统计异步方法调用时间的一种非常好的模式。

重点看一下异步！

三、异步回调模式

 1     private void asyncCallback(final Invoker<?> invoker, final Invocation invocation) {
 2         Future<?> f = RpcContext.getContext().getFuture();
 3         if (f instanceof FutureAdapter) {
 4             ResponseFuture future = ((FutureAdapter<?>) f).getFuture();
 5             // 3.1 调用服务之后：设置回调ResponseCallback对象到DefaultFuture中，当provider返回响应时，执行DefaultFuture.doReceived方法，该方法会调用ResponseCallback对象的done或者caught方法
 6             future.setCallback(new ResponseCallback() {
 7                 public void done(Object rpcResult) {
 8                     if (rpcResult == null) {
 9                         logger.error(new IllegalStateException("invalid result value : null, expected " + Result.class.getName()));
10                         return;
11                     }
12                     ///must be rpcResult
13                     if (!(rpcResult instanceof Result)) {
14                         logger.error(new IllegalStateException("invalid result type :" + rpcResult.getClass() + ", expected " + Result.class.getName()));
15                         return;
16                     }
17                     Result result = (Result) rpcResult;
18                     if (result.hasException()) {
19                         fireThrowCallback(invoker, invocation, result.getException());
20                     } else {
21                         fireReturnCallback(invoker, invocation, result.getValue());
22                     }
23                 }
24 
25                 public void caught(Throwable exception) {
26                     fireThrowCallback(invoker, invocation, exception);
27                 }
28             });
29         }
30     }

上述的future对象是DefaultFuture，这里首先new了一个ResponseCallback回调函数，设置到了DefaultFuture的ResponseCallback callback属性中。来看一下DefaultFuture类：

 1     private volatile Response response;
 2     private volatile ResponseCallback callback;
 3 
 4     public boolean isDone() {
 5         return response != null;
 6     }
 7 
 8     public void setCallback(ResponseCallback callback) {
 9         if (isDone()) {
10             invokeCallback(callback);
11         } else {
12             boolean isdone = false;
13             lock.lock();
14             try {
15                 if (!isDone()) {
16                     this.callback = callback;
17                 } else {
18                     isdone = true;
19                 }
20             } finally {
21                 lock.unlock();
22             }
23             if (isdone) {
24                 invokeCallback(callback);
25             }
26         }
27     }

 1     private void invokeCallback(ResponseCallback c) {
 2         ResponseCallback callbackCopy = c;
 3         if (callbackCopy == null) {
 4             throw new NullPointerException("callback cannot be null.");
 5         }
 6         c = null;
 7         Response res = response;
 8         if (res == null) {
 9             throw new IllegalStateException("response cannot be null. url:" + channel.getUrl());
10         }
11 
12         if (res.getStatus() == Response.OK) {
13             try {
14                 // 返回正常，回调ResponseCallback回调函数的done方法
15                 callbackCopy.done(res.getResult());
16             } catch (Exception e) {
17                 logger.error("callback invoke error .reasult:" + res.getResult() + ",url:" + channel.getUrl(), e);
18             }
19         } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
20             try {
21                 TimeoutException te = new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
22                 // 如果超时，回调ResponseCallback回调函数的caught方法
23                 callbackCopy.caught(te);
24             } catch (Exception e) {
25                 logger.error("callback invoke error ,url:" + channel.getUrl(), e);
26             }
27         } else {
28             try {
29                 RuntimeException re = new RuntimeException(res.getErrorMessage());
30                 // 其他异常，回调ResponseCallback回调函数的caught方法
31                 callbackCopy.caught(re);
32             } catch (Exception e) {
33                 logger.error("callback invoke error ,url:" + channel.getUrl(), e);
34             }
35         }
36     }

从setCallback(ResponseCallback callback)，如果此时provider端已经返回了响应（response!=null），则直接执行ResponseCallback回调函数中的done方法或者caught方法；否则，将上边创建的ResponseCallback实例赋值给DefaultFuture的ResponseCallback callback属性中。那么之后会在什么时候执行回调函数的方法呢？当consumer接收到provider的响应的时候！

 1     public static void received(Channel channel, Response response) {
 2         try {
 3             DefaultFuture future = FUTURES.remove(response.getId());
 4             if (future != null) {
 5                 future.doReceived(response);
 6             } else {
 7                 logger.warn("The timeout response finally returned at "
 8                         + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
 9                         + ", response " + response
10                         + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
11                         + " -> " + channel.getRemoteAddress()));
12             }
13         } finally {
14             CHANNELS.remove(response.getId());
15         }
16     }
17 
18     private void doReceived(Response res) {
19         lock.lock();
20         try {
21             response = res;
22             if (done != null) {
23                 done.signal();
24             }
25         } finally {
26             lock.unlock();
27         }
28         // 调用回调函数
29         if (callback != null) {
30             invokeCallback(callback);
31         }
32     }

当provider返回响应时，会调用DefaultFuture.received(Channel channel, Response response)方法（9.3 客户端接收响应信息(异步转同步的实现)），此时会执行回调函数。事件通知的源码就分析完了！最后看一个回调模式的使用场景：统计异步方法的调用时间。

 1     private void asyncCallback(final Invoker<?> invoker, final Invocation invocation) {
 2         Future<?> f = RpcContext.getContext().getFuture();
 3         final long start = System.currentTimeMillis();
 4         if (f instanceof FutureAdapter) {
 5             ResponseFuture future = ((FutureAdapter<?>) f).getFuture();
 6             future.setCallback(new ResponseCallback() {
 7                 public void done(Object rpcResult) {
 8                     long cost = System.currentTimeMillis() - start;
 9                 }
10             });
11         }
12     }

上边的代码只是一个形式，实际上start时间需要在调用sayHello方法之前进行记录。

免费体验云安全(易盾)内容安全、验证码等服务

更多网易技术、产品、运营经验分享请点击。

相关文章：
【推荐】 MongoDB复制集与Raft协议异同点分析
【推荐】 Wireshark对HTTPS数据的解密
【推荐】 网易云易盾与A10 Networks达成战略合作 携手打造抗DDoS攻击的解决方案