前言
在前面的章节中,我们已经完成Dubbo服务暴露的流程分析。今天我们一起来看Dubbo怎么引用这些服务的。
关于服务引用,Dubbo有两种方式。一种是基于注册中心进行服务引用,一种是服务直连进行引用。服务直连主要用于测试联调阶段,生产环境不推荐使用。它的配置也比较简单,在消费者端指定服务url即可。
<dubbo:reference id="infoUserService"
interface="com.viewscenes.netsupervisor.service.InfoUserService"
url="dubbo://192.168.139.129:20880"/>
本文将重点分析通过注册中心方式,服务引用的过程。开始之前,我们再回顾一下消费者端项目整体的XML文件配置。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
xmlns="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<!-- 消费者方应用信息,用于计算依赖关系 -->
<dubbo:application name="dubbo_consumer"/>
<!-- 用于配置连接注册中心相关信息 -->
<dubbo:registry protocol="zookeeper" address="192.168.139.131:2181" client="zkclient" />
<!-- 引用配置 用于创建一个远程接口服务代理 -->
<dubbo:reference id="infoUserService"
interface="com.viewscenes.netsupervisor.service.InfoUserService"/>
</beans>
一、引用的是什么?
在上述配置文件中,infoUserService
只是Spring中的一个Bean。在代码中,我们通过getBean()
来获取它,然后调用它的方法。
我们在RPC基本原理以及如何用Netty来实现RPC这篇文章中已经分析过,这里的bean其实是一个FactoryBean
,通过它可以返回一个接口的代理对象,完成调用逻辑的处理。
在代码中,我们通过这样来获取这个Bean。
InfoUserService infoUserService = (InfoUserService) context.getBean("infoUserService");
这里的infoUserService
就是一个代理对象,比如像proxy@2903
这种。当然了,它必然会包含一个InvocationHandler
,如下所示:
服务引用就是其实通过动态代理给接口创建代理对象并返回。当我们调用接口方法时,则调用到InvocationHandler
相关方法,处理相关请求。
二、从哪里开始?
在上面的配置文件中,dubbo:reference
对应的是ReferenceBean
处理类。那么Spring在实例化这个Bean的时候,就调用到里面方法。我们先看看它的类结构
package com.alibaba.dubbo.config.spring;
public class ReferenceBean<T> extends ReferenceConfig<T> implements
FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
private transient ApplicationContext applicationContext;
public ReferenceBean() {
super();
}
public ReferenceBean(Reference reference) {
super(reference);
}
//设置applicationContext
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
//返回代理对象
public Object getObject() throws Exception {
return get();
}
//返回代理对象接口类型
public Class<?> getObjectType() {
return getInterfaceClass();
}
//是否为单例
@Parameter(excluded = true)
public boolean isSingleton() {
return true;
}
//Bean初始化方法
public void afterPropertiesSet() throws Exception {
if(isInit()){
getObject();
}
}
}
首先,映入眼帘的是FactoryBean
接口。有了它,则证明当前bean是一个工厂Bean,所以我们重点关注它的getObject
方法,当我们的服务被注入到其他类中时,就会调用到此方法。
其次,是InitializingBean
接口。它是Bean的初始化方法,当Bean完成实例化之后被调用。在上述代码中,当init()
条件判断成立后调用getObject
。我们可以在配置文件中以这种方式来激活它:init="true"
以上就是服务引用的两个不同时机。再专业点来讲,一个是懒汉式,一个是饿汉式。Dubbo默认是懒汉式引用,需要时才会调用。
上述代码中,getObject
方法会调用到父类的init
方法。这个方法内容比较多,前面的部分是各种配置检查、赋值,然后就是创建代理对象返回。
private void init() {
//避免重复初始化
if (initialized) {
return;
}
//省略相关配置检查、赋值等代码...
//创建代理
ref = createProxy(map);
}
上述代码中,重点是createProxy
方法。它负责创建Invoker实例和创建代理对象。
也许我们还记得,在服务暴露的时候,会分为本地暴露和远程暴露。在这里,服务引用也是这样。Dubbo首先判断服务引用是本地引用还是远程引用,默认是远程引用。然后判断是否为直连服务,根据协议调用refer
方法创建invoker对象 。最后创建服务代理对象并返回。
private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
final boolean isJvmRefer;
//获取配置判断是否为本地引用
if (isInjvm() == null) {
if (url != null && url.length() > 0) {
isJvmRefer = false;
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
// by default, reference local service if there is
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
isJvmRefer = isInjvm().booleanValue();
}
//本地引用
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不为空则代表是服务直连
if (url != null && url.length() > 0) {
//切割多个url
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);
}
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
List<URL> us = loadRegistries(false);
if (us != null && !us.isEmpty()) {
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;
}
}
if (registryURL != null) {
// use AvailableCluster only when register's cluster is available
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else {
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
//创建代理类
return (T) proxyFactory.getProxy(invoker);
}
前面的整个流程我们可以分为两大部分:创建Invoker和创建代理。
三、创建Invoker
1、连接注册中心
我们以本文开头的XML配置文件为例,它是基于单个注册中心的远程服务引用。所以,我们重点分析invoker = refprotocol.refer(interfaceClass, urls.get(0));
我们需要清楚的认识到,urls.get(0)
获取到的是注册中心的地址。内容如下:registry://192.168.139.131:2181/org.apache.dubbo.registry.RegistryService?...
所以,我们先来看RegistryProtocol.refer
方法。它主要是获取注册中心,并执行服务引用。
public class RegistryProtocol implements Protocol {
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
//转换协议
url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
//通过zookeeper连接到注册中心
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
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);
}
}
return doRefer(cluster, registry, type, url);
}
}
我们接着看doRefer(cluster, registry, type, url);
我们可以认为它主要做了3件事:
- 向注册中心注册服务消费者(本身)
- 向注册中心订阅服务提供者
- 通过Cluster创建Invoker对象
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
// 创建 RegistryDirectory 对象,并设置注册中心
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
//创建订阅 URL
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.
remove("register.ip"), 0, type.getName(), parameters);
if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
&& url.getParameter(Constants.REGISTER_KEY, true)) {
//向注册中心注册服务消费者
registry.register(subscribeUrl.addParameters("category", "consumers",
"check", String.valueOf(false)));
}
// 向注册中心订阅服务提供者
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
//创建Invoker对象
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;
}
如上代码, 它的重点在于directory.subscribe
,它通过订阅注册中心的服务提供者,调用相关协议的引用方法。比如DubboProtocol.refer
,下面我们来看它是怎么调用的。
2、订阅服务提供者
我们通过directory.subscribe
这句代码, 会定位到ZookeeperRegistry.doSubscribe
方法。
public class ZookeeperRegistry extends FailbackRegistry {
protected void doSubscribe(final URL url, final NotifyListener listener) {
List<URL> urls = new ArrayList<URL>();
//获取注册中心除了消费者信息之外的所有节点
//比如configurators, routers, providers
String[] paths = toCategoriesPath(url);
for (String path :paths) {
//获取节点监听器管理容器
ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
if (listeners == null) {
zkListeners.putIfAbsent(url,
new ConcurrentHashMap<NotifyListener, ChildListener>());
listeners = zkListeners.get(url);
}
//获取节点监听器
ChildListener zkListener = listeners.get(listener);
//如果为空,则创建一个监听器
if (zkListener == null) {
listeners.putIfAbsent(listener, new ChildListener() {
@Override
public void childChanged(String parentPath, List<String> currentChilds) {
ZookeeperRegistry.this.notify(url, listener,
toUrlsWithEmpty(url, parentPath, currentChilds));
}
});
zkListener = listeners.get(listener);
}
//创建节点
zkClient.create(path, false);
//订阅节点path 数据变化,通知到zkListener
//即通知到ZookeeperRegistry.this.notify
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
//通知
notify(url, listener, urls);
}
}
如上代码,重点就是通过注册中心获取到服务提供者的url,调用notify方法。同时还订阅了服务提供者的节点数据变化,如果有变化也调用notify方法。
而在notify方法中,最终会根据url调用服务引用方法refer,再封装成InvokerDelegate对象。
invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
这样不仅完成了本次的服务引用,而且当服务提供者上下线的时候,消费者也可以动态感知。
3、Listener和Filter
以DubboProtocol为例,它是Wrapper包装类的对象,关于这一点我们一定要牢记。
回忆服务暴露的过程,我们可以监听服务暴露的过程,在构建调用链的时候可以增加Filter。在这里,服务引用也是一样的。
- 服务引用监听
在InvokerListener 接口中,有两个方法。分别是服务引用后调用和服务销毁时调用。
@SPI
public interface InvokerListener {
//服务引用
void referred(Invoker<?> invoker) throws RpcException;
//服务销毁
void destroyed(Invoker<?> invoker);
}
我们先自定义监听器,来实现接口的方法。
public class MyInvokerListener1 implements InvokerListener{
public void referred(Invoker<?> invoker) throws RpcException {
System.out.println("服务引用111.."+invoker.getUrl());
}
public void destroyed(Invoker<?> invoker) {
System.out.println("服务销毁111.."+invoker.getUrl());
}
}
然后创建扩展点配置文件,文件名称为:
com.alibaba.dubbo.rpc.InvokerListener
内容为:
listener1=com.viewscenes.netsupervisor.listener.MyInvokerListener1
最后,修改XML配置文件
<dubbo:consumer listener="listener1"/>
- Filter过滤器
首先自定义过滤器
public class MyFilter implements Filter{
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
System.out.println("调用之前-----------");
Result result = invoker.invoke(invocation);
System.out.println("调用之后-----------");
return result;
}
}
然后创建扩展点配置文件,文件名称为:
com.alibaba.dubbo.rpc.Filter
内容为:
myfilter1=com.viewscenes.netsupervisor.filter.MyFilter
最后,修改XML配置文件,说明自定义的过滤器在默认过滤器之前。
filter="myfilter1,default""/>
4、DubboProtocol
前面啰啰嗦嗦一大堆,终于来到正主。这里主要是获取通信客户端,拿到ExchangeClient,然后封装为DubboInvoker对象并返回。
public class DubboProtocol extends AbstractProtocol {
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
optimizeSerialization(url);
//获取通信客户端
ExchangeClient[] client = getClients(url);
//创建DubboInvoker对象
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, client, invokers);
invokers.add(invoker);
return invoker;
}
}
这里主要分为两个步骤:
- 创建通信客户端
在getClients方法里面,主要判断是否为共享连接的客户端。
private ExchangeClient[] getClients(URL url) {
// 是否共享连接
boolean service_share_connect = false;
// 获取连接数,默认为0,表示未配置
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默认创建共享连接客户端。我们继续看getSharedClient方法。
private ExchangeClient getSharedClient(URL url) {
//服务地址
String key = url.getAddress();
//查询缓存是否已经创建了客户端
ReferenceCountExchangeClient client = referenceClientMap.get(key);
if (client != null) {
if (!client.isClosed()) {
client.incrementAndGetCount();
return client;
} else {
referenceClientMap.remove(key);
}
}
synchronized (key.intern()) {
//初始化客户端
ExchangeClient exchangeClient = initClient(url);
//封装成ReferenceCountExchangeClient对象,用于引用计数
client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
//放入缓存并返回
referenceClientMap.put(key, client);
ghostClientMap.remove(key);
return client;
}
}
如上代码,先从缓存中获取客户端,未命中则调用initClient方法初始化客户端。并将它封装成ReferenceCountExchangeClient 对象返回。在initClient方法中,设置客户端类型、编解码和心跳信息,并调用connect创建客户端。
private ExchangeClient initClient(URL url) {
//客户端类型 默认为netty
String str = url.getParameter("client", url.getParameter("server","netty"));
//编解码
url = url.addParameter("codes", "dubbo");
//心跳
url = url.addParameterIfAbsent("heartbeat","60000");
ExchangeClient client;
try {
if (url.getParameter("lazy", 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;
}
如上代码,connect会调用到HeaderExchangeClient.connect方法。
public class HeaderExchanger implements Exchanger {
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
HeaderExchangeHandler headerExchangeHandler = new HeaderExchangeHandler(handler);
DecodeHandler decodeHandler = new DecodeHandler(headerExchangeHandler);
Client client = Transporters.connect(url, decodeHandler);
return new HeaderExchangeClient(client,true);
}
}
我们重点来看Transporters.connect
,它会调用到NettyClient
的构造方法。最终调用doOpen和doConnect方法,连接通信服务器。
public class NettyClient extends AbstractClient {
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
bootstrap = new ClientBootstrap(channelFactory);
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getTimeout());
//处理器
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(),
getUrl(), NettyClient.this);
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
}
protected void doConnect() throws Throwable {
long start = System.currentTimeMillis();
//获取服务器地址,并连接
ChannelFuture future = bootstrap.connect(getConnectAddress());
try {
boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
//连接成功后获取Channel
if (ret && future.isSuccess()) {
Channel newChannel = future.getChannel();
newChannel.setInterestOps(Channel.OP_READ_WRITE);
try {
// Close old channel
Channel oldChannel = NettyClient.this.channel; // copy reference
if (oldChannel != null) {
try {
oldChannel.close();
} finally {
NettyChannel.removeChannelIfDisconnected(oldChannel);
}
}
} finally {
if (NettyClient.this.isClosed()) {
try {
newChannel.close();
} finally {
NettyClient.this.channel = null;
NettyChannel.removeChannelIfDisconnected(newChannel);
}
} else {
//设置当前Channel
NettyClient.this.channel = newChannel;
}
}
}
}
}
}
以上大部分都是Netty的代码,连接服务器成功后获取到Channel保存。 我们重点看NettyHandler
,它是事件处理器。这里的messageReceived
负责接收处理返回的消息。接收到消息后,通过线程池拿到线程去执行后续的处理。不同的是,这里的线程池类型是CachedThreadPool ,即缓存型线程池。
这里的流程处理完毕,我们再把目光回到DubboInvoker
,它通过一个构造方法返回Invoker对象。
public DubboInvoker(Class<T> serviceType, URL url,
ExchangeClient[] clients, Set<Invoker<?>> invokers) {
this.clients = clients;
this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
this.invokers = invokers;
}
至此,Invoker对象已经创建完毕。我们回到开头的createProxy
方法。最后一步是为这个Invoker对象创建代理。
五、创建代理
在Dubbo,默认使用Javassist来创建代理。
public class JavassistProxyFactory extends AbstractProxyFactory {
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
//调用处理程序
InvocationHandler invocationHandler = new InvokerInvocationHandler(invoker)
//创建代理
return (T) Proxy.getProxy(interfaces).newInstance(invocationHandler);
}
}
创建的过程很简单,我们主要来看InvocationHandler
。既然是代理对象,我们每次调用接口方法的时候,会调用到这个对象的invoke
方法。
public class InvokerInvocationHandler implements InvocationHandler {
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]);
}
//把请求的方法名和参数等封装成RpcInvocation对象
//调用invoke
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
}
至此,服务引用的大体流程我们已经分析完了。我们再来总结一下它的过程:
- 通过FactoryBean触发服务引用方法
- 连接注册中心,获取服务提供者列表,创建invoker
- 订阅服务提供者节点,数据变更时获得通知
- 连接服务器,获得客户端
- 创建代理