1. 程式人生 > >dubbo事件通知機制(1)

dubbo事件通知機制(1)

此文已由作者嶽猛授權網易雲社群釋出。

歡迎訪問網易雲社群,瞭解更多網易技術產品運營經驗。

一、使用方式

兩個服務:

  • DemoService:真正要呼叫的服務

  • Notify:事件通知服務(用在consumer端)

provider

1 package com.alibaba.dubbo.demo;
2 
3 public interface DemoService {
4     String sayHello(String name);
5 }
1 public class DemoServiceImpl implements DemoService {
2     @Override
3     public String sayHello(String name) {
4         throw new RpcException("ex, param: " + name);//測試onthrow方法
5 //        return "Hello " + name;//測試onreturn方法
6     }
7 }

consumer

通知服務:Notify

1 package com.alibaba.dubbo.demo.consumer.eventnotify;
2 
3 public interface Notify {
4     void oninvoke(String name); // 呼叫之前
5     void onreturnWithoutParam(String result); // 呼叫之後
6     void onreturn(String result, String name); // 呼叫之後
7     void onthrow(Throwable ex, String name);  // 出現異常
8 }
  1 package com.alibaba.dubbo.demo.consumer.eventnotify;
 2 
 3 public class NotifyService implements Notify {
 4     @Override
 5     public void oninvoke(String name) {
 6         System.out.println("======oninvoke======, param: " + name);
 7     }
 8 
 9     @Override
10     public void onreturnWithoutParam(String result) {
11         System.out.println("======onreturn======, result: " + result);
12     }
13 
14     @Override
15     public void onreturn(String result, String name) {
16         System.out.println("======onreturn======, param: " + name + ", result: " + result);
17     }
18 
19     @Override
20     public void onthrow(Throwable ex, String name) {
21         System.out.println("======onthrow======, param: " + name + ", exception: " + ex.getMessage());
22     }
23 }

xml配置:

1     <bean id="notifyService"  class="com.alibaba.dubbo.demo.consumer.eventnotify.NotifyService"/>
2     <dubbo:reference id="demoService" check="false" interface="com.alibaba.dubbo.demo.DemoService">
3         <dubbo:method name="sayHello" timeout="60000" oninvoke="notifyService.oninvoke" onreturn="notifyService.onreturnWithoutParam" onthrow="notifyService.onthrow"/>
4     </dubbo:reference>

之後就可以執行Consumer啟動類,之後呼叫demoService.sayHello(String name)了。

