服务注册(provider服务暴露)
Table of Contents generated with DocToc (opens new window)
# 架构
# 官网架构图
# 角色说明
Provider 暴露服务的服务提供方
Consumer 调用远程服务的服务消费方
Registry 服务注册与发现的注册中心
Monitor 统计服务的调用次调和调用时间的监控中心
Container 服务运行容器
# 调用关系说明
- 服务容器负责启动,加载,运行服务提供者。
- 服务提供者在启动时,向注册中心注册自己提供的服务。
- 服务消费者在启动时,向注册中心订阅自己所需的服务。
- 注册中心返回服务提供者地址列表给消费者,如果有变更,注册中心将基于长连接推送变更数据给消费者。
- 服务消费者,从提供者地址列表中,基于软负载均衡算法,选一台提供者进行调用,如果调用失败,再选另一台调用。
- 服务消费者和提供者,在内存中累计调用次数和调用时间,定时每分钟发送一次统计数据到监控中心。
# 1. 解析配置
# 1.1 解析XML
提示
在很多情况下,我们需要为系统提供可配置化支持,简单的做法可以直接基于Spring的标准Bean来配置,但配置较为复杂或者需要更多丰富控制的时候,会显得非常笨拙。一般的做法会用原生态的方式去解析定义好的xml文件,然后转化为配置对象,这种方式当然可以解决所有问题,但实现起来比较繁琐,特别是是在配置非常复杂的时候,解析工作是一个不得不考虑的负担。Spring提供了可扩展Schema的支持,这是一个不错的折中方案,完成一个自定义配置一般需要以下步骤:
- 设计配置属性和JavaBean
- 编写XSD文件
- 编写NamespaceHandler和BeanDefinitionParser完成解析工作
- 编写spring.handlers和spring.schemas串联起所有部件
- 在Bean文件中应用
同样的dubbo为了解决这个问题,起了一个叫dubbo-config-spring的模块。这个模块的下面的resources/META-INF文件下面有三个这样的文件dubbo.xsd、spring.handlers、spring.schemas
Dubbo也是利用了Spring提供的可扩展Schema机制实现了dubbo的xml配置文件解析
public class DubboNamespaceHandler extends NamespaceHandlerSupport implements ConfigurableSourceBeanMetadataElement {
public DubboNamespaceHandler() {
}
public void init() {
this.registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
this.registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
this.registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
this.registerBeanDefinitionParser("config-center", new DubboBeanDefinitionParser(ConfigCenterBean.class, true));
this.registerBeanDefinitionParser("metadata-report", new DubboBeanDefinitionParser(MetadataReportConfig.class, true));
this.registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
this.registerBeanDefinitionParser("metrics", new DubboBeanDefinitionParser(MetricsConfig.class, true));
this.registerBeanDefinitionParser("ssl", new DubboBeanDefinitionParser(SslConfig.class, true));
this.registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
this.registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
this.registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
this.registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
this.registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
this.registerBeanDefinitionParser("annotation", new AnnotationBeanDefinitionParser());
}
public BeanDefinition parse(Element element, ParserContext parserContext) {
BeanDefinitionRegistry registry = parserContext.getRegistry();
this.registerAnnotationConfigProcessors(registry);
DubboBeanUtils.registerCommonBeans(registry);
BeanDefinition beanDefinition = super.parse(element, parserContext);
this.setSource(beanDefinition);
return beanDefinition;
}
private void registerAnnotationConfigProcessors(BeanDefinitionRegistry registry) {
AnnotationConfigUtils.registerAnnotationConfigProcessors(registry);
}
static {
Version.checkDuplicate(DubboNamespaceHandler.class);
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
从这里知道所有的dubbo的标签,都是统一由DubboBeanDefinitionParser来解析的,每一个标签都会统一解析成对应的Bean对象。dubbo定义了以下配置类(见dubbo/dubbo-config/dubbo-config-api,准确说应该是Bean的父类), Bean(见dubbo/dubbo-config/dubbo-config-spring)
# 相关配置类
配置Dubbo服务时的DubboService标签对应的Bean就是ServiceBean,对应的配置类就是ServiceConfig 调用Dubbo服务时的DubboReference标签对应的Bean就是ReferenceBean,对应的配置类就是ReferenceConfig 这两个配置类都有一个xxxConfigBase,这个Base类里面的属性就是能配置的信息
# 1.2 检查配置
- 检查interface属性是否合法
- 检查providerConfig和ApplicationConfig等核心配置类对象是否为空,为空会从其他配置类获取对应的实例;
- 检查并处理泛化服务和普通服务类
- 检查本地存根配置
- 对ApplicationConfig、RegisterConfig等配置类进行检测,为空则尝试创建,无法创建则抛异常;
# 1.3 (重点)Spring容器启动完成后的事件开始执行暴露
我们都知道当 Spring 容器启动完成会发出 ApplicationContextEvent 事件,我们可以看到这个org.apache.dubbo.config.spring.context .DubboBootstrapApplicationListener# onApplicationContextEvent
方法核心代码:
public void onApplicationContextEvent(ApplicationContextEvent event) {
if (event instanceof ContextRefreshedEvent) {
//Spring 容器启动完成
onContextRefreshedEvent((ContextRefreshedEvent) event);
} else if (event instanceof ContextClosedEvent) {
//Spring 容器关闭
onContextClosedEvent((ContextClosedEvent) event);
}
}
private void onContextRefreshedEvent(ContextRefreshedEvent event) {
dubboBootstrap.start();
}
2
3
4
5
6
7
8
9
10
11
12
13
当接收到ContextRefreshedEvent事件后会最终会调用org.apache.dubbo.config.bootstrap. DubboBootstrap#start
方法启动服务注册核心代码如下:
public DubboBootstrap start() {
//判断服务是否启动 防止重复暴露服务 原子操作
if (started.compareAndSet(false, true)) {
initialize();
//..
// 暴露Dubbo服务
exportServices();
//...
}
return this;
}
2
3
4
5
6
7
8
9
10
11
上面的代码先进行原子操作去设置启动标识防止重复暴露服务,其exportServices
代码如下:
private void exportServices() {
//循环所有需要服务暴露的配置
configManager.getServices().forEach(sc -> {
// TODO, compatible with ServiceConfig.export()
ServiceConfig serviceConfig = (ServiceConfig) sc;
serviceConfig.setBootstrap(this);
//是否异步导出
if (exportAsync) {
//获取线程池
ExecutorService executor = executorRepository.getServiceExporterExecutor();
Future<?> future = executor.submit(() -> {
//异步服务暴露
sc.export();
});
asyncExportingFutures.add(future);
} else {
//同步服务暴露导出
sc.export();
exportedServices.add(sc);
}
});
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
上面的代码是循环对所有的 Dubbo 服务进行暴露,注意这里有一个exportAsync标识来判断是否异步暴露服务(异步暴露服务是指在另外的线程执行不阻塞当前线程)。下面我们看到主要的服务暴露代码org.apache.dubbo.config.ServiceConfig#export
方法:
/**
*
* 服务暴露
**/
public synchronized void export() {
//...
//是否延迟暴露判断
if (shouldDelay()) {
//延迟暴露
//使用延时调度器进行服务暴露
DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS);
} else {
//服务暴露
doExport();
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
进入doExport()
protected synchronized void doExport() {
if (unexported) {
throw new IllegalStateException("The service " + interfaceClass.getName() + " has already unexported!");
}
if (exported) {
return;
}
exported = true;
if (StringUtils.isEmpty(path)) {
path = interfaceName;
}
doExportUrls();
}
2
3
4
5
6
7
8
9
10
11
12
13
然后对每个协议的服务进行导出
private void doExportUrls() {
// 加载注册中心链接
List<URL> registryURLs = loadRegistries(true);
// 遍历 protocols,并在每个协议下导出服务
for (ProtocolConfig protocolConfig : protocols) {
doExportUrlsFor1Protocol(protocolConfig, registryURLs);
}
}
2
3
4
5
6
7
8
# 1.5 解析URL
走到这一步url大概是这个样子: registry://192.168.22.130:2181/org.apache.dubbo.registry.RegistryService/…… 这一步是根据配置信息组装URL, 而且URL驱动流程的执行,会再各个模块之间一直往下传,后续会不断修改URL头或协议地址,并根据URL地址中的信息做相应的处理; 在doExportUrlsFor1Protocol方法中做了URL的组装处理,通过反射的方式获取到版本、时间戳、方法名以及各种配置对象的字段信息,然后放入map,源码太长了,这块只贴一下伪代码帮助理解:
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
// 获取 ArgumentConfig 列表
for (遍历 ArgumentConfig 列表) {
if (type 不为 null,也不为空串) { // 分支1
1. 通过反射获取 interfaceClass 的方法列表
for (遍历方法列表) {
1. 比对方法名,查找目标方法
2. 通过反射获取目标方法的参数类型数组 argtypes
if (index != -1) { // 分支2
1. 从 argtypes 数组中获取下标 index 处的元素 argType
2. 检测 argType 的名称与 ArgumentConfig 中的 type 属性是否一致
3. 添加 ArgumentConfig 字段信息到 map 中,或抛出异常
} else { // 分支3
1. 遍历参数类型数组 argtypes,查找 argument.type 类型的参数
2. 添加 ArgumentConfig 字段信息到 map 中
}
}
} else if (index != -1) { // 分支4
1. 添加 ArgumentConfig 字段信息到 map 中
}
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 2. 导出Dubbo服务
准备工作做完了,接下来进行服务导出。服务导出分为导出到本地JVM和导出到远程。 代码根据 url 中的 scope 参数决定服务导出方式,分别如下:
- scope = none,不导出服务
- scope != remote,导出到本地
- scope != local,导出到远程
服务发布的本质就是把export的每个服务转为一个对应的Invoker可执行体,然后把转换后的Invoker都放到一个exporterMap(key,invoker)集合中
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
// 省略无关代码
if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.hasExtension(url.getProtocol())) {
// 加载 ConfiguratorFactory,并生成 Configurator 实例,然后通过实例配置 url
url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.getExtension(url.getProtocol()).getConfigurator(url).configure(url);
}
String scope = url.getParameter(Constants.SCOPE_KEY);
// 如果 scope = none,则什么都不做
if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
// scope != remote,导出到本地
if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
exportLocal(url);
}
// scope != local,导出到远程
if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
if (registryURLs != null && !registryURLs.isEmpty()) {
for (URL registryURL : registryURLs) {
url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY));
// 加载监视器链接
URL monitorUrl = loadMonitor(registryURL);
if (monitorUrl != null) {
// 将监视器链接作为参数添加到 url 中
url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
}
String proxy = url.getParameter(Constants.PROXY_KEY);
if (StringUtils.isNotEmpty(proxy)) {
registryURL = registryURL.addParameter(Constants.PROXY_KEY, proxy);
}
// 为服务提供类(ref)生成 Invoker
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
// DelegateProviderMetaDataInvoker 用于持有 Invoker 和 ServiceConfig
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
// 导出服务,并生成 Exporter
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
}
// 不存在注册中心,仅导出服务
} else {
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
}
}
}
this.urls.add(url);
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
不管是导出到本地,还是远程。进行服务导出之前,均需要先创建 Invoker,这是一个很重要的步骤。因此下面先来分析 Invoker 的创建过程
# 2.1 Invoker创建过程
Invoker是一个重要模型,在服务的提供端和调用端都会出现Invoker
Invoker 是实体域,它是 Dubbo 的核心模型,其它模型都向它靠扰,或转换成它,它代表一个可执行体,可向它发起 invoke 调用。
它有可能是一个本地的实现,也可能是一个远程的实现,也可能一个集群实现
Invoker由ProxyFactory创建而来,Dubbo默认的ProxyFactory实现类是JavassistProxyFactory。来看一下JavassistProxyFactory类创建Invoker的过程
public class JavassistProxyFactory extends AbstractProxyFactory {
@Override
@SuppressWarnings("unchecked")
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// 为目标类创建 Wrapper
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
// 创建匿名 Invoker 类对象,并实现 doInvoke 方法 返回
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
doInvoke
重写后的doInvoke逻辑比较简单,仅仅是将调用请求转发给了Wrapper类的invokeMethod,Wrapper用于“包裹”目标类,Wrapper是一个抽象类,仅可通过getWrapper方法传入的Class对象进行解析,拿到类的信息,生成invokeMethod方法和其他方法代码,代码生成完毕之后,通过javassist生成Class对象,最后再通过反射创建Wrapper实例。
public static Wrapper getWrapper(Class<?> c) {
while (ClassGenerator.isDynamicClass(c))
c = c.getSuperclass();
if (c == Object.class)
return OBJECT_WRAPPER;
// 从缓存中获取 Wrapper 实例
Wrapper ret = WRAPPER_MAP.get(c);
if (ret == null) {
// 缓存未命中,创建 Wrapper
ret = makeWrapper(c);
// 写入缓存
WRAPPER_MAP.put(c, ret);
}
return ret;
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 2.2 导出服务到本地
private void exportLocal(URL url) {
// 如果 URL 的协议头等于 injvm,说明已经导出到本地了,无需再次导出
if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
URL local = URL.valueOf(url.toFullString())
.setProtocol(Constants.LOCAL_PROTOCOL) // 设置协议头为 injvm
.setHost(LOCALHOST)
.setPort(0);
ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));
// 创建 Invoker,并导出服务,这里的 protocol 会在运行时调用InjvmProtocol的export方法
Exporter<?> exporter = protocol.export(
proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
exporters.add(exporter);
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
提示
exportLocal 方法比较简单,首先根据 URL 协议头决定是否导出服务。若需导出,则创建一个新的 URL 并将协议头、主机名以及端口设置成新的值。然后创建 Invoker,并调用 InjvmProtocol 的 export 方法导出服务。下面我们来看一下 InjvmProtocol 的 export 方法都做了哪些事情
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
// 创建 InjvmExporter
return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
}
2
3
4
如上,InjvmProtocol 的 export 方法仅创建了一个 InjvmExporter,无其他逻辑。到此导出服务到本地就分析完了,接下来,我们继续分析导出服务到远程的过程。
# 2.3 导出服务到远程
与导出服务到本地相比,导出服务到远程包括了服务导出与服务注册两个过程;
# 2.3.1 服务导出
RegistryProtocol
的 export
方法
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
// 导出服务
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
// 获取注册中心 URL,以 zookeeper 注册中心为例,得到的示例 URL 如下:
// zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.2&export=dubbo%3A%2F%2F172.17.48.52%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider
URL registryUrl = getRegistryUrl(originInvoker);
// 根据 URL 加载 Registry 实现类,比如 ZookeeperRegistry
final Registry registry = getRegistry(originInvoker);
// 获取已注册的服务提供者 URL,比如:
// dubbo://172.17.48.52:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello
final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);
// 获取 register 参数
boolean register = registeredProviderUrl.getParameter("register", true);
// 向服务提供者与消费者注册表中注册服务提供者
ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);
// 根据 register 的值决定是否注册服务
if (register) {
// 向注册中心注册服务
register(registryUrl, registeredProviderUrl);
ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
}
// 获取订阅 URL,比如:
// provider://172.17.48.52:20880/com.alibaba.dubbo.demo.DemoService?category=configurators&check=false&anyhost=true&application=demo-provider&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
// 创建监听器
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
// 向注册中心进行订阅 override 数据
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
// 创建并返回 DestroyableExporter
return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 主要逻辑
- 调用doLocalExport 导出服务
- 向注册中心注册服务
- 向注册中心订阅override数据
- 创建并返回DestroyableExporter
# 2.3.1.1 第一步 doLocalExport
老版本代码使用典型的双重检查锁机制来创建exporter
新版本代码使用ConcurrentHashMap中的线程安全的computeIfAbsent方法
来保证创建的线程安全
// 老版本代码
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
String key = getCacheKey(originInvoker);
// 访问缓存
ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
synchronized (bounds) {
exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
// 创建 Invoker 为委托类对象
final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
// 核心代码在这里:
//调用 protocol 的 export 方法导出服务
exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
// 写缓存
bounds.put(key, exporter);
}
}
}
return exporter;
}
// 新代码
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
// 获取缓存key
String key = getCacheKey(originInvoker);
// computeIfAbsent方法,如果缓存中不存在,就放入缓存
// bounds是类型是ConcurrentHashMap
return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {
// 这个方法是线程安全的 取代了原来的双重检测锁方式
// 创建 Invoker 为委托类对象
Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
//调用 protocol 的 export 方法导出服务
return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);
});
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
重点在于protocol.export
这部分,假定运行时协议为dubbo(默认),此处protocol变量在运行时候将会加载DubboProtocol,并调用DubboProtocol类的export方法,下面来分析DubboProtocol的export方法:
# DubboProtocol类中的export方法
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// 获取服务标识,理解成服务坐标也行。由服务组名,服务名,服务版本号以及端口组成。比如:
// demoGroup/com.alibaba.dubbo.demo.DemoService:1.0.1:20880
String key = serviceKey(url);
// 创建 DubboExporter
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
// 将 <key, exporter> 键值对放入缓存中
exporterMap.put(key, exporter);
// 本地存根相关代码
Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
if (isStubSupportEvent && !isCallbackservice) {
String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
// 省略日志打印代码
}
}
// 核心方法:-- 启动服务器
openServer(url);
// 优化序列化
optimizeSerialization(url);
return exporter;
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 2.3.1.2 启动服务器
private void openServer(URL url) {
// 获取 host:port,并将其作为服务器实例的 key,用于标识当前的服务器实例
String key = url.getAddress();
boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
if (isServer) {
// 访问缓存
ProtocolServer server = serverMap.get(key);
// 在同一台机器上(单网卡),同一个端口上仅允许启动一个服务器实例。
// 若某个端口上已有服务器实例,此时则调用 reset 方法重置服务器的一些配置
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) {
// 核心方法:创建服务器实例
serverMap.put(key, createServer(url));
}
}
} else {
// 服务器已创建,则根据 url 中的配置重置服务器
server.reset(url);
}
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 2.3.1.3 创建服务器
/*
createServer 包含三个核心的逻辑
1.检测是否存在 server 参数所代表的 Transporter 拓展,不存在则抛出异常
2.创建服务器实例
3.检测是否支持 client 参数所表示的 Transporter 拓展,不存在也是抛出异常
*/
private ProtocolServer createServer(URL url) {
url = URLBuilder.from(url)
// send readonly event when server closes, it's enabled by default
.addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
// 添加心跳检测配置到 url 中
.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
// 添加编码解码器参数
.addParameter(CODEC_KEY, DubboCodec.NAME)
.build();
// 获取 server 参数,默认为 netty
String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);
// 通过 SPI 检测是否存在 server 参数所代表的 Transporter 拓展,不存在则抛出异常
if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
throw new RpcException("Unsupported server type: " + str + ", url: " + url);
}
ExchangeServer server;
try {
// 核心方法--创建服务器 ExchangeServer
server = Exchangers.bind(url, requestHandler);
} catch (RemotingException e) {
throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
}
// 获取 client 参数,可指定 netty,mina
str = url.getParameter(CLIENT_KEY);
if (str != null && str.length() > 0) {
// 获取所有的 Transporter 实现类名称集合,比如 supportedTypes = [netty, mina]
Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
// 检测当前 Dubbo 所支持的 Transporter 实现类名称列表中,
// 是否包含 client 所表示的 Transporter,若不包含,则抛出异常
if (!supportedTypes.contains(str)) {
throw new RpcException("Unsupported client type: " + str);
}
}
return new DubboProtocolServer(server);
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# Exchangers.bind(URL url, ExchangeHandler handler)
public static ExchangeServer bind(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");
// 获取 Exchanger,默认为 HeaderExchanger。
// 紧接着调用 HeaderExchanger 的 bind 方法创建 ExchangeServer 实例
return getExchanger(url).bind(url, handler);
}
// HeaderExchanger 的 bind 方法
// 创建 ExchangeServer 实例
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
// 创建 HeaderExchangeServer 实例,该方法包含了多个逻辑,分别如下:
// 1. new HeaderExchangeHandler(handler)
// 2. new DecodeHandler(new HeaderExchangeHandler(handler))
// 3. Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
// Transporters 的 bind 方法
public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handlers == null || handlers.length == 0) {
throw new IllegalArgumentException("handlers == null");
}
ChannelHandler handler;
if (handlers.length == 1) {
handler = handlers[0];
} else {
// 如果 handlers 元素数量大于1,则创建 ChannelHandler 分发器
handler = new ChannelHandlerDispatcher(handlers);
}
// 获取自适应 Transporter 实例,并调用实例方法
// getTransporter() 方法获取的 Transporter 是在运行时动态创建的,
// 类名为 TransporterAdaptive--自适应扩展类,
// 它会根据运行时传入的url参数决定加载什么类型的transporter,默认为NettyTransporter
return getTransporter().bind(url, handler);
}
// NettyTransporter 的 bind 方法
// 还有其他的比如 Netty4 mina Grizzly的
public Server bind(URL url, ChannelHandler listener) throws RemotingException {
// 创建 NettyServer
return new NettyServer(url, listener);
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
创建NettyServer后会调用模板类中的 doOpen() 启动服务器,该方法需要子类具体来实现
@Override
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
// https://issues.jboss.org/browse/NETTY-365
// https://issues.jboss.org/browse/NETTY-379
// final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
bootstrap.setOption("child.tcpNoDelay", true);
bootstrap.setOption("backlog", getUrl().getPositiveParameter(BACKLOG_KEY, Constants.DEFAULT_BACKLOG));
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout();
if (idleTimeout > 10000) {
pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0));
}*/
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
// bind
channel = bootstrap.bind(getBindAddress());
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
服务导出的过程到这里分析就结束了
# 2.3.2 服务注册
提示
服务注册的过程相对于dubbo来说不是必须的,通过服务直连的方式就可以绕过注册中心。但通常我们不会这么做,直连方式不利于服务治理,仅在测试时使用,注册中心对于Dubbo来说虽然不是必须的,但确实必要的,下面主要将zk作为分析目标。
服务注册的入口方法还在RegistryProtocol 的 export()
上,如下:
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
// ${省略导出服务}
// 省略其他代码
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
//export invoker
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
// url to registry
final Registry registry = getRegistry(originInvoker);
final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);boolean register = registeredProviderUrl.getParameter("register", true);
if (register) {
// 注册服务
register(registryUrl, registeredProviderUrl);
ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
}
// register stated url on provider model
registerStatedUrl(registryUrl, registeredProviderUrl, register);
exporter.setRegisterUrl(registeredProviderUrl);
exporter.setSubscribeUrl(overrideSubscribeUrl);
// 订阅 override 数据
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
// 省略部分代码
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
RegistryProtocol的export()方法包含了服务导出,注册,以及数据订阅等逻辑。其中服务导出上面已经分析过了,这里主要分析注册的逻辑,入口方法是register()
;
public void register(URL registryUrl, URL registedProviderUrl) {
// 获取 Registry实例
Registry registry = registryFactory.getRegistry(registryUrl);
// 向注册中心注册服务
registry.register(registedProviderUrl);
}
2
3
4
5
6
# 2.3.2.1 创建注册中心
Dubbo支持的注册中心有zk,Nacos,Redis,Multicast,Simple等,目前官方较为推荐的是zk,这里以zk为例进行分析, 抽象类AbstractRegistryFactory 中提供了getRegistry方法,各个子类可以对其实现,主要关注zk的实现。
AbstractRegistryFactory.getRegistry
方法
public Registry getRegistry(URL url) {
url = URLBuilder.from(url)
.setPath(RegistryService.class.getName())
.addParameter(INTERFACE_KEY, RegistryService.class.getName())
.removeParameters(EXPORT_KEY, REFER_KEY)
.build();
String key = createRegistryCacheKey(url);
LOCK.lock();
try {
// 访问缓存
Registry registry = REGISTRIES.get(key);
if (registry != null) {
return registry;
}
// 缓存未命中,创建 Registry 实例
registry = createRegistry(url);
if (registry == null) {
throw new IllegalStateException("Can not create registry " + url);
}
// 写入缓存
REGISTRIES.put(key, registry);
return registry;
} finally {
LOCK.unlock();
}
}
protected abstract Registry createRegistry(URL url);
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
提示
getRegistry方法先访问缓存,缓存未命中则调用creteRegistry创建Registry,然后写入缓存,createRegistry是一个模板方法,由具体的子类实现,下面正式进入ZookeeperRegistryFactory
的实现
public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
// zookeeperTransporter 由 SPI 在运行时注入,类型为 ZookeeperTransporter$Adaptive
private ZookeeperTransporter zookeeperTransporter;
public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
this.zookeeperTransporter = zookeeperTransporter;
}
@Override
public Registry createRegistry(URL url) {
// 创建 ZookeeperRegistry
return new ZookeeperRegistry(url, zookeeperTransporter);
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
实例化一个ZookeeperRegistry:
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
if (url.isAnyHost()) {
throw new IllegalStateException("registry address == null");
}
// 获取组名,默认为 dubbo
String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
if (!group.startsWith(PATH_SEPARATOR)) {
// group = "/" + group
group = PATH_SEPARATOR + group;
}
this.root = group;
// 创建 Zookeeper 客户端,默认为 CuratorZookeeperTransporter
// connect方法创建zkclient的客户端 这里需要重点关注
zkClient = zookeeperTransporter.connect(url);
// 添加状态监听器 StateListener接口实现类
zkClient.addStateListener((state) -> {
if (state == StateListener.RECONNECTED) {
// 省略日志打印
ZookeeperRegistry.this.fetchLatestAddresses();
} else if (state == StateListener.NEW_SESSION_CREATED) {
// 省略日志打印
try {
ZookeeperRegistry.this.recover();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
} else if (state == StateListener.SESSION_LOST) {
// 省略日志打印
} else if (state == StateListener.SUSPENDED) {
} else if (state == StateListener.CONNECTED) {
}
});
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
zookeeperTransporter类型为自适应扩展类,因此connect方法会再被调用时决定加载什么类型的zookeeperTransporter扩展,默认为CuratorZookeeperTransporter
public class CuratorZookeeperTransporter extends AbstractZookeeperTransporter {
@Override
public ZookeeperClient createZookeeperClient(URL url) {
return new CuratorZookeeperClient(url);
}
}
2
3
4
5
6
public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorWatcher> {
private final CuratorFramework client;
//CuratorZookeeperClient构造方法用于创建和启动CuratorFramework 实例
public CuratorZookeeperClient(URL url) {
super(url);
try {
int timeout = url.getParameter(TIMEOUT_KEY, DEFAULT_CONNECTION_TIMEOUT_MS);
int sessionExpireMs = url.getParameter(ZK_SESSION_EXPIRE_KEY, DEFAULT_SESSION_TIMEOUT_MS);
// 创建 CuratorFramework 构造器
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
.connectString(url.getBackupAddress())
.retryPolicy(new RetryNTimes(1, 1000))
.connectionTimeoutMs(timeout)
.sessionTimeoutMs(sessionExpireMs);
String authority = url.getAuthority();
if (authority != null && authority.length() > 0) {
builder = builder.authorization("digest", authority.getBytes());
}
// 构建 CuratorFramework 实例
client = builder.build();
// 添加监听器
client.getConnectionStateListenable().addListener(new CuratorConnectionStateListener(url));
// 启动客户端
client.start();
boolean connected = client.blockUntilConnected(timeout, TimeUnit.MILLISECONDS);
if (!connected) {
throw new IllegalStateException("zookeeper not connected");
}
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 2.3.2.2 节点创建
以Zookeeper为例,服务注册的本质是将服务配置数据写入到zk某个路径的节点下。可以通过zk的可视化客户端ZooInsepector查看节点数据
提示
DemoService服务对应的配置信息(存储在URL中)最终被注册到providers节点下,如果多台机器上都部署了这个服务,当前这个providers下面就会有多个url;
下面分析注册的过程: 服务注册的接口为register(URL), 这个方法定义在FailbackRegistry抽象类中
public void register(URL url) {
// 校验URL是否满足要求
if (!acceptable(url)) {
logger.info("URL " + url + " will not be registered to Registry. Registry " + url + " does not accept service of this protocol type.");
return;
}
super.register(url);
removeFailedRegistered(url);
removeFailedUnregistered(url);
try {
// 调用模板方法,由子类实现
doRegister(url);
} catch (Exception e) {
Throwable t = e;
// 获取 check 参数,若 check = true 将会直接抛出异常
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true)
&& !CONSUMER_PROTOCOL.equals(url.getProtocol());
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if (skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
// 记录注册失败的链接
addFailedRegistered(url);
}
}
// 模板方法,子类实现,重点关注 FailbackRegistry 子类 ZookeeperRegistry 的实现
protected abstract void doRegister(URL url);
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
ZookeeperRegistry.doRegister()
protected void doRegister(URL url) {
try {
// 通过 Zookeeper 客户端创建节点,节点路径由 toUrlPath 方法生成,路径格式如下:
// /${group}/${serviceInterface}/providers/${url}
// 比如
// /dubbo/org.apache.dubbo.DemoService/providers/dubbo%3A%2F%2F127.0.0.1......
zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
2
3
4
5
6
7
8
9
10
11
服务注册到这里就结束了,整个过程可简单总结为: 先创建注册中心实例,然后在通过注册中心实例注册服务。