该文介绍Dubbo的服务引入.在Dubbo中,我们可以通过两种方式引用远程服务。第一种是使用服务直连的方式引用服务,第二种方式是基于注册中心进行引用.服务直连的方式仅适合在调试或测试服务的场景下使用,不适合在线上环境使用.因此,本文我将重点分析通过注册中心引用服务的过程.
运行 demo-dubbo--》dubbo-demo-api--》dubbo-demo-api-consumer 中 Application:
public class ApplicationConsumer {
public static void main(String[] args) {
ReferenceConfig<DemoService> reference = new ReferenceConfig<>();
reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
reference.setInterface(DemoService.class);
DemoService service = reference.get();
String message = service.sayHello("dubbo");
System.out.println(message);
}
}断点进入org.apache.dubbo.config.ReferenceConfig#get:
代码语言：javascript复制public synchronized T get() {
checkAndUpdateSubConfigs();
if (destroyed) {
throw new IllegalStateException("The invoker of ReferenceConfig(" url ") has already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}该方法中会判断接口的代理对象是否为空,如果为空,则初始化,进入org.apache.dubbo.config.ReferenceConfig#init,该方法中大部分代码是对属性检查、获取,最关键的是看代理对象的创建,进入方法org.apache.dubbo.config.ReferenceConfig#createProxy:
代码语言：javascript复制private T createProxy(Map<String, String> map) {
if (shouldJvmRefer(map)) { //本地引用
...
...
} else { // 远程引用
urls.clear(); // reference retry init will add url to urls, lead to OOM
// url 不为空，表明用户可能想进行点对点调用
if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
// 当需要配置多个 url 时，可用分号进行分割，这里会进行切分
String[] us = SEMICOLON_SPLIT_PATTERN.split(url);
if (us != null && us.length > 0) {
...
...
}
} else { // assemble URL from register center's configuration
// if protocols not injvm checkRegistry
if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
checkRegistry();
// 加载注册中心 url
List<URL> us = loadRegistries(false);
...
...
}
}
// 单个注册中心或服务提供者(服务直连，下同)
if (urls.size() == 1) {
invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
} else { // 多个注册中心或多个服务提供者，或者两者混合
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // use last registry url
}
}
if (registryURL != null) { // registry url is available
// use RegistryAwareCluster only when register's CLUSTER is available
URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
// The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
invoker = CLUSTER.join(new StaticDirectory(u, invokers));
} else { // not a registry url, must be direct invoke.
invoker = CLUSTER.join(new StaticDirectory(invokers));
}
}
}
...
...
// create service proxy
return (T) PROXY_FACTORY.getProxy(invoker);
}创建Invoker
该方法首先判断是否为本地引用服务,否则远程,这里走的是远程引用服务,如果单个注册中心,则直接获取Invoker:
代码语言：javascript复制invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));REF_PROTOCOL对象是在运行时由Dubbo SPI机制创建的,这里会根据自适应扩展创建对象包装对象,结构如下:
代码语言：javascript复制org.apache.dubbo.rpc.protocol.ProtocolFilterWrapper
org.apache.dubbo.rpc.protocol.ProtocolListenerWrapper
org.apache.dubbo.registry.integration.RegistryProtocol前面文章也分析过,这里直接看RegistryProtocol中的refer:
代码语言：javascript复制// 修改协议为dubbo
url = URLBuilder.from(url)
.setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
.removeParameter(REGISTRY_KEY)
.build();
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_KEY));
String group = qs.get(GROUP_KEY);
if (group != null && group.length() > 0) {
if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
return doRefer(getMergeableCluster(), registry, type, url);
}
}
return doRefer(cluster, registry, type, url);继续看doRefer方法:
代码语言：javascript复制// 创建服务目录 RegistryDirectory 实例
RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
directory.setRegistry(registry);
directory.setProtocol(protocol);
// all attributes of REFER_KEY
Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
// 生成服务消费者链接
URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
// 注册服务消费者，在 consumers 目录下新节点
registry.register(directory.getRegisteredConsumerUrl());
}
directory.buildRouterChain(subscribeUrl);
// 订阅 providers、configurators、routers 等节点数据
directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
PROVIDERS_CATEGORY "," CONFIGURATORS_CATEGORY "," ROUTERS_CATEGORY));
// 一个注册中心可能有多个服务提供者，因此这里需要将多个服务提供者合并为一个
Invoker invoker = cluster.join(directory);
ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
return invoker;上面注释也很清楚,具体看:
代码语言：javascript复制Invoker invoker = cluster.join(directory);如果一个注册中心有多个服务提供者,这里需要将多个服务提供者合并为一个,cluster对象也是在运行时由Dubbo SPI生成的对象,会根据自适应创建包装类org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterWrapper,结构如下:
代码语言：javascript复制org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterWrapper
org.apache.dubbo.rpc.cluster.support.FailbackClusterDubbo提供了Failover Cluster(失败自动切换)、Failfast Cluster(快速失败)、Failsafe Cluster(失败安全)、Failback Cluster(失败自动恢复)、Forking Cluster(并行调用多个服务提供者)五种集群容错机制,默认为Failover,关于集群的这里不多作介绍, 继续进入MockClusterWrapper的join方法:
代码语言：javascript复制public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new MockClusterInvoker<T>(directory,
this.cluster.join(directory));
}在创建MockClusterInvoker对象之前,会调用org.apache.dubbo.rpc.cluster.support.FailbackCluster#join:
代码语言：javascript复制public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailbackClusterInvoker<T>(directory);
}会直接创建对象FailbackClusterInvoker,然后返回,到这里Invoker的对象创建已经完成.
创建代理对象
回到org.apache.dubbo.config.ReferenceConfig#createProxy方法中,最后一段代码用于创建代理对象,同理PROXY_FACTORY对象也是通过Dubbo SPI创建,根据自适应实现,创建对象StubProxyFactoryWrapper,进入org.apache.dubbo.rpc.proxy.wrapper.StubProxyFactoryWrapper#getProxy:
代码语言：javascript复制T proxy = proxyFactory.getProxy(invoker);
...
...进入org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory#getProxy:
代码语言：javascript复制public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}上面使用了jdk的动态代理实现代理对象的创建,具体看InvokerInvocationHandler的代码:
代码语言：javascript复制public class InvokerInvocationHandler implements InvocationHandler {
private static final Logger logger = LoggerFactory.getLogger(InvokerInvocationHandler.class);
private final Invoker<?> invoker;
public InvokerInvocationHandler(Invoker<?> handler) {
this.invoker = handler;
}
@Override
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();
}
}invoke方法里会实现Invoker的invoke方法调用.到这里代理对象的创建已经完成.
参考文章
- http://dubbo.apache.org/zh-cn/docs/sourcecodeguide/refer-service.html
