Dubbo 30 服务注册

发表于

这是我参与8月更文挑战的第5天，活动详情查看：8月更文挑战

首先分享之前的所有文章 , 欢迎点赞收藏转发三连下次一定 >>>> 😜😜😜

文章合集 : 🎁 juejin.cn/post/694164…

Github : 👉 github.com/black-ant

CASE 备份 : 👉 gitee.com/antblack/ca…

一 . 前言

服务发现的作用主要范围包括向注册中心注册服务 , 以及服务的发现 (Feign 的负载均衡是在 Client 端做的 , 猜测Dubbo负载均衡也是)

Dubbo 提供多种注册中心 , 常见的有 Nacos , Zookeeper 和 Redis , 此篇也只看看3种

二 . 服务发现流程

2.1 服务发现的调用逻辑

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码// 服务发现的起点还是 BootStrap , 由 Listener 发起流程

C- DubboBootstrapApplicationListener # onContextRefreshedEvent
C- DubboBootstrap # start : 开始初始化
C- DubboBootstrap # exportServices : 注册 DubboService
C- ServiceConfig # export : 调用init 执行初始化 , 校验 config , 设置 serviceMetadata 元数据
C- ServiceConfig # doExport : 没太多逻辑 , 下层调用而已
C- ServiceConfig # doExportUrls : ServiceRepository 进行保存和下层调用
C- ServiceConfig # doExportUrlsFor1Protocol
C- ProtocolFilterWrapper # export
C- ProtocolListenerWrapper # export
C- RegistryProtocol # export : 发起注册逻辑

Zookeeper Service 详情

1
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
json复制代码{
	"name": "dubbo-demo-annotation-provider",
	"id": "10.10.100.140:20880",
	"address": "10.10.100.140",
	"port": 20880,
	"sslPort": null,
	"payload": {
		"@class": "org.apache.dubbo.registry.zookeeper.ZookeeperInstance",
		"id": null,
		"name": "dubbo-demo-annotation-provider",
		"metadata": {
			"REGISTRY_CLUSTER": "registryConfig",
			"anyhost": "true",
			"application": "dubbo-demo-annotation-provider",
			"deprecated": "false",
			"dubbo": "2.0.2",
			"dubbo.endpoints": "[{\"port\":20880,\"protocol\":\"dubbo\"}]",
			"dubbo.metadata-service.url-params": "{\"version\":\"1.0.0\",\"dubbo\":\"2.0.2\",\"port\":\"20880\",\"protocol\":\"dubbo\"}",
			"dubbo.metadata.revision": "3c1be7b82505c5da23a5bc1cd1211417",
			"dubbo.metadata.storage-type": "local",
			"dynamic": "true",
			"generic": "false",
			"interface": "org.apache.dubbo.demo.DemoService",
			"metadata-type": "remote",
			"methods": "sayHello,sayHelloAsync",
			"pid": "21576",
			"release": "",
			"side": "provider",
			"timestamp": "1626856228483"
		}
	},
	"registrationTimeUTC": 1626856236245,
	"serviceType": "DYNAMIC",
	"uriSpec": null
}

Zookeeper 中数据情况

Dubbo-zookeeper-service.png

2.2 调用的起点

1
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
JAVA复制代码C- DubboBootstrap # start
C- DubboBootstrap # exportServices : start 逻辑中发起 export 处理