注意

  • oninvoke方法:

    • 必須具有與真實的被呼叫方法sayHello相同的入參列表:例如,oninvoke(String name)

  • onreturn方法:

    • 至少要有一個入參且第一個入參必須與sayHello的返回型別相同,接收返回結果:例如,onreturnWithoutParam(String result)

    • 可以有多個引數,多個引數的情況下,第一個後邊的所有引數都是用來接收sayHello入參的:例如, onreturn(String result, String name)

  • onthrow方法:

    • 至少要有一個入參且第一個入參型別為Throwable或其子類,接收返回結果;例如,onthrow(Throwable ex)

    • 可以有多個引數,多個引數的情況下,第一個後邊的所有引數都是用來接收sayHello入參的:例如,onthrow(Throwable ex, String name)

  • 如果是consumer在呼叫provider的過程中,出現異常時不會走onthrow方法的,onthrow方法只會在provider返回的RpcResult中含有Exception物件時,才會執行。(dubbo中下層服務的Exception會被放在響應RpcResult的exception物件中傳遞給上層服務

二、原始碼解析

整個事件通知的邏輯都在FutureFilter中,來看一下原始碼:

1 /**
  2  * EventFilter
  3  */
  4 @Activate(group = Constants.CONSUMER)
  5 public class FutureFilter implements Filter {
  6 
  7     protected static final Logger logger = LoggerFactory.getLogger(FutureFilter.class);
  8 
  9     public Result invoke(final Invoker<?> invoker, final Invocation invocation) throws RpcException {
 10         final boolean isAsync = RpcUtils.isAsync(invoker.getUrl(), invocation);
 11 
 12         //1 呼叫服務之前:執行xxxService.oninvoke方法
 13         fireInvokeCallback(invoker, invocation);
 14         //2 呼叫服務
 15         Result result = invoker.invoke(invocation);
 16         //3 呼叫服務之後
 17         if (isAsync) {
 18             asyncCallback(invoker, invocation);
 19         } else {
 20             syncCallback(invoker, invocation, result); 21         }
 22         //4 返回呼叫結果
 23         return result;
 24     }
 25 
 26     private void syncCallback(final Invoker<?> invoker, final Invocation invocation, final Result result) {
 27         if (result.hasException()) {
 28             //3.1 呼叫服務之後:如果返回結果異常資訊(注意:如果是consumer自己throw的異常,會在2的時候直接拋走,不會走到這裡),直接執行xxxService.onthrow方法
 29             fireThrowCallback(invoker, invocation, result.getException());
 30         } else {
 31             //3.2 呼叫服務之後:如果返回值正常,執行xxxService.onreturn方法
 32             fireReturnCallback(invoker, invocation, result.getValue());
 33         }
 34     }
 35 
 36     private void asyncCallback(final Invoker<?> invoker, final Invocation invocation) {
 37         Future<?> f = RpcContext.getContext().getFuture();
 38         if (f instanceof FutureAdapter) {
 39             ResponseFuture future = ((FutureAdapter<?>) f).getFuture();
 40             // 3.1 呼叫服務之後:設定回撥ResponseCallback物件到DefaultFuture中,當provider返回響應時,執行DefaultFuture.doReceived方法,該方法會呼叫ResponseCallback物件的done或者caught方法
 41             future.setCallback(new ResponseCallback() {
 42                 public void done(Object rpcResult) {
 43                     if (rpcResult == null) {
 44                         logger.error(new IllegalStateException("invalid result value : null, expected " + Result.class.getName()));
 45                         return;
 46                     }
 47                     ///must be rpcResult
 48                     if (!(rpcResult instanceof Result)) {
 49                         logger.error(new IllegalStateException("invalid result type :" + rpcResult.getClass() + ", expected " + Result.class.getName()));
 50                         return;
 51                     }
 52                     Result result = (Result) rpcResult;
 53                     if (result.hasException()) {
 54                         fireThrowCallback(invoker, invocation, result.getException());
 55                     } else {
 56                         fireReturnCallback(invoker, invocation, result.getValue());
 57                     }
 58                 }
 59 
 60                 public void caught(Throwable exception) {
 61                     fireThrowCallback(invoker, invocation, exception);
 62                 }
 63             });
 64         }
 65     }
 66 
 67     /**
 68      * 反射執行xxxService.oninvoke方法:必須具有與真實的被呼叫方法sayHello相同的入參列表。
 69      */
 70     private void fireInvokeCallback(final Invoker<?> invoker, final Invocation invocation) {
 71         final Method onInvokeMethod = (Method) StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_INVOKE_METHOD_KEY));
 72         final Object onInvokeInst = StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_INVOKE_INSTANCE_KEY));
 73 
 74         if (onInvokeMethod == null && onInvokeInst == null) {
 75             return;
 76         }
 77         if (onInvokeMethod == null || onInvokeInst == null) {
 78             throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onreturn callback config , but no such " + (onInvokeMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
 79         }
 80         if (onInvokeMethod != null && !onInvokeMethod.isAccessible()) {
 81             onInvokeMethod.setAccessible(true);
 82         }
 83         // 獲取真實方法sayHello傳入的引數
 84         Object[] params = invocation.getArguments();
 85         try {
 86             onInvokeMethod.invoke(onInvokeInst, params);
 87         } catch (InvocationTargetException e) {
 88             fireThrowCallback(invoker, invocation, e.getTargetException());
 89         } catch (Throwable e) {
 90             fireThrowCallback(invoker, invocation, e);
 91         }
 92     }
 93 
 94     /**
 95      * 反射執行xxxService.onreturn方法:至少要有一個入參,接收返回結果
 96      */
 97     private void fireReturnCallback(final Invoker<?> invoker, final Invocation invocation, final Object result) {
 98         final Method onReturnMethod = (Method) StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_RETURN_METHOD_KEY));
 99         final Object onReturnInst = StaticContext.getSystemContext().get(StaticContext.getKey(invoker.getUrl(), invocation.getMethodName(), Constants.ON_RETURN_INSTANCE_KEY));
100 
101         //not set onreturn callback
102         if (onReturnMethod == null && onReturnInst == null) {
103             return;
104         }
105 
106         if (onReturnMethod == null || onReturnInst == null) {
107             throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onreturn callback config , but no such " + (onReturnMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
108         }
109         if (onReturnMethod != null && !onReturnMethod.isAccessible()) {
110             onReturnMethod.setAccessible(true);
111         }
112 
113         Object[] args = invocation.getArguments();
114         Object[] params;
115         Class<?>[] rParaTypes = onReturnMethod.getParameterTypes();
116         if (rParaTypes.length > 1) {
117             // onreturn(xx, Object[]) 兩個引數:第一個引數與真實方法sayHello方法返回結果型別相同,第二個接收所有的真實請求引數
118             if (rParaTypes.length == 2 && rParaTypes[1].isAssignableFrom(Object[].class)) {
119                 params = new Object[2];
120                 params[0] = result; // 真實方法的執行結果
121                 params[1] = args;   // 真實方法sayHello傳入的引數
122             // onreturn(xx, Object... args) 多個引數:第一個引數與真實方法sayHello方法返回結果型別相同,後邊幾個接收所有的真實請求引數
123             } else {
124                 params = new Object[args.length + 1];
125                 params[0] = result; // 真實方法的執行結果
126                 System.arraycopy(args, 0, params, 1, args.length);
127             }
128         } else {
129             // onreturn(xx) 只有一個引數:接收返回執行結果
130             params = new Object[]{result}; // 執行結果
131         }
132         try {
133             onReturnMethod.invoke(onReturnInst, params);
134         } catch (InvocationTargetException e) {
135             fireThrowCallback(invoker, invocation, e.getTargetException());
136         } catch (Throwable e) {
137             fireThrowCallback(invoker, invocation, e);
138         }
139     }
140 
141     /**

更多網易技術、產品、運營經驗分享請點選