文章目录
- 引言
- 正文
- 服务订阅
- Invoker的创建
- 单注册中心的Invoker创建
- Dubbo直连的Invoker创建
- 创建代理类
- Invoker的创建
- 服务订阅
引言
上一篇我们分析了服务发布的原理，可以看到默认是创建了一个Netty server，并通过Invoker调用服务，同样，在客户端也会创建一个Inovker对象，下面就一起来看看这个引用创建过程。
正文
服务订阅
服务端的dubbo:service配置对应的类为ServiceBean，同样的，dubbo:reference对应有一个ReferenceBean，该类中的getObject方法，就是获取客户端代理类以及订阅服务的开端（默认情况下，Dubbo使用懒加载方式，在ReferenceBean对应的服务被引用或注入到其它类的时候调用getObject方法；否则，在bean初始化完成后就会调用afterPropertiesSet方法，而该方法也会调用getObject方法），所以我们就从这个方法开始。
代码语言：javascript复制public Object getObject() throws Exception {
return get();
}
public synchronized T get() {
if (destroyed){
throw new IllegalStateException("Already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}这两个方法不用多说，判断接口代理类是否已经存在，不存在调用init方法初始化，而该方法中大部分是检查配置，关键点是createProxy方法：
代码语言：javascript复制private T createProxy(Map<String, String> map) {
.......
// 本地JVM调用
if (isJvmRefer) {
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
invoker = refprotocol.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " interfaceClass.getName());
}
} else {
// 配置了url属性，可能是点对点调用，也可能是写的注册中心的url
if (url != null && url.length() > 0) {
String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (url.getPath() == null || url.getPath().length() == 0) {
url = url.setPath(interfaceName);
}
// 如果是registry协议，说明是想连接注册中心，就设置refer参数到url
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else {
urls.add(ClusterUtils.mergeUrl(url, map));
}
}
}
} else {
// 加载注册中心url，并将zookeeper协议转为registry协议
List<URL> us = loadRegistries(false);
if (us != null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
}
}
}
if (urls.size() == 1) {
// 只有一个注册中心或者服务直连
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {
// 多个注册中心或者多个服务提供者直连，或者两者混合
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(refprotocol.refer(interfaceClass, url));
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // 用了最后一个registry url
}
}
if (registryURL != null) { // 有 注册中心协议的URL
// 对有注册中心的Cluster 只用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // 不是 注册中心的URL
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
// 创建服务代理
return (T) proxyFactory.getProxy(invoker);
}代码很长，我截取了关键的部分，首先判断是否为本地injvm调用，若不是则加载服务直连的url或注册中心的url，接着根据url数量判断生成相应的invoker（url数量为1说明是只有一个注册中心或者服务直连，则直接调用protocol.refer获得相应的invoker；大于1代表是多个注册中心或者多个服务提供者直连，或者两者混合，则会生成多个invoker，通过cluster.join合并），最后调用proxyFactory.getProxy(invoker)生成相应代理类。所以，这里主要逻辑就在invoker和代理类的生成。
Invoker的创建
在上一篇讲过，Invoker在服务端是服务提供类的代理类，通过proxyFactory.getInvoker创建；而在客户端，则用于执行远程调用，通过protocol.refer调用。但是Protocol的实现类有很多，这里主要分析单注册中心和dubbo协议直连的情况。
单注册中心的Invoker创建
如果是Zookeeper的单注册中心方式，invoker就是通过invoker = refprotocol.refer(interfaceClass, urls.get(0))这句代码创建的，这里我们能够很快的定位到RegistryProtocol.refer方法中：
代码语言：javascript复制public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 将registry协议转为zookeeper协议
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
// 这个代码上一篇文章也分析过了，首先这里是通过依赖注入注入的RegistryFactory$Adpative
// 对象，最终会创建一个Zookeeper连接并返回ZookeeperRegistry对象
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// 分组服务走这里
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
String group = qs.get(Constants.GROUP_KEY);
if (group != null && group.length() > 0 ) {
if ( ( Constants.COMMA_SPLIT_PATTERN.split( group ) ).length > 1
|| "*".equals( group ) ) {
return doRefer( getMergeableCluster(), registry, type, url );
}
}
// 非分组服务走这里，注意这里的cluster也是依赖注入进来的
return doRefer(cluster, registry, type, url);
}需要注意cluster对象，它是通过依赖注入进来的，猜猜它是个什么对象？是FailoverCluster吗？其实并不是的，实际应该为MockClusterWrapper(FailoverCluster)对象（在服务发布一节中已有相应的分析），有什么用，我们后面再分析，这里暂时不讨论。 在refer中主要是创建zookeeper连接，并获取Registry对象，然后通过doRefer订阅和调用服务。
代码语言：javascript复制private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// 服务目录，暂时不详细分析它，知道它包含了所有的服务url就行了
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
// 注册consumer服务
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
&& url.getParameter(Constants.REGISTER_KEY, true)) {
registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
Constants.CHECK_KEY, String.valueOf(false)));
}
// 监听providers、configurators、routers、category节点的变化
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
"," Constants.CONFIGURATORS_CATEGORY
"," Constants.ROUTERS_CATEGORY));
// 一个注册中心可能存在多个相同的服务提供者，需要将其合并为一个Invoker
return cluster.join(directory);
}这个方法首先创建了一个服务目录，该目录会监听除consumer节点以外（本身就是consumer，当然没必要再监听consumer的变化了）的其它节点状态，最后通过cluster合并服务目录并返回invoker（Directory和Cluster等待后文来分析）：
代码语言：javascript复制// MockClusterWrapper.join
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
// 这里的cluster是FailoverCluster
return new MockClusterInvoker<T>(directory,
this.cluster.join(directory));
}
// FailoverCluster.join
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailoverClusterInvoker<T>(directory);
}从上面可以看到，这里返回的是MockClusterInvoker对象，并持有FailoverClusterInvoker的引用，这个在分析服务调用过程时会用到。
Dubbo直连的Invoker创建
如果是通过Dubbo协议直连服务的话，也是通过invoker = refprotocol.refer(interfaceClass, urls.get(0))创建Invoker，只不过会进入到DubboProtocol.refer方法中：
代码语言：javascript复制public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}这个方法主要通过getClients获取客户端ExchangeClient的实例，并实例化DubboInvoker返回。但是实际通信的肯定不是ExchangeClient，因为服务端默认使用的是Netty，那么客户端也应该是，我们可以点进去看看如何创建的。
代码语言：javascript复制private ExchangeClient[] getClients(URL url){
// 是否共享连接
boolean service_share_connect = false;
int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
//如果connections不配置，则共享连接，否则每服务每次新建连接
if (connections == 0){
service_share_connect = true;
connections = 1;
}
ExchangeClient[] clients = new ExchangeClient[connections];
for (int i = 0; i < clients.length; i ) {
if (service_share_connect){
// 获取共享连接
clients[i] = getSharedClient(url);
} else {
// 创建配置的连接数
clients[i] = initClient(url);
}
}
return clients;
}dubbo:reference可以通过connections配置连接数，调用initClient方法创建连接，不配置默认使用共享连接，调用getSharedClient获取，首先来看看getSharedClient方法：
代码语言：javascript复制private ExchangeClient getSharedClient(URL url){
// 从缓存中获取客户端，该客户端具有计数功能
String key = url.getAddress();
ReferenceCountExchangeClient client = referenceClientMap.get(key);
if ( client != null ){
if ( !client.isClosed()){
// 每被引用一次，计数就 1
client.incrementAndGetCount();
return client;
} else {
referenceClientMap.remove(key);
}
}
// 创建客户端实例
ExchangeClient exchagneclient = initClient(url);
// 使用ReferenceCountExchangeClient装饰，使其具有引用计数功能
client = new ReferenceCountExchangeClient(exchagneclient, ghostClientMap);
referenceClientMap.put(key, client);
ghostClientMap.remove(key);
return client;
}该方法中也调用了initClient方法创建客户端，并使用ReferenceCountExchangeClient装饰，使其具有引用计数的功能，每被使用一次计数就 1。再看看客户端是如何被创建的：
代码语言：javascript复制private ExchangeClient initClient(URL url) {
// 获取客户端类型，默认使用netty
String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
boolean compatible = (version != null && version.startsWith("1.0."));
url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
//默认开启heartbeat
url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
// BIO存在严重性能问题，暂时不允许使用
if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported client type: " str ","
" supported client type is " StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
}
ExchangeClient client ;
try {
// 创建客户端时是否立即创建连接，lazy表示不立即创建
if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){
client = new LazyConnectExchangeClient(url ,requestHandler);
} else {
client = Exchangers.connect(url ,requestHandler);
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service(" url
"): " e.getMessage(), e);
}
return client;
}默认创建一个netty客户端，并根据配置判断是否立即创建连接，若使用懒加载则会在请求服务时才创建连接。但不管是否是懒加载，都是通过Exchangers.connect方法创建的连接：
代码语言：javascript复制public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
return getExchanger(url).connect(url, handler);
}这里的getExchanger获取到的和服务端一样，也是HeaderExchanger：
代码语言：javascript复制public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}是不是感觉似曾相识，还记得服务端是怎么创建连接的吧，只不过当时是调用的Transporters.bind，而这里是Transporters.connect：
代码语言：javascript复制public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
ChannelHandler handler;
if (handlers == null || handlers.length == 0) {
handler = new ChannelHandlerAdapter();
} else if (handlers.length == 1) {
handler = handlers[0];
} else {
handler = new ChannelHandlerDispatcher(handlers);
}
return getTransporter().connect(url, handler);
}getTransporter方法也不用分析了，最终会进入到NettyTransporter.connect方法中：
代码语言：javascript复制public Client connect(URL url, ChannelHandler listener) throws RemotingException {
return new NettyClient(url, listener);
}就是去创建一个NettyClient客户端连接Server进行通信（Netty的源码分析不是本篇的重点，就先不分析了），这样DubboInvoker中就持有了NettyClient的引用了。至此，Invoker创建流程分析完成，下面一起来看看代理类的创建。
创建代理类
上面用大量的篇幅分析了Invoker的创建，拿到Invoker对象之后通过proxyFactory.getProxy创建代理类对象，这个proxyFactory很熟悉了，也是通过自适应扩展机制获取到的对象，所以应该会调用JavassistProxyFactory的getProxy方法，但是该类中没有与之匹配的方法（方法名相同，但是参数列表不同），所以首先应该进入其父类AbstractProxyFactory的getProxy方法中：
代码语言：javascript复制public <T> T getProxy(Invoker<T> invoker) throws RpcException {
Class<?>[] interfaces = null;
// 从url中获取接口
String config = invoker.getUrl().getParameter("interfaces");
if (config != null && config.length() > 0) {
String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
if (types != null && types.length > 0) {
interfaces = new Class<?>[types.length 2];
interfaces[0] = invoker.getInterface();
interfaces[1] = EchoService.class;
for (int i = 0; i < types.length; i ) {
interfaces[i 1] = ReflectUtils.forName(types[i]);
}
}
}
// 若url中未配置，则直接从invoker中获取
if (interfaces == null) {
interfaces = new Class<?>[] {invoker.getInterface(), EchoService.class};
}
// 调用Javaassist的getProxy方法
return getProxy(invoker, interfaces);
}
// JavaassistProxyFactory
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
// 生成 Proxy 子类Proxy0
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}这里会通过模板方法调用JavassistProxyFactory.getProxy，然后通过Proxy类生成一个接口代理类proxy0和一个Proxy的子类Proxy0。Proxy0主要是实现父类的抽象方法newInstance(InvocationHandler handler)：
代码语言：javascript复制public Object newInstance(java.lang.reflect.InvocationHandler h) {
return new com.alibaba.dubbo.common.bytecode.proxy0($1);
}可以看到该方法就是传入一个InvocationHandler并初始化proxy0对象，刚刚说了proxy0是服务接口的代理类，因此它是实现了我们定义的服务接口及方法：
代码语言：javascript复制private java.lang.reflect.InvocationHandler handler;
public java.lang.String sayHello(java.lang.String arg0) {
Object[] args = new Object[1];
args[0] = ($w)$1;
Object ret = handler.invoke(this, methods[0], args);
return (java.lang.String)ret;
}
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}最终就是去调用Invoker的invoker方法，所以这里会进入到相应的MockClusterInvoker.invoke方法（非服务直连创建的Invoker）以及AbstractInvoker.invoke方法（Dubbo直连时，由于DubboInvoker中没有invoker方法，所以调用其父类AbstractInvoker的）中，详细的调用过程下一篇再分析。