private void exportServices() {
    // Step 1 : configManager.getServices() 获取 Service 列表 -> PRO00001
    for (ServiceConfigBase sc : configManager.getServices()) {
    
        // 获取当前 ServiceConfig , 并且为其设置容器
        ServiceConfig<?> serviceConfig = (ServiceConfig<?>) sc;
        serviceConfig.setBootstrap(this);
        if (!serviceConfig.isRefreshed()) {
            // 需要刷新时对其进行刷新
            serviceConfig.refresh();
        }
        
        // 需要异步处理 -> export-async
        if (sc.shouldExportAsync()) {
        
            // 明显看到这里使用 ScheduledExecutorService + future 发起了异步调用
            ExecutorService executor = executorRepository.getExportReferExecutor();
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                try {
                    if (!sc.isExported()) {
                        sc.export();
                        exportedServices.add(sc);
                    }
                } catch (Throwable t) {
                    logger.error("export async catch error : " + t.getMessage(), t);
                }
            }, executor);

            asyncExportingFutures.add(future);
        } else {
            // 这里逻辑都是一样的 , 没有导出过的进行导出 ,同时添加到对应集合中
            if (!sc.isExported()) {
                sc.export();
                exportedServices.add(sc);
            }
        }
    }
}


// PRO00001 : configManager.getServices 列表
<dubbo:service path="org.apache.dubbo.demo.DemoService" ref="org.apache.dubbo.demo.provider.DemoServiceImpl@60990e5c" generic="false" deprecated="false" dynamic="true" id="ServiceBean:org.apache.dubbo.demo.DemoService" />

补充 configManager

该方法从configsCache 中获取对应的参数 ,此处取得是 service

ConfigsCache.png

添加的流程 :

  • C- AbstractConfig # addIntoConfigManager -> org.apache.dubbo.config.spring.ServiceBean
  • C- ConfigManager # addConfig : 添加 Config , 用于前文循环时使用

可以看到 , 第一步时使用的就是 serverBean , 这个Bean 在扫描时创建 , 详见 -> Dubbo 3.0 : DubboService 的扫描

2.3 解析出 Service

中间没有太多东西 , 主要调用流程为 :

  • C- ServiceConfig # export :
  • C- ServiceConfig # doExport :
    • C- ServiceConfig # doExportUrls :
    • C- ServiceConfig # exported :
1
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
java复制代码// 在此阶段 , 构建出 Registry URL , 用于 Zookeeper 注册
private void doExportUrls() {
    
    // PRO0001 : 
    ServiceRepository repository = ApplicationModel.getServiceRepository();
    // PRO0002 : 
    ServiceDescriptor serviceDescriptor = repository.registerService(getInterfaceClass());
    repository.registerProvider(
            getUniqueServiceName(),
            ref,
            serviceDescriptor,
            this,
            serviceMetadata
    );
	
    // PRO0003 : 
    List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this, true);

    // PRO0004 : 
    for (ProtocolConfig protocolConfig : protocols) {
        // org.apache.dubbo.demo.DemoService
        String pathKey = URL.buildKey(getContextPath(protocolConfig)
                .map(p -> p + "/" + path)
                .orElse(path), group, version);
        // 如果用户指定了路径，则再次注册服务，将其映射到路径
        repository.registerService(pathKey, interfaceClass);
        doExportUrlsFor1Protocol(protocolConfig, registryURLs);
    }
}

// PRO0001 : ServiceRepository 结构
{
    "services": {
        "serviceName": "org.apache.dubbo.rpc.service.EchoService",
        "serviceName": "org.apache.dubbo.rpc.service.GenericService",
        "serviceName": "org.apache.dubbo.monitor.MetricsService",
        "serviceName": "org.apache.dubbo.monitor.MonitorService",
        "serviceName": "org.apache.dubbo.demo.DemoService",
        
    },
    "providers": {
        "org.apache.dubbo.demo.DemoService": "ProviderModel"
    },
    "providersWithoutGroup": "org.apache.dubbo.demo.DemoService:0.0.0"
}
    
    


// PRO0003 : 构建 RegistryURL 
service-discovery-registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?REGISTRY_CLUSTER=registryConfig&application=dubbo-demo-annotation-provider&dubbo=2.0.2&pid=28256&registry=zookeeper&timestamp=1626858537404

registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?REGISTRY_CLUSTER=registryConfig&application=dubbo-demo-annotation-provider&dubbo=2.0.2&pid=28256&registry=zookeeper&timestamp=1626858537404

