Dubbo（四）：深入理解Dubbo核心模型Invoker
方块人 人气:1一、Dubbo中Invoker介绍
为什么说Invoker是Dubbo核心模型呢？
Invoker是Dubbo中的实体域，也就是真实存在的。其他模型都向它靠拢或转换成它，它也就代表一个可执行体，可向它发起invoke调用。在服务提供方，Invoker用于调用服务提供类。在服务消费方，Invoker用于执行远程调用。
二、服务提供方的Invoker
在服务提供方中的Invoker是由ProxyFactory创建而来的，Dubbo默认的ProxyFactory实现类为JavassistProxyFactory。
创建Invoker的入口方法getInvoker：
1 public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
2 // 为目标类创建 Wrapper
3 final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf(36) < 0 ? proxy.getClass() : type);
4 // 创建匿名 Invoker 类对象，并实现 doInvoke 方法。
5 return new AbstractProxyInvoker<T>(proxy, type, url) {
6 @Override
7 protected Object doInvoke(T proxy, String methodName,
8 Class<?>[] parameterTypes,
9 Object[] arguments) throws Throwable {
10 // 调用 Wrapper 的 invokeMethod 方法，invokeMethod 最终会调用目标方法
11 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
12 }
13 };
14 }
JavassistProxyFactory创建了一个继承自AbstractProxyInvoker类的匿名对象，并覆写了抽象方法doInvoke。覆写后的doInvoke 逻辑比较简单，仅是将调用请求转发给了Wrapper类的invokeMethod 方法。以及生成 invokeMethod 方法代码和其他一些方法代码。代码生成完毕后，通过 Javassist 生成 Class 对象，最后再通过反射创建 Wrapper 实例。
注：Wapper是一个包装类。主要用于“包裹”目标类，仅可以通过getWapper(Class)方法创建子类。在创建子类过程中，子类代码会对传进来的Class对象进行解析，拿到类方法，类成员变量等信息。而这个包装类持有实际的扩展点实现类。也可以把扩展点的公共逻辑全部移到包装类中，功能上就是作为AOP实现。
创建包装类的构造方法：
1 public static Wrapper getWrapper(Class<?> c) {
2 while (ClassGenerator.isDynamicClass(c))
3 c = c.getSuperclass();
4
5 if (c == Object.class)
6 return OBJECT_WRAPPER;
7
8 // 从缓存中获取 Wrapper 实例
9 Wrapper ret = WRAPPER_MAP.get(c);
10 if (ret == null) {
11 // 缓存未命中，创建 Wrapper
12 ret = makeWrapper(c);
13 // 写入缓存
14 WRAPPER_MAP.put(c, ret);
15 }
16 return ret;
17 }
在缓存中获取不到Wapper就会进入下面的方法makeWapper：
1 private static Wrapper makeWrapper(Class<?> c) {
2 // 检测 c 是否为基本类型，若是则抛出异常
3 if (c.isPrimitive())
4 throw new IllegalArgumentException("Can not create wrapper for primitive type: " + c);
5
6 String name = c.getName();
7 ClassLoader cl = ClassHelper.getClassLoader(c);
8
9 // c1 用于存储 setPropertyValue 方法代码
10 StringBuilder c1 = new StringBuilder("public void setPropertyValue(Object o, String n, Object v){ ");
11 // c2 用于存储 getPropertyValue 方法代码
12 StringBuilder c2 = new StringBuilder("public Object getPropertyValue(Object o, String n){ ");
13 // c3 用于存储 invokeMethod 方法代码
14 StringBuilder c3 = new StringBuilder("public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws " + InvocationTargetException.class.getName() + "{ ");
15
16 // 生成类型转换代码及异常捕捉代码，比如：
17 // DemoService w; try { w = ((DemoServcie) $1); }}catch(Throwable e){ throw new IllegalArgumentException(e); }
18 c1.append(name).append(" w; try{ w = ((").append(name).append(")$1); }catch(Throwable e){ throw new IllegalArgumentException(e); }");
19 c2.append(name).append(" w; try{ w = ((").append(name).append(")$1); }catch(Throwable e){ throw new IllegalArgumentException(e); }");
20 c3.append(name).append(" w; try{ w = ((").append(name).append(")$1); }catch(Throwable e){ throw new IllegalArgumentException(e); }");
21
22 // pts 用于存储成员变量名和类型
23 Map<String, Class<?>> pts = new HashMap<String, Class<?>>();
24 // ms 用于存储方法描述信息（可理解为方法签名）及 Method 实例
25 Map<String, Method> ms = new LinkedHashMap<String, Method>();
26 // mns 为方法名列表
27 List<String> mns = new ArrayList<String>();
28 // dmns 用于存储“定义在当前类中的方法”的名称
29 List<String> dmns = new ArrayList<String>();
30
31 // --------------------------------✨ 分割线1 ✨-------------------------------------
32
33 // 获取 public 访问级别的字段，并为所有字段生成条件判断语句
34 for (Field f : c.getFields()) {
35 String fn = f.getName();
36 Class<?> ft = f.getType();
37 if (Modifier.isStatic(f.getModifiers()) || Modifier.isTransient(f.getModifiers()))
38 // 忽略关键字 static 或 transient 修饰的变量
39 continue;
40
41 // 生成条件判断及赋值语句，比如：
42 // if( $2.equals("name") ) { w.name = (java.lang.String) $3; return;}
43 // if( $2.equals("age") ) { w.age = ((Number) $3).intValue(); return;}
44 c1.append(" if( $2.equals(\"").append(fn).append("\") ){ w.").append(fn).append("=").append(arg(ft, "$3")).append("; return; }");
45
46 // 生成条件判断及返回语句，比如：
47 // if( $2.equals("name") ) { return ($w)w.name; }
48 c2.append(" if( $2.equals(\"").append(fn).append("\") ){ return ($w)w.").append(fn).append("; }");
49
50 // 存储 <字段名, 字段类型> 键值对到 pts 中
51 pts.put(fn, ft);
52 }
53
54 // --------------------------------✨ 分割线2 ✨-------------------------------------
55
56 Method[] methods = c.getMethods();
57 // 检测 c 中是否包含在当前类中声明的方法
58 boolean hasMethod = hasMethods(methods);
59 if (hasMethod) {
60 c3.append(" try{");
61 }
62 for (Method m : methods) {
63 if (m.getDeclaringClass() == Object.class)
64 // 忽略 Object 中定义的方法
65 continue;
66
67 String mn = m.getName();
68 // 生成方法名判断语句，比如：
69 // if ( "sayHello".equals( $2 )
70 c3.append(" if( \"").append(mn).append("\".equals( $2 ) ");
71 int len = m.getParameterTypes().length;
72 // 生成“运行时传入的参数数量与方法参数列表长度”判断语句，比如：
73 // && $3.length == 2
74 c3.append(" && ").append(" $3.length == ").append(len);
75
76 boolean override = false;
77 for (Method m2 : methods) {
78 // 检测方法是否存在重载情况，条件为：方法对象不同 && 方法名相同
79 if (m != m2 && m.getName().equals(m2.getName())) {
80 override = true;
81 break;
82 }
83 }
84 // 对重载方法进行处理，考虑下面的方法：
85 // 1. void sayHello(Integer, String)
86 // 2. void sayHello(Integer, Integer)
87 // 方法名相同，参数列表长度也相同，因此不能仅通过这两项判断两个方法是否相等。
88 // 需要进一步判断方法的参数类型
89 if (override) {
90 if (len > 0) {
91 for (int l = 0; l < len; l++) {
92 // 生成参数类型进行检测代码，比如：
93 // && $3[0].getName().equals("java.lang.Integer")
94 // && $3[1].getName().equals("java.lang.String")
95 c3.append(" && ").append(" $3[").append(l).append("].getName().equals(\"")
96 .append(m.getParameterTypes()[l].getName()).append("\")");
97 }
98 }
99 }
100
101 // 添加 ) {，完成方法判断语句，此时生成的代码可能如下（已格式化）：
102 // if ("sayHello".equals($2)
103 // && $3.length == 2
104 // && $3[0].getName().equals("java.lang.Integer")
105 // && $3[1].getName().equals("java.lang.String")) {
106 c3.append(" ) { ");
107
108 // 根据返回值类型生成目标方法调用语句
109 if (m.getReturnType() == Void.TYPE)
110 // w.sayHello((java.lang.Integer)$4[0], (java.lang.String)$4[1]); return null;
111 c3.append(" w.").append(mn).append('(').append(args(m.getParameterTypes(), "$4")).append(");").append(" return null;");
112 else
113 // return w.sayHello((java.lang.Integer)$4[0], (java.lang.String)$4[1]);
114 c3.append(" return ($w)w.").append(mn).append('(').append(args(m.getParameterTypes(), "$4")).append(");");
115
116 // 添加 }, 生成的代码形如（已格式化）：
117 // if ("sayHello".equals($2)
118 // && $3.length == 2
119 // && $3[0].getName().equals("java.lang.Integer")
120 // && $3[1].getName().equals("java.lang.String")) {
121 //
122 // w.sayHello((java.lang.Integer)$4[0], (java.lang.String)$4[1]);
123 // return null;
124 // }
125 c3.append(" }");
126
127 // 添加方法名到 mns 集合中
128 mns.add(mn);
129 // 检测当前方法是否在 c 中被声明的
130 if (m.getDeclaringClass() == c)
131 // 若是，则将当前方法名添加到 dmns 中
132 dmns.add(mn);
133 ms.put(ReflectUtils.getDesc(m), m);
134 }
135 if (hasMethod) {
136 // 添加异常捕捉语句
137 c3.append(" } catch(Throwable e) { ");
138 c3.append(" throw new java.lang.reflect.InvocationTargetException(e); ");
139 c3.append(" }");
140 }
141
142 // 添加 NoSuchMethodException 异常抛出代码
143 c3.append(" throw new " + NoSuchMethodException.class.getName() + "(\"Not found method \\\"\"+$2+\"\\\" in class " + c.getName() + ".\"); }");
144
145 // --------------------------------✨ 分割线3 ✨-------------------------------------
146
147 Matcher matcher;
148 // 处理 get/set 方法
149 for (Map.Entry<String, Method> entry : ms.entrySet()) {
150 String md = entry.getKey();
151 Method method = (Method) entry.getValue();
152 // 匹配以 get 开头的方法
153 if ((matcher = ReflectUtils.GETTER_METHOD_DESC_PATTERN.matcher(md)).matches()) {
154 // 获取属性名
155 String pn = propertyName(matcher.group(1));
156 // 生成属性判断以及返回语句，示例如下：
157 // if( $2.equals("name") ) { return ($w).w.getName(); }
158 c2.append(" if( $2.equals(\"").append(pn).append("\") ){ return ($w)w.").append(method.getName()).append("(); }");
159 pts.put(pn, method.getReturnType());
160
161 // 匹配以 is/has/can 开头的方法
162 } else if ((matcher = ReflectUtils.IS_HAS_CAN_METHOD_DESC_PATTERN.matcher(md)).matches()) {
163 String pn = propertyName(matcher.group(1));
164 // 生成属性判断以及返回语句，示例如下：
165 // if( $2.equals("dream") ) { return ($w).w.hasDream(); }
166 c2.append(" if( $2.equals(\"").append(pn).append("\") ){ return ($w)w.").append(method.getName()).append("(); }");
167 pts.put(pn, method.getReturnType());
168
169 // 匹配以 set 开头的方法
170 } else if ((matcher = ReflectUtils.SETTER_METHOD_DESC_PATTERN.matcher(md)).matches()) {
171 Class<?> pt = method.getParameterTypes()[0];
172 String pn = propertyName(matcher.group(1));
173 // 生成属性判断以及 setter 调用语句，示例如下：
174 // if( $2.equals("name") ) { w.setName((java.lang.String)$3); return; }
175 c1.append(" if( $2.equals(\"").append(pn).append("\") ){ w.").append(method.getName()).append("(").append(arg(pt, "$3")).append("); return; }");
176 pts.put(pn, pt);
177 }
178 }
179
180 // 添加 NoSuchPropertyException 异常抛出代码
181 c1.append(" throw new " + NoSuchPropertyException.class.getName() + "(\"Not found property \\\"\"+$2+\"\\\" filed or setter method in class " + c.getName() + ".\"); }");
182 c2.append(" throw new " + NoSuchPropertyException.class.getName() + "(\"Not found property \\\"\"+$2+\"\\\" filed or setter method in class " + c.getName() + ".\"); }");
183
184 // --------------------------------✨ 分割线4 ✨-------------------------------------
185
186 long id = WRAPPER_CLASS_COUNTER.getAndIncrement();
187 // 创建类生成器
188 ClassGenerator cc = ClassGenerator.newInstance(cl);
189 // 设置类名及超类
190 cc.setClassName((Modifier.isPublic(c.getModifiers()) ? Wrapper.class.getName() : c.getName() + "$sw") + id);
191 cc.setSuperClass(Wrapper.class);
192
193 // 添加默认构造方法
194 cc.addDefaultConstructor();
195
196 // 添加字段
197 cc.addField("public static String[] pns;");
198 cc.addField("public static " + Map.class.getName() + " pts;");
199 cc.addField("public static String[] mns;");
200 cc.addField("public static String[] dmns;");
201 for (int i = 0, len = ms.size(); i < len; i++)
202 cc.addField("public static Class[] mts" + i + ";");
203
204 // 添加方法代码
205 cc.addMethod("public String[] getPropertyNames(){ return pns; }");
206 cc.addMethod("public boolean hasProperty(String n){ return pts.containsKey($1); }");
207 cc.addMethod("public Class getPropertyType(String n){ return (Class)pts.get($1); }");
208 cc.addMethod("public String[] getMethodNames(){ return mns; }");
209 cc.addMethod("public String[] getDeclaredMethodNames(){ return dmns; }");
210 cc.addMethod(c1.toString());
211 cc.addMethod(c2.toString());
212 cc.addMethod(c3.toString());
213
214 try {
215 // 生成类
216 Class<?> wc = cc.toClass();
217
218 // 设置字段值
219 wc.getField("pts").set(null, pts);
220 wc.getField("pns").set(null, pts.keySet().toArray(new String[0]));
221 wc.getField("mns").set(null, mns.toArray(new String[0]));
222 wc.getField("dmns").set(null, dmns.toArray(new String[0]));
223 int ix = 0;
224 for (Method m : ms.values())
225 wc.getField("mts" + ix++).set(null, m.getParameterTypes());
226
227 // 创建 Wrapper 实例
228 return (Wrapper) wc.newInstance();
229 } catch (RuntimeException e) {
230 throw e;
231 } catch (Throwable e) {
232 throw new RuntimeException(e.getMessage(), e);
233 } finally {
234 cc.release();
235 ms.clear();
236 mns.clear();
237 dmns.clear();
238 }
239 }
代码较长，同样注释也很多。大致说一下里面逻辑：
-
- 创建c1，c2，c3三个字符串，用于存储类型转换代码和异常捕捉代码，而后pts用于存储成员变量名和类型，ms用于存储方法描述信息（可理解为方法签名）及Method实例，mns为方法名列表，dmns用于存储“定义在当前类中的方法”的名称。在这里做完了一些初始工作
- 获取所有public字段，用c1存储条件判断及赋值语句，可以理解为通过c1能够为public字段赋值，而c2是条件判断及返回语句，同样的是得到public字段的值。再用pts存储<字段名，字段类型>。也就是现在能对目标类字段进行操作了，而要操作一些私有字段，是要访问set开头和get开头的方法，同样这些方法也都对应使用c1存set，c2存get，pts存储<属性名，属性类型>
- 现在到类中的方法，先检查方法中的参数，然后再检查是否有重载的方法。通过c3存储调用目标方法的语句以及方法中可能会抛出的异常，而后用mns集合进行存储方法名，对已经声明的方法存到ms中，未声明但是定义了的方法存在dmns中。
- 通过ClassGenerator为刚刚生成的代码构建Class类，并通过反射创建对象。ClassGenerator是Dubbo自己封装的，该类的核心是toClass()的重载方法 toClass(ClassLoader, ProtectionDomain)，该方法通过javassist构建Class。
最后在创建完成Wapper类，回到上面的getInvoker方法然后通过下面这条语句
1 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
2
3 //进入到invokeMethod中
4
5 public Object invokeMethod(Object instance, String mn, Class<?>[] types, Object[] args) throws NoSuchMethodException {
6 if ("getClass".equals(mn)) {
7 return instance.getClass();
8 } else if ("hashCode".equals(mn)) {
9 return instance.hashCode();
10 } else if ("toString".equals(mn)) {
11 return instance.toString();
12 } else if ("equals".equals(mn)) {
13 if (args.length == 1) {
14 return instance.equals(args[0]);
15 } else {
16 throw new IllegalArgumentException("Invoke method [" + mn + "] argument number error.");
17 }
18 } else {
19 throw new NoSuchMethodException("Method [" + mn + "] not found.");
20 }
21 }
22 };
到这里Invoker就能实现调用服务提供类的方法了。也就是服务提供类的Invoker实体域创建完成。底层是通过javassist来构建对象的。
三、服务消费方的Invoker
在服务消费方，Invoker用于执行远程调用。Invoker是由 Protocol实现类构建而来。Protocol实现类有很多但是最常用的两个，分别是RegistryProtocol和DubboProtocol。
DubboProtocol的refer方法：
1 public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
2 optimizeSerialization(url);
3 // 创建 DubboInvoker
4 DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
5 invokers.add(invoker);
6 return invoker;
7 }
上述方法较为简单，最重要的一个在于getClients。这个方法用于获取客户端实例，实例类型为ExchangeClient。ExchangeClient实际上并不具备通信能力，它需要基于更底层的客户端实例进行通信。比如NettyClient、MinaClient等，默认情况下，Dubbo使用NettyClient进行通信。每次创建好的Invoker都会添加到invokers这个集合里。也就是可以认为服务消费方的Invoker是一个具有通信能力的Netty客户端
getClients方法：
1 private ExchangeClient[] getClients(URL url) {
2 // 是否共享连接
3 boolean service_share_connect = false;
4 // 获取连接数，默认为0，表示未配置
5 int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
6 // 如果未配置 connections，则共享连接
7 if (connections == 0) {
8 service_share_connect = true;
9 connections = 1;
10 }
11
12 ExchangeClient[] clients = new ExchangeClient[connections];
13 for (int i = 0; i < clients.length; i++) {
14 if (service_share_connect) {
15 // 获取共享客户端
16 clients[i] = getSharedClient(url);
17 } else {
18 // 初始化新的客户端
19 clients[i] = initClient(url);
20 }
21 }
22 return clients;
23 }
24
25 //进入到获取共享客户端方法
26
27 private ExchangeClient getSharedClient(URL url) {
28 String key = url.getAddress();
29 // 获取带有“引用计数”功能的 ExchangeClient
30 ReferenceCountExchangeClient client = referenceClientMap.get(key);
31 if (client != null) {
32 if (!client.isClosed()) {
33 // 增加引用计数
34 client.incrementAndGetCount();
35 return client;
36 } else {
37 referenceClientMap.remove(key);
38 }
39 }
40
41 locks.putIfAbsent(key, new Object());
42 synchronized (locks.get(key)) {
43 if (referenceClientMap.containsKey(key)) {
44 return referenceClientMap.get(key);
45 }
46
47 // 创建 ExchangeClient 客户端
48 ExchangeClient exchangeClient = initClient(url);
49 // 将 ExchangeClient 实例传给 ReferenceCountExchangeClient，这里使用了装饰模式
50 client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
51 referenceClientMap.put(key, client);
52 ghostClientMap.remove(key);
53 locks.remove(key);
54 return client;
55 }
56 }
57
58 //进入到初始化客户端方法
59
60 private ExchangeClient initClient(URL url) {
61
62 // 获取客户端类型，默认为 netty
63 String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
64
65 // 添加编解码和心跳包参数到 url 中
66 url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
67 url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
68
69 // 检测客户端类型是否存在，不存在则抛出异常
70 if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
71 throw new RpcException("Unsupported client type: ...");
72 }
73
74 ExchangeClient client;
75 try {
76 // 获取 lazy 配置，并根据配置值决定创建的客户端类型
77 if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
78 // 创建懒加载 ExchangeClient 实例
79 client = new LazyConnectExchangeClient(url, requestHandler);
80 } else {
81 // 创建普通 ExchangeClient 实例
82 client = Exchangers.connect(url, requestHandler);
83 }
84 } catch (RemotingException e) {
85 throw new RpcException("Fail to create remoting client for service...");
86 }
87 return client;
88 }
89
90 //进入到connect方法中，getExchanger 会通过 SPI 加载 HeaderExchangeClient 实例，这个方法比较简单，大家自己看一下吧。接下来分析 HeaderExchangeClient 的实现。
91
92 public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
93 if (url == null) {
94 throw new IllegalArgumentException("url == null");
95 }
96 if (handler == null) {
97 throw new IllegalArgumentException("handler == null");
98 }
99 url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
100 // 获取 Exchanger 实例，默认为 HeaderExchangeClient
101 return getExchanger(url).connect(url, handler);
102 }
103
104 //创建HeaderExchangeClient实例
105
106 public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
107 // 这里包含了多个调用，分别如下：
108 // 1. 创建 HeaderExchangeHandler 对象
109 // 2. 创建 DecodeHandler 对象
110 // 3. 通过 Transporters 构建 Client 实例
111 // 4. 创建 HeaderExchangeClient 对象
112 return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
113 }
114
115 //通过 Transporters 构建 Client 实例
116
117 public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
118 if (url == null) {
119 throw new IllegalArgumentException("url == null");
120 }
121 ChannelHandler handler;
122 if (handlers == null || handlers.length == 0) {
123 handler = new ChannelHandlerAdapter();
124 } else if (handlers.length == 1) {
125 handler = handlers[0];
126 } else {
127 // 如果 handler 数量大于1，则创建一个 ChannelHandler 分发器
128 handler = new ChannelHandlerDispatcher(handlers);
129 }
130
131 // 获取 Transporter 自适应拓展类，并调用 connect 方法生成 Client 实例
132 return getTransporter().connect(url, handler);
133 }
134
135 //创建Netty对象
136
137 public Client connect(URL url, ChannelHandler listener) throws RemotingException {
138 // 创建 NettyClient 对象
139 return new NettyClient(url, listener);
140 }
上面的源码大概分一下几个逻辑：
-
- 通过refer方法进入DubboInvoker实例的创建，在这个实例中其实serviceType，url，以及invokers都已经是不用去关心的，invokers可以说是存储以及创建好的Invoker。而最关键的在于getClient方法。可以这么认为，现在的Invoker是一个Netty客户端。而在服务提供方的Invoker是一个Wapper类。
- 在getClient方法里面首先根据connections数量决定是获取共享客户端还是创建新的客户端实例，默认情况下是获取共享客户端，但是获取共享客户端中若缓存中拿不到对应客户端也会新建一个客户端。最终返回的是ExchangeClient，而当前的ExchangeClient也没有通信能力，需要更加底层的Netty客户端。
- initClient方法首先获取用户配置的客户端类型，默认为Netty，然后检测用户配置的客户端类型是否存在，不存在就要抛出异常，最后根据lazy配置觉得创建什么类型的客户端。LazyConnectExchangeClient代码并不是很复杂，该类会在request方法被调用时通过Exchangers的connect方法创建 ExchangeClient客户端
- getExchanger会通过SPI加载HeaderExchangeClient实例。最后通过Transporter实现类以及调用Netty的API来创建Netty客户端。最后层层返回，就最后成为了底层为Netty上层为DubboInvoker实例的这样一个类。
RegistryProtocol中的refer：
1 public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
2 // 取 registry 参数值，并将其设置为协议头
3 url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
4 // 获取注册中心实例
5 Registry registry = registryFactory.getRegistry(url);
6 if (RegistryService.class.equals(type)) {
7 return proxyFactory.getInvoker((T) registry, type, url);
8 }
9
10 // 将 url 查询字符串转为 Map
11 Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
12 // 获取 group 配置
13 String group = qs.get(Constants.GROUP_KEY);
14 if (group != null && group.length() > 0) {
15 if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
16 || "*".equals(group)) {
17 // 通过 SPI 加载 MergeableCluster 实例，并调用 doRefer 继续执行服务引用逻辑
18 return doRefer(getMergeableCluster(), registry, type, url);
19 }
20 }
21
22 // 调用 doRefer 继续执行服务引用逻辑
23 return doRefer(cluster, registry, type, url);
24 }
25 private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
26 // 创建 RegistryDirectory 实例
27 RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
28 // 设置注册中心和协议
29 directory.setRegistry(registry);
30 directory.setProtocol(protocol);
31 Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
32 // 生成服务消费者链接
33 URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
34
35 // 注册服务消费者，在 consumers 目录下新节点
36 if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
37 && url.getParameter(Constants.REGISTER_KEY, true)) {
38 registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
39 Constants.CHECK_KEY, String.valueOf(false)));
40 }
41
42 // 订阅 providers、configurators、routers 等节点数据
43 directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
44 Constants.PROVIDERS_CATEGORY
45 + "," + Constants.CONFIGURATORS_CATEGORY
46 + "," + Constants.ROUTERS_CATEGORY));
47
48 // 一个注册中心可能有多个服务提供者，因此这里需要将多个服务提供者合并为一个
49 Invoker invoker = cluster.join(directory);
50 ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
51 return invoker;
52 }
53
54 //进入到集群创建Invoker模式
55
56 @SPI(FailoverCluster.NAME)
57 public interface Cluster {
58
59 /**
60 * 合并其中Directory的Invoker为一个Invoker
61 */
62 @Adaptive
63 <T> Invoker<T> join(Directory<T> directory) throws RpcException;
64 }
65
66 //进入到MockerClusterWrapper实现类中
67
68 public class MockClusterWrapper implements Cluster {
69
70 private Cluster cluster;
71
72 public MockClusterWrapper(Cluster cluster) {
73 this.cluster = cluster;
74 }
75
76 public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
77 return new MockClusterInvoker<T>(directory, this.cluster.join(directory));
78 }
79 }
80
81 //具体的invoke方法
82
83 public Result invoke(Invocation invocation) throws RpcException {
84 Result result = null;
85
86 String value = directory.getUrl().getMethodParameter(invocation.getMethodName(),
87 Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
88 if (value.length() == 0 || value.equalsIgnoreCase("false")){
89 //no mock
90 result = this.invoker.invoke(invocation);
91 } else if (value.startsWith("force")) {
92 if (logger.isWarnEnabled()) {
93 logger.info("force-mock: " + invocation.getMethodName() +
94 " force-mock enabled , url : " + directory.getUrl());
95 }
96 //force:direct mock
97 result = doMockInvoke(invocation, null);
98 } else {
99 //fail-mock
100 try {
101 result = this.invoker.invoke(invocation);
102 }catch (RpcException e) {
103 if (e.isBiz()) {
104 throw e;
105 } else {
106 if (logger.isWarnEnabled()) {
107 logger.info("fail-mock: " + invocation.getMethodName() +
108 " fail-mock enabled , url : " + directory.getUrl(), e);
109 }
110 //fail:mock
111 result = doMockInvoke(invocation, e);
112 }
113 }
114 }
115 return result;
116 }
大致说一下上面的逻辑：
-
- 当前的Invoker底层依然是NettyClient，但是此时注册中心是集群搭建模式。所以需要将多个Invoker合并为一个，这里是逻辑合并的。实际上Invoker底层还是会有多个，只是通过一个集群模式来管理。所以暴露出来的就是一个集群模式的Invoker。于是进入Cluster.join方法。
- Cluster是一个通用代理类，会根据URL中的cluster参数值定位到实际的Cluster实现类也就是FailoverCluster。这里用到了@SPI注解，也就是需要ExtensionLoader扩展点加载机制，而该机制在实例化对象是，会在实例化后自动套上Wapper
- 但是是集群模式所以需要Dubbo中另外一个核心机制——Mock。Mock可以在测试中模拟服务调用的各种异常情况，还可以实现服务降级。在MockerClusterInvoker中，Dubbo先检查URL中是否存在mock参数。（这个参数可以通过服务治理后台Consumer端的屏蔽和容错进行设置或者直接动态设置mock参数值）如果存在force开头，这不发起远程调用直接执行降级逻辑。如果存在fail开头，则在远程调用异常时才会执行降级逻辑。
- 可以说注册中心为集群模式时，Invoker就会外面多包裹一层mock逻辑。是通过Wapper机制实现的。最终可以在调用或者重试时，每次都通过Dubbo内部的负载均衡机制选出多个Invoker中的一个进行调用
四、总结
到这里Invoker的实现就可以是说完了，总结一下，在服务提供方Invoker是javassist创建的服务类的实例，可以实现调用服务类内部的方法和修改字段。而在服务消费方的Invoker是基于Netty的客户端。最终通过服务消费方Netty客户端获得服务提供方创建的服务类实例。而后消费方为保护服务类就需要为其创建代理类，这样就可以在不实例化服务类情况下安全有效的远程调用服务类内部方法并且得到具体数据了。
加载全部内容