2.4 执行主流程

其中主步骤分为 3 步 :

  • Step 1 : 属性准备
  • Step 2 : 获取 host port , 构建 url
  • Step 3 : 构建 Invoke , export 导出
1
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
java复制代码private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
    String name = protocolConfig.getName();
    if (StringUtils.isEmpty(name)) {
        name = DUBBO;
    }

    // Step 1 : 属性准备
    Map<String, String> map = new HashMap<String, String>();\
    // String PROVIDER_SIDE = "provider";
    // String SIDE_KEY = "side";
    map.put(SIDE_KEY, PROVIDER_SIDE);
    
    // 此环节添加属性
    ServiceConfig.appendRuntimeParameters(map);
    AbstractConfig.appendParameters(map, getMetrics());
    AbstractConfig.appendParameters(map, getApplication());
    AbstractConfig.appendParameters(map, getModule());
    // remove 'default.' prefix for configs from ProviderConfig
    // appendParameters(map, provider, Constants.DEFAULT_KEY);
    AbstractConfig.appendParameters(map, provider);
    AbstractConfig.appendParameters(map, protocolConfig);
    AbstractConfig.appendParameters(map, this);
    
    MetadataReportConfig metadataReportConfig = getMetadataReportConfig();
    if (metadataReportConfig != null && metadataReportConfig.isValid()) {
        map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);
    }
    if (CollectionUtils.isNotEmpty(getMethods())) {
        //TODO Improve method config processing
        //... 此处如果 dubbo:method 进行了方法级处理 , 此处会深入处理 , 同样是添加 method 处理
    }

    if (ProtocolUtils.isGeneric(generic)) {
        map.put(GENERIC_KEY, generic);
        map.put(METHODS_KEY, ANY_VALUE);
    } else {
        String revision = Version.getVersion(interfaceClass, version);
        if (revision != null && revision.length() > 0) {
            map.put(REVISION_KEY, revision);
        }

        String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
        // ["sayHello","sayHelloAsync"]
        if (methods.length == 0) {
            // 省略为空逻辑
        } else {
            map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
        }
    }

    // 提供程序配置的令牌值用于将值分配给ServiceConfig#令牌
    if (ConfigUtils.isEmpty(token) && provider != null) {
        token = provider.getToken();
    }

    if (!ConfigUtils.isEmpty(token)) {
        if (ConfigUtils.isDefault(token)) {
            map.put(TOKEN_KEY, UUID.randomUUID().toString());
        } else {
            map.put(TOKEN_KEY, token);
        }
    }
    // 讲所有的 map 放入 元数据中
    serviceMetadata.getAttachments().putAll(map);

    // Step 2 : 获取 host port , 构建 url
    String host = findConfigedHosts(protocolConfig, registryURLs, map);
    Integer port = findConfigedPorts(protocolConfig, name, map);
    // PRO24001 -> url 
    URL url = new ServiceConfigURL(name, null, null, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);

    // 自定义Configurator以附加额外参数
    if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
            .hasExtension(url.getProtocol())) {
        url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
                .getExtension(url.getProtocol()).getConfigurator(url).configure(url);
    }

    String scope = url.getParameter(SCOPE_KEY);
    // don't export when none is configured
    if (!SCOPE_NONE.equalsIgnoreCase(scope)) {

        // 如果配置不是远程的，则导出到本地
        if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
            exportLocal(url);
        }
        // 如果配置不是本地的，则导出到远程
        if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
            if (CollectionUtils.isNotEmpty(registryURLs)) {
                // PIC24002 -> registryURLs
                for (URL registryURL : registryURLs) {
                    //if protocol is only injvm ,not register
                    if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
                        continue;
                    }
                    url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
                    URL monitorUrl = ConfigValidationUtils.loadMonitor(this, registryURL);
                    if (monitorUrl != null) {
                        url = url.putAttribute(MONITOR_KEY, monitorUrl);
                    }

                    // For providers, this is used to enable custom proxy to generate invoker
                    String proxy = url.getParameter(PROXY_KEY);
                    if (StringUtils.isNotEmpty(proxy)) {
                        registryURL = registryURL.addParameter(PROXY_KEY, proxy);
                    }

                    // Step 3 : 构建 Invoke ,  export 导出
                    Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.putAttribute(EXPORT_KEY, url));
                    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
                    // -> 2.5 Protocol 注册
                    Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);
                    exporters.add(exporter);
                }
            } else {
                if (MetadataService.class.getName().equals(url.getServiceInterface())) {
                    MetadataUtils.saveMetadataURL(url);
                }
                Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);
                DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

                Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);
                exporters.add(exporter);
            }

            MetadataUtils.publishServiceDefinition(url);
        }
    }
    this.urls.add(url);
}

PRO24001 案例 : providerUrl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
java复制代码dubbo://192.168.181.2:20880/org.apache.dubbo.demo.DemoService
    ?anyhost=true
    &application=dubbo-demo-annotation-provider
    &bind.ip=192.168.181.2
    &bind.port=20880
    &deprecated=false
    &dubbo=2.0.2
    &dynamic=true
    &generic=false
    &interface=org.apache.dubbo.demo.DemoService
    &metadata-type=remote
    &methods=sayHello,sayHelloAsync
    &pid=17324
    &release=
    &side=provider
    &timestamp=1628690334012

PRO24002 案例 : registryUrl

1
2
3
4
5
6
7
java复制代码service-discovery-registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService
?REGISTRY_CLUSTER=registryConfig
&application=dubbo-demo-annotation-provider
&dubbo=2.0.2
&pid=17324
&registry=zookeeper
&timestamp=1628690331733

PIC24002
image.png

2.5 传入代理类

在这一步构建了 url ,同时生成了 ServiceDiscoveryRegistry

1
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
59
60
61
62
63
64
65
JAVA复制代码C- RegistryProtocol
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {

    // service-discovery-registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService....
    URL registryUrl = getRegistryUrl(originInvoker);
    // dubbo://192.168.181.2:20880/org.apache.dubbo.demo.DemoService?anyhost=true....
    URL providerUrl = getProviderUrl(originInvoker);

    // PRO0003 : URL 参数
    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
    
    // 重写 URL
    providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
    //export invoker
    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);

    // Step 3 :  通过 Registry Factory 获取使用的 Registry , 例如 : ZookeeperRegistry -> 3.1 
    final Registry registry = getRegistry(registryUrl);
    // Step 4 :  构建注册对象 -> PRO0003
    final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);

    // Step 5 :  注册发布服务
    boolean register = providerUrl.getParameter(REGISTER_KEY, true);
    if (register) **{**
        // -> 2.5.2 Registry 的管理
        register(registry, registeredProviderUrl);
    }

    // 在提供程序模型上注册声明的url
    registerStatedUrl(registryUrl, registeredProviderUrl, register);


    exporter.setRegisterUrl(registeredProviderUrl);
    exporter.setSubscribeUrl(overrideSubscribeUrl);

    // Deprecated! Subscribe to override rules in 2.6.x or before.
    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

    notifyExport(exporter);
    //Ensure that a new exporter instance is returned every time export
    return new DestroyableExporter<>(exporter);
}


// PRO0003 : URL 参数
provider://192.168.181.2:20880/org.apache.dubbo.demo.DemoService
?anyhost=true
&application=dubbo-demo-annotation-provider
&bind.ip=192.168.181.2
&bind.port=20880
&category=configurators
&check=false
&deprecated=false
&dubbo=2.0.2
&dynamic=true
&generic=false
&interface=org.apache.dubbo.demo.DemoService
&metadata-type=remote
&methods=sayHello,sayHelloAsync
&pid=17324
&release=
&side=provider
&timestamp=1628690334012

Registry 可以分为以下几个部分 :

  • 由 getRegistry 发起获取 Registry
  • createRegistry 发起创建逻辑
  • RegistryProtocol 调用 doRegister 执行 Register

三 . Registry 的创建

共用流程

1
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
java复制代码// Step 1 : 获取 Registry
C- RegistryFactoryWrapper
public Registry getRegistry(URL url) {
	return new ListenerRegistryWrapper(registryFactory.getRegistry(url),
                Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(RegistryServiceListener.class)
                        .getActivateExtension(url, "registry.listeners")));
}

// Step 2 : 调用工厂创建 registry
C- AbstractRegistryFactory
public Registry getRegistry(URL url) {

    Registry defaultNopRegistry = getDefaultNopRegistryIfDestroyed();
    if (null != defaultNopRegistry) {
        return defaultNopRegistry;
    }

    url = URLBuilder.from(url)
            .setPath(RegistryService.class.getName())
            .addParameter(INTERFACE_KEY, RegistryService.class.getName())
            .removeParameters(EXPORT_KEY, REFER_KEY, TIMESTAMP_KEY)
            .build();
    String key = createRegistryCacheKey(url);
    // 上锁
    LOCK.lock();
    try {

        //..........
        
        // Map<String, Registry> REGISTRIES = new HashMap<>();
        Registry registry = REGISTRIES.get(key);
        if (registry != null) {
            return registry;
        }
        //通过 spi/ioc 的方式获取 Registry -> org.apache.dubbo.registry.client.ServiceDiscoveryRegistry
        registry = createRegistry(url);
        
        // 放入缓存
        REGISTRIES.put(key, registry);
        return registry;
    } finally {
        // Release the lock
        LOCK.unlock();
    }
}

3.1 创建 ServiceDiscoveryRegistry

1
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
java复制代码// Step 3 : ServiceDiscoveryRegistryFactory 
protected Registry createRegistry(URL url) {
    if (UrlUtils.hasServiceDiscoveryRegistryProtocol(url)) {
        // 此处拿到的为 zookeeper
        String protocol = url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY);
        // zookeeper://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?REGISTRY_CLUSTER=registryConfig&application=dubbo-demo-annotation-provider&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=2816
        url = url.setProtocol(protocol).removeParameter(REGISTRY_KEY);
    }
        
    return new ServiceDiscoveryRegistry(url);
}

public ServiceDiscoveryRegistry(URL registryURL) {
	this.registryURL = registryURL;
	this.serviceDiscovery = createServiceDiscovery(registryURL);
	this.writableMetadataService = WritableMetadataService.getDefaultExtension();
}

// 下次调用 , 
protected ServiceDiscovery createServiceDiscovery(URL registryURL) {
    ServiceDiscovery serviceDiscovery = getServiceDiscovery(registryURL);
    execute(() -> {
        // 初始化操作
        serviceDiscovery.initialize(registryURL.addParameter(INTERFACE_KEY, ServiceDiscovery.class.getName())
                    .removeParameter(REGISTRY_TYPE_KEY));
    });
    return serviceDiscovery;
}

static WritableMetadataService getDefaultExtension() {
	return getExtensionLoader(WritableMetadataService.class).getDefaultExtension();
}

3.2 创建 ZookeeperRegistry

Zookeeper 只是 Dubbo 支持的注册中心之一 , 以其为例看一下创建和使用流程 :

这里有一个核心点 , ZookeeperRegistry 的创建是在 , 它创建时会被包裹为一个 ListenerRegistryWrapper 对象

1
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
JAVA复制代码C- ZookeeperRegistryFactory
public Registry createRegistry(URL url) {
	return new ZookeeperRegistry(url, zookeeperTransporter);
}

// 创建 ZookeeperRegistry
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("registry address == null");
        }
        String group = url.getGroup(DEFAULT_ROOT);
        if (!group.startsWith(PATH_SEPARATOR)) {
            group = PATH_SEPARATOR + group;
        }
        this.root = group;
        zkClient = zookeeperTransporter.connect(url);
        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) {

            }
        });
    }


C- ExtensionLoader 
private T createExtension(String name, boolean wrap) {
        Class<?> clazz = getExtensionClasses().get(name);
        if (clazz == null || unacceptableExceptions.contains(name)) {
            throw findException(name);
        }
        try {
            T instance = (T) EXTENSION_INSTANCES.get(clazz);
            if (instance == null) {
                EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.getDeclaredConstructor().newInstance());
                instance = (T) EXTENSION_INSTANCES.get(clazz);
            }
            injectExtension(instance);


            if (wrap) {

                List<Class<?>> wrapperClassesList = new ArrayList<>();
                if (cachedWrapperClasses != null) {
                    wrapperClassesList.addAll(cachedWrapperClasses);
                    wrapperClassesList.sort(WrapperComparator.COMPARATOR);
                    Collections.reverse(wrapperClassesList);
                }

                if (CollectionUtils.isNotEmpty(wrapperClassesList)) {
                    for (Class<?> wrapperClass : wrapperClassesList) {
                        Wrapper wrapper = wrapperClass.getAnnotation(Wrapper.class);
                        if (wrapper == null
                                || (ArrayUtils.contains(wrapper.matches(), name) && !ArrayUtils.contains(wrapper.mismatches(), name))) {
                            instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));
                        }
                    }
                }
            }

            initExtension(instance);
            return instance;
        } catch (Throwable t) {
            throw new IllegalStateException("Extension instance (name: " + name + ", class: " +
                    type + ") couldn't be instantiated: " + t.getMessage(), t);
        }
    }

补充 : ListenerRegistryWrapper
可以看到 , 这里 ListenerRegistryWrapper 中包含了一个 Registry 对象

1
2
3
4
5
6
7
java复制代码public class ListenerRegistryWrapper implements Registry {
    private static final Logger logger = LoggerFactory.getLogger(ListenerRegistryWrapper.class);

    private final Registry registry;
    private final List<RegistryServiceListener> listeners;
    
}

四 . Registry 的调用

Registry 在 RegistryProtocol # register 进行注册 , 会调用父类 FailbackRegistry , 最终调用各种的实现类

FailbackRegistry 通用处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码// 此处以 FailbackRegistry 为例 
 public void register(URL url) {
        if (!acceptable(url)) {
            return;
        }
        super.register(url);
        removeFailedRegistered(url);
        removeFailedUnregistered(url);
        try {
            // Sending a registration request to the server side
            doRegister(url);
        } catch (Exception e) {
            // 省略部分异常处理逻辑
            // Record a failed registration request to a failed list, retry regularly
            addFailedRegistered(url);
        }
    }

System-AbstractRegistry.png

4.1 ServiceDiscoveryRegistry 注册

1
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
JAVA复制代码// Step 1 : 发起 Registry
C- RegistryProtocol
private void register(Registry registry, URL registeredProviderUrl) {
	registry.register(registeredProviderUrl);
}


C- ServiceDiscoveryRegistry
public final void register(URL url) {
	if (!shouldRegister(url)) { // Should Not Register
		return;
	}
	doRegister(url);
        //--> url = addRegistryClusterKey(url); 其中调用 addRegistryClusterKey
}

// Step 2 : 获取 RegistryClusterKey
private URL addRegistryClusterKey(URL url) {
        String registryCluster = serviceDiscovery.getUrl().getParameter(REGISTRY_CLUSTER_KEY);
        if (registryCluster != null && url.getParameter(REGISTRY_CLUSTER_KEY) == null) {
            url = url.addParameter(REGISTRY_CLUSTER_KEY, registryCluster);
        }
        // dubbo://192.168.181.2:20880/org.apache.dubbo.demo.DemoService?REGISTRY_CLUSTER=registryConfig&anyhost=true&application=dubbo-demo-annotation-provider&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=org.apache.dubbo.demo.DemoService&metadata-type=remote&methods=sayHello,sayHelloAsync&pid=13200&release=&side=provider&timestamp=1626146555672
        return url;
}

// Step 3 : doRegister 进行注册
public void doRegister(URL url) {
    url = addRegistryClusterKey(url);
    // 其中主要是对 Map 进行操作
    // ConcurrentNavigableMap<String, SortedSet<URL>> exportedServiceURLs = new ConcurrentSkipListMap<>(); 
    if (writableMetadataService.exportURL(url)) {
       
    } else {
      
    }
}

4.2 ZookeeperRegistry 注册

1
2
3
4
5
6
7
java复制代码public void doRegister(URL url) {
    try {
        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);
    }
}

五 . 补充深入

5.1 服务注册器的选择

1
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
java复制代码// Dubbo 提供了 Zookeeper , Nacos 等多个服务注册器 , 这里来看一下此处是如何选择的 , 此处以 Nacos 为例

// 环节一 : 判断是否支持 Nacos ExtensionLoader
C- DubboBootstrap
private boolean isUsedRegistryAsCenter(RegistryConfig registryConfig, Supplier<Boolean> usedRegistryAsCenter,
                                           String centerType,
                                           Class<?> extensionClass) {
	final boolean supported;

	Boolean configuredValue = usedRegistryAsCenter.get();
	if (configuredValue != null) { // If configured, take its value.
		supported = configuredValue.booleanValue();
	} else {                       // Or check the extension existence
		String protocol = registryConfig.getProtocol();
		supported = supportsExtension(extensionClass, protocol);
	}


	return supported;
}

C- DubboBootstrap
private boolean supportsExtension(Class<?> extensionClass, String name) {
	if (isNotEmpty(name)) {
		ExtensionLoader extensionLoader = getExtensionLoader(extensionClass);
		return extensionLoader.hasExtension(name);
	}
	return false;
}

// 环节二 : 获取 ExtensionLoader

Nacos 请求逻辑

1
2
3
4
5
6
7
8
9
10
JAVA复制代码// 调用逻辑
C- DubboBootstrapApplicationListener # onApplicationContextEvent 
C- DubboBootstrapApplicationListener # onContextRefreshedEvent
C- DubboBootstrap # start
C- DubboBootstrap  # exportServices
C- ServiceConfig # export + doExport + doExportUrlsFor1Protocol : 从配置中获取URL
C- RegistryProtocol # export
C- NacosNamingService # getAllInstances
C- NamingGrpcClientProxy # subscribe
C- NamingGrpcClientProxy # requestToServer

Zookeeper 请求逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码// 调用逻辑
C- CuratorZookeeperClient.<init>
C- CuratorZookeeperTransporter.createZookeeperClient
C- AbstractZookeeperTransporter.connect
C- ZookeeperTransporter$Adaptive.connect
C- ZookeeperDynamicConfiguration.<init>
C- ZookeeperDynamicConfigurationFactory.createDynamicConfiguration
C- AbstractDynamicConfigurationFactory.getDynamicConfiguration
C- DynamicConfiguration.getDynamicConfiguration
C- DubboBootstrap.prepareEnvironment
C- DubboBootstrap.startConfigCenter
C- DubboBootstrap.initialize
C- DubboBootstrap.start
C- DubboBootstrapApplicationListener.onContextRefreshedEvent
C- DubboBootstrapApplicationListener.onApplicationContextEvent

总结 :

ServiceDiscoveryRegistry 这一块没深入太多 , 后面有时间再看看 , 下面来看一下完整的流程图

Dubbo-Registry-Service.png

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%