mirror of
https://gitee.com/mirrors/Spring-Cloud-Alibaba.git
synced 2021-06-26 13:25:11 +08:00
余黄彬
GitHub
commit
480fa336bf
15
README-zh.md
15
README-zh.md
@ -48,15 +48,16 @@ Spring Cloud Alibaba 致力于提供微服务开发的一站式解决方案。
|
||||
更多组件请参考 [Roadmap](https://github.com/alibaba/spring-cloud-alibaba/blob/master/Roadmap-zh.md)。
|
||||
|
||||
## 如何构建
|
||||
|
||||
* master 分支对应的是 Spring Cloud Greenwich，最低支持 JDK 1.8。
|
||||
* 2020.0 分支对应的是 Spring Cloud 2020，最低支持 JDK 1.8。
|
||||
* master 分支对应的是 Spring Cloud Hoxton，最低支持 JDK 1.8。
|
||||
* greenwich 分支对应的是 Spring Cloud Greenwich，最低支持 JDK 1.8。
|
||||
* finchley 分支对应的是 Spring Cloud Finchley，最低支持 JDK 1.8。
|
||||
* 1.x 分支对应的是 Spring Cloud Edgware，最低支持 JDK 1.7。
|
||||
|
||||
Spring Cloud 使用 Maven 来构建，最快的使用方式是将本项目 clone 到本地，然后执行以下命令：
|
||||
|
||||
```bash
|
||||
./mvnw install
|
||||
|
||||
```
|
||||
执行完毕后，项目将被安装到本地 Maven 仓库。
|
||||
|
||||
## 如何使用
|
||||
@ -64,7 +65,7 @@ Spring Cloud 使用 Maven 来构建，最快的使用方式是将本项目 clone
|
||||
### 如何引入依赖
|
||||
|
||||
如果需要使用已发布的版本，在 `dependencyManagement` 中添加如下配置。
|
||||
|
||||
```xml
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
@ -76,7 +77,7 @@ Spring Cloud 使用 Maven 来构建，最快的使用方式是将本项目 clone
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
```
|
||||
然后在 `dependencies` 中添加自己所需使用的依赖即可使用。
|
||||
|
||||
## 演示 Demo
|
||||
@ -111,7 +112,7 @@ Example 列表：
|
||||
* 2.0.x 版本适用于 Spring Boot 2.0.x
|
||||
* 2.1.x 版本适用于 Spring Boot 2.1.x
|
||||
* 2.2.x 版本适用于 Spring Boot 2.2.x
|
||||
|
||||
* 2021.x 版本适用于 Spring Boot 2.4.x
|
||||
|
||||
## 社区交流
|
||||
|
||||
|
||||
16
README.md
16
README.md
@ -51,22 +51,23 @@ For more features, please refer to [Roadmap](https://github.com/alibaba/spring-c
|
||||
For more features please refer to [Roadmap](https://github.com/alibaba/spring-cloud-alibaba/blob/master/Roadmap.md).
|
||||
|
||||
## How to build
|
||||
|
||||
* **master branch**: Corresponds to Spring Cloud Greenwich & Spring Boot 2.x. JDK 1.8 or later versions are supported.
|
||||
* **finchley branch**: Corresponds to Spring Cloud Finchley & Spring Boot 2.x. JDK 1.8 or later versions are supported.
|
||||
* **2020.0 branch**: Corresponds to Spring Cloud 2020 & Spring Boot 2.4.x. JDK 1.8 or later versions are supported.
|
||||
* **master branch**: Corresponds to Spring Cloud Hoxton & Spring Boot 2.2.x. JDK 1.8 or later versions are supported.
|
||||
* **greenwich branch**: Corresponds to Spring Cloud Greenwich & Spring Boot 2.1.x. JDK 1.8 or later versions are supported.
|
||||
* **finchley branch**: Corresponds to Spring Cloud Finchley & Spring Boot 2.0.x. JDK 1.8 or later versions are supported.
|
||||
* **1.x branch**: Corresponds to Spring Cloud Edgware & Spring Boot 1.x, JDK 1.7 or later versions are supported.
|
||||
|
||||
Spring Cloud uses Maven for most build-related activities, and you should be able to get off the ground quite quickly by cloning the project you are interested in and typing:
|
||||
|
||||
```bash
|
||||
./mvnw install
|
||||
|
||||
```
|
||||
|
||||
## How to Use
|
||||
|
||||
### Add maven dependency
|
||||
|
||||
These artifacts are available from Maven Central and Spring Release repository via BOM:
|
||||
|
||||
```xml
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
@ -78,7 +79,7 @@ These artifacts are available from Maven Central and Spring Release repository v
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
```
|
||||
add the module in `dependencies`.
|
||||
|
||||
|
||||
@ -118,6 +119,7 @@ As the interfaces and annotations of Spring Boot 1 and Spring Boot 2 have been c
|
||||
* 2.0.x for Spring Boot 2.0.x
|
||||
* 2.1.x for Spring Boot 2.1.x
|
||||
* 2.2.x for Spring Boot 2.2.x
|
||||
* 2020.x for Spring Boot 2.4.x
|
||||
|
||||
## Code of Conduct
|
||||
This project is a sub-project of Spring Cloud, it adheres to the Contributor Covenant [code of conduct](https://github.com/spring-cloud/spring-cloud-build/blob/master/docs/src/main/asciidoc/code-of-conduct.adoc). By participating, you are expected to uphold this code. Please report unacceptable behavior to spring-code-of-conduct@pivotal.io.
|
||||
|
||||
@ -21,7 +21,7 @@
|
||||
<revision>2.2.6-SNAPSHOT</revision>
|
||||
<sentinel.version>1.8.1</sentinel.version>
|
||||
<seata.version>1.3.0</seata.version>
|
||||
<nacos.client.version>1.4.1</nacos.client.version>
|
||||
<nacos.client.version>1.4.2</nacos.client.version>
|
||||
<nacos.config.version>0.8.0</nacos.config.version>
|
||||
<spring.context.support.version>1.0.10</spring.context.support.version>
|
||||
|
||||
|
||||
@ -260,7 +260,7 @@ Nacos 内部有 https://nacos.io/zh-cn/docs/concepts.html[Namespace 的概念]:
|
||||
[quote]
|
||||
用于进行租户粒度的配置隔离。不同的命名空间下，可以存在相同的 Group 或 Data ID 的配置。Namespace 的常用场景之一是不同环境的配置的区分隔离，例如开发测试环境和生产环境的资源（如配置、服务）隔离等。
|
||||
|
||||
在没有明确指定 `${spring.cloud.nacos.config.namespace}` 配置的情况下， 默认使用的是 Nacos 上 Public 这个namespae。如果需要使用自定义的命名空间，可以通过以下配置来实现：
|
||||
在没有明确指定 `${spring.cloud.nacos.config.namespace}` 配置的情况下， 默认使用的是 Nacos 上 Public 这个namespace。如果需要使用自定义的命名空间，可以通过以下配置来实现：
|
||||
[source,properties]
|
||||
----
|
||||
spring.cloud.nacos.config.namespace=b3404bc0-d7dc-4855-b519-570ed34b62d7
|
||||
@ -348,7 +348,13 @@ Nacos Config 目前提供了三种配置能力从 Nacos 拉取相关的配置
|
||||
* B: 通过 `spring.cloud.nacos.config.ext-config[n].data-id` 的方式支持多个扩展 Data Id 的配置
|
||||
* C: 通过内部相关规则(应用名、应用名+ Profile )自动生成相关的 Data Id 配置
|
||||
|
||||
当三种方式共同使用时，他们的一个优先级关系是:A < B < C
|
||||
当三种方式共同使用时，他们的一个优先级关系是:
|
||||
|
||||
->A为优先级最高的
|
||||
|
||||
->B的优先级低于A，
|
||||
|
||||
->C的优先级是最低的
|
||||
|
||||
=== Nacos Config 对外暴露的 Endpoint
|
||||
|
||||
|
||||
@ -57,7 +57,7 @@ Send messages:
|
||||
sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
|
||||
```
|
||||
|
||||
Output when the message is successfuly sent: `SendResult [sendStatus=SEND_OK, msgId= ...`
|
||||
Output when the message is successfully sent: `SendResult [sendStatus=SEND_OK, msgId= ...`
|
||||
|
||||
Receive messages:
|
||||
|
||||
|
||||
@ -82,7 +82,7 @@ Job Description： Empty
|
||||
Custom Parameters： Empty
|
||||
----
|
||||
|
||||
The job above is a “Simple Single-Server Job”, and speficied a Cron expression of "0 * * * * ?" . This means that the job will be executed once and once only in every minute.
|
||||
The job above is a “Simple Single-Server Job”, and specified a Cron expression of "0 * * * * ?" . This means that the job will be executed once and once only in every minute.
|
||||
|
||||
For more job types, refer to https://help.aliyun.com/document_detail/43136.html[SchedulerX Documentation].
|
||||
|
||||
|
||||
@ -73,7 +73,7 @@ Provider端在application.properties文件中定义dubbo相关的配置，比如
|
||||
|
||||
定义具体的服务：
|
||||
|
||||
@Service(
|
||||
@DubboService(
|
||||
version = "${foo.service.version}",
|
||||
application = "${dubbo.application.id}",
|
||||
protocol = "${dubbo.protocol.id}",
|
||||
@ -111,7 +111,7 @@ Consumer端在服务调用之前，先定义限流规则。
|
||||
|
||||
根据Provider端中发布的定义，使用Dubbo的@Reference注解注入服务对应的Bean：
|
||||
|
||||
@Reference(version = "${foo.service.version}", application = "${dubbo.application.id}",
|
||||
@DubboReference(version = "${foo.service.version}", application = "${dubbo.application.id}",
|
||||
path = "dubbo://localhost:12345", timeout = 30000)
|
||||
private FooService fooService;
|
||||
|
||||
|
||||
@ -16,14 +16,14 @@
|
||||
|
||||
package com.alibaba.cloud.examples;
|
||||
|
||||
import org.apache.dubbo.config.annotation.Reference;
|
||||
import org.apache.dubbo.config.annotation.DubboReference;
|
||||
|
||||
/**
|
||||
* @author fangjian
|
||||
*/
|
||||
public class FooServiceConsumer {
|
||||
|
||||
@Reference(version = "${foo.service.version}",
|
||||
@DubboReference(version = "${foo.service.version}",
|
||||
application = "${dubbo.application.id}",
|
||||
url = "dubbo://localhost:12345?version=1.0.0", timeout = 30000)
|
||||
private FooService fooService;
|
||||
|
||||
@ -16,12 +16,12 @@
|
||||
|
||||
package com.alibaba.cloud.examples;
|
||||
|
||||
import org.apache.dubbo.config.annotation.Service;
|
||||
import org.apache.dubbo.config.annotation.DubboService;
|
||||
|
||||
/**
|
||||
* @author fangjian
|
||||
*/
|
||||
@Service(version = "${foo.service.version}", application = "${dubbo.application.id}",
|
||||
@DubboService(version = "${foo.service.version}", application = "${dubbo.application.id}",
|
||||
protocol = "${dubbo.protocol.id}", registry = "${dubbo.registry.id}")
|
||||
public class FooServiceImpl implements FooService {
|
||||
|
||||
|
||||
@ -132,7 +132,6 @@ public class DataSourcePropertiesConfiguration {
|
||||
if (!ObjectUtils.isEmpty(field.get(this))) {
|
||||
return field.getName();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
catch (IllegalAccessException e) {
|
||||
// won't happen
|
||||
|
||||
@ -86,9 +86,8 @@ public abstract class SentinelConverter<T extends Object>
|
||||
});
|
||||
|
||||
for (Object obj : sourceArray) {
|
||||
String item = null;
|
||||
try {
|
||||
item = objectMapper.writeValueAsString(obj);
|
||||
String item = objectMapper.writeValueAsString(obj);
|
||||
Optional.ofNullable(convertRule(item))
|
||||
.ifPresent(convertRule -> ruleCollection.add(convertRule));
|
||||
}
|
||||
|
||||
@ -32,6 +32,16 @@ public class NacosConfigHealthIndicator extends AbstractHealthIndicator {
|
||||
|
||||
private final ConfigService configService;
|
||||
|
||||
/**
|
||||
* status up .
|
||||
*/
|
||||
private final String STATUS_UP = "UP";
|
||||
|
||||
/**
|
||||
* status down .
|
||||
*/
|
||||
private final String STATUS_DOWN = "DOWN";
|
||||
|
||||
public NacosConfigHealthIndicator(ConfigService configService) {
|
||||
this.configService = configService;
|
||||
}
|
||||
@ -43,10 +53,10 @@ public class NacosConfigHealthIndicator extends AbstractHealthIndicator {
|
||||
// Set the status to Builder
|
||||
builder.status(status);
|
||||
switch (status) {
|
||||
case "UP":
|
||||
case STATUS_UP:
|
||||
builder.up();
|
||||
break;
|
||||
case "DOWN":
|
||||
case STATUS_DOWN:
|
||||
builder.down();
|
||||
break;
|
||||
default:
|
||||
|
||||
@ -31,6 +31,16 @@ import org.springframework.boot.actuate.health.HealthIndicator;
|
||||
*/
|
||||
public class NacosDiscoveryHealthIndicator extends AbstractHealthIndicator {
|
||||
|
||||
/**
|
||||
* status up.
|
||||
*/
|
||||
private static final String STATUS_UP = "UP";
|
||||
|
||||
/**
|
||||
* status down.
|
||||
*/
|
||||
private static final String STATUS_DOWN = "DOWN";
|
||||
|
||||
private final NamingService namingService;
|
||||
|
||||
public NacosDiscoveryHealthIndicator(NamingService namingService) {
|
||||
@ -44,10 +54,10 @@ public class NacosDiscoveryHealthIndicator extends AbstractHealthIndicator {
|
||||
// Set the status to Builder
|
||||
builder.status(status);
|
||||
switch (status) {
|
||||
case "UP":
|
||||
case STATUS_UP:
|
||||
builder.up();
|
||||
break;
|
||||
case "DOWN":
|
||||
case STATUS_DOWN:
|
||||
builder.down();
|
||||
break;
|
||||
default:
|
||||
|
||||
@ -51,6 +51,10 @@ public class DubboCloudProperties {
|
||||
|
||||
private String registryType = DUBBO_CLOUD_REGISTRY_PROPERTY_VALUE;
|
||||
|
||||
private int maxReSubscribeMetadataTimes = 1000;
|
||||
|
||||
private int reSubscribeMetadataIntervial = 5;
|
||||
|
||||
public String getSubscribedServices() {
|
||||
return subscribedServices;
|
||||
}
|
||||
@ -91,4 +95,20 @@ public class DubboCloudProperties {
|
||||
this.registryType = registryType;
|
||||
}
|
||||
|
||||
public int getMaxReSubscribeMetadataTimes() {
|
||||
return maxReSubscribeMetadataTimes;
|
||||
}
|
||||
|
||||
public void setMaxReSubscribeMetadataTimes(int maxReSubscribeMetadataTimes) {
|
||||
this.maxReSubscribeMetadataTimes = maxReSubscribeMetadataTimes;
|
||||
}
|
||||
|
||||
public int getReSubscribeMetadataIntervial() {
|
||||
return reSubscribeMetadataIntervial;
|
||||
}
|
||||
|
||||
public void setReSubscribeMetadataIntervial(int reSubscribeMetadataIntervial) {
|
||||
this.reSubscribeMetadataIntervial = reSubscribeMetadataIntervial;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -19,6 +19,7 @@ package com.alibaba.cloud.dubbo.metadata.repository;
|
||||
import java.util.Collections;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.LinkedHashSet;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
@ -41,6 +42,7 @@ import com.alibaba.cloud.dubbo.util.JSONUtils;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import com.fasterxml.jackson.databind.type.TypeFactory;
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.constants.CommonConstants;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@ -237,9 +239,6 @@ public class DubboServiceMetadataRepository
|
||||
dispatchEvent(new SubscribedServicesChangedEvent(this, oldSubscribedServices,
|
||||
newSubscribedServices));
|
||||
|
||||
// clear old one, help GC
|
||||
oldSubscribedServices.clear();
|
||||
|
||||
return newSubscribedServices.stream();
|
||||
}
|
||||
|
||||
@ -407,9 +406,35 @@ public class DubboServiceMetadataRepository
|
||||
|
||||
public List<URL> getExportedURLs(String serviceInterface, String group,
|
||||
String version) {
|
||||
if (group != null) {
|
||||
List<URL> urls = new LinkedList<>();
|
||||
if (CommonConstants.ANY_VALUE.equals(group)) {
|
||||
String serviceKey = URL.buildKey(serviceInterface, group, version);
|
||||
String expectKey = serviceKey.substring(2);
|
||||
for (String key : allExportedURLs.keySet()) {
|
||||
if (key.endsWith(expectKey)) {
|
||||
urls.addAll(allExportedURLs.get(key));
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
String[] groups = group.split(CommonConstants.COMMA_SEPARATOR);
|
||||
for (String expectKey : groups) {
|
||||
String serviceKey = URL.buildKey(serviceInterface, expectKey,
|
||||
version);
|
||||
List<URL> urlList = allExportedURLs.get(serviceKey);
|
||||
if (urlList != null) {
|
||||
urls.addAll(urlList);
|
||||
}
|
||||
}
|
||||
}
|
||||
return urls;
|
||||
}
|
||||
else {
|
||||
String serviceKey = URL.buildKey(serviceInterface, null, version);
|
||||
return allExportedURLs.getOrDefault(serviceKey, Collections.emptyList());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initialize the specified service's {@link ServiceRestMetadata}.
|
||||
|
||||
@ -23,6 +23,9 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
@ -45,12 +48,10 @@ import org.springframework.cloud.client.discovery.DiscoveryClient;
|
||||
import org.springframework.context.ApplicationListener;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
import static java.lang.String.format;
|
||||
import static java.util.Collections.emptyList;
|
||||
import static java.util.stream.StreamSupport.stream;
|
||||
import static org.apache.dubbo.common.URLBuilder.from;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.PID_KEY;
|
||||
@ -110,11 +111,23 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
|
||||
private final String currentApplicationName;
|
||||
|
||||
private final Map<URL, NotifyListener> urlNotifyListenerMap = new ConcurrentHashMap<>();
|
||||
|
||||
private final Map<String, ReSubscribeMetadataJob> reConnectJobMap = new ConcurrentHashMap<>();
|
||||
|
||||
private final ScheduledThreadPoolExecutor reConnectPool = new ScheduledThreadPoolExecutor(
|
||||
2);
|
||||
|
||||
private final int maxReSubscribeMetadataTimes;
|
||||
|
||||
private final int reSubscribeMetadataIntervial;
|
||||
|
||||
public DubboCloudRegistry(URL url, DiscoveryClient discoveryClient,
|
||||
DubboServiceMetadataRepository repository,
|
||||
DubboMetadataServiceProxy dubboMetadataConfigServiceProxy,
|
||||
JSONUtils jsonUtils, DubboGenericServiceFactory dubboGenericServiceFactory,
|
||||
ConfigurableApplicationContext applicationContext) {
|
||||
ConfigurableApplicationContext applicationContext,
|
||||
int maxReSubscribeMetadataTimes, int reSubscribeMetadataIntervial) {
|
||||
|
||||
super(url);
|
||||
this.servicesLookupInterval = url
|
||||
@ -127,6 +140,11 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
this.applicationContext = applicationContext;
|
||||
this.dubboMetadataUtils = getBean(DubboMetadataUtils.class);
|
||||
this.currentApplicationName = dubboMetadataUtils.getCurrentApplicationName();
|
||||
this.maxReSubscribeMetadataTimes = maxReSubscribeMetadataTimes;
|
||||
this.reSubscribeMetadataIntervial = reSubscribeMetadataIntervial;
|
||||
|
||||
reConnectPool.setKeepAliveTime(10, TimeUnit.MINUTES);
|
||||
reConnectPool.allowCoreThreadTimeOut(true);
|
||||
}
|
||||
|
||||
private <T> T getBean(Class<T> beanClass) {
|
||||
@ -177,6 +195,7 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
}
|
||||
else { // for general Dubbo Services
|
||||
subscribeURLs(url, listener);
|
||||
urlNotifyListenerMap.put(url, listener);
|
||||
}
|
||||
}
|
||||
|
||||
@ -188,19 +207,34 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
// Async subscription
|
||||
registerServiceInstancesChangedListener(url,
|
||||
|
||||
new ApplicationListener<ServiceInstancesChangedEvent>() {
|
||||
|
||||
private final URL url2subscribe = url;
|
||||
new ServiceInstanceChangeListener() {
|
||||
|
||||
@Override
|
||||
public int getOrder() {
|
||||
return Ordered.LOWEST_PRECEDENCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
@Order
|
||||
public void onApplicationEvent(ServiceInstancesChangedEvent event) {
|
||||
|
||||
Set<String> serviceNames = getServices(url);
|
||||
|
||||
String serviceName = event.getServiceName();
|
||||
|
||||
if (serviceNames.contains(serviceName)) {
|
||||
logger.debug(
|
||||
"handle serviceInstanceChange of general service, serviceName = {}, subscribeUrl={}",
|
||||
event.getServiceName(), url.getServiceKey());
|
||||
try {
|
||||
subscribeURLs(url, serviceNames, listener);
|
||||
reConnectJobMap.remove(serviceName);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.warn(String.format(
|
||||
"subscribeURLs failed, serviceName = %s, try reSubscribe again",
|
||||
serviceName), e);
|
||||
addReSubscribeMetadataJob(serviceName, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -212,8 +246,19 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
});
|
||||
}
|
||||
|
||||
private void subscribeURLs(URL url, Set<String> serviceNames,
|
||||
NotifyListener listener) {
|
||||
void addReSubscribeMetadataJob(String serviceName, int count) {
|
||||
if (count > maxReSubscribeMetadataTimes) {
|
||||
logger.error(
|
||||
"reSubscribe failed too many times, serviceName = {}, count = {}",
|
||||
serviceName, count);
|
||||
return;
|
||||
}
|
||||
ReSubscribeMetadataJob job = new ReSubscribeMetadataJob(serviceName, this, count);
|
||||
reConnectJobMap.put(serviceName, job);
|
||||
reConnectPool.schedule(job, reSubscribeMetadataIntervial, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
void subscribeURLs(URL url, Set<String> serviceNames, NotifyListener listener) {
|
||||
|
||||
List<URL> subscribedURLs = new LinkedList<>();
|
||||
|
||||
@ -252,6 +297,10 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
serviceName));
|
||||
}
|
||||
}
|
||||
else {
|
||||
logger.debug("subscribe from serviceName = {}, size = {}", serviceName,
|
||||
serviceInstances.size());
|
||||
}
|
||||
|
||||
List<URL> exportedURLs = getExportedURLs(subscribedURL, serviceName,
|
||||
serviceInstances);
|
||||
@ -389,7 +438,7 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
return metadata.containsKey(METADATA_SERVICE_URLS_PROPERTY_NAME);
|
||||
}
|
||||
|
||||
private Set<String> getServices(URL url) {
|
||||
Set<String> getServices(URL url) {
|
||||
Set<String> subscribedServices = repository.getSubscribedServices();
|
||||
// TODO Add the filter feature
|
||||
return subscribedServices;
|
||||
@ -419,11 +468,6 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
listener.notify(subscribedURLs);
|
||||
}
|
||||
|
||||
private List<ServiceInstance> getServiceInstances(Iterable<String> serviceNames) {
|
||||
return stream(serviceNames.spliterator(), false).map(this::getServiceInstances)
|
||||
.flatMap(Collection::stream).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
private List<ServiceInstance> getServiceInstances(String serviceName) {
|
||||
return hasText(serviceName) ? doGetServiceInstances(serviceName) : emptyList();
|
||||
}
|
||||
@ -471,27 +515,38 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
private void subscribeDubboMetadataServiceURLs(URL subscribedURL,
|
||||
NotifyListener listener) {
|
||||
|
||||
// Sync subscription
|
||||
subscribeDubboMetadataServiceURLs(subscribedURL, listener,
|
||||
getServiceName(subscribedURL));
|
||||
|
||||
// Sync subscription
|
||||
if (containsProviderCategory(subscribedURL)) {
|
||||
registerServiceInstancesChangedListener(subscribedURL,
|
||||
new ApplicationListener<ServiceInstancesChangedEvent>() {
|
||||
|
||||
private final URL url2subscribe = subscribedURL;
|
||||
registerServiceInstancesChangedListener(subscribedURL,
|
||||
new ServiceInstanceChangeListener() {
|
||||
|
||||
@Override
|
||||
public int getOrder() {
|
||||
return Ordered.LOWEST_PRECEDENCE - 1;
|
||||
}
|
||||
|
||||
@Override
|
||||
@Order(Ordered.LOWEST_PRECEDENCE - 1)
|
||||
public void onApplicationEvent(
|
||||
ServiceInstancesChangedEvent event) {
|
||||
String sourceServiceName = event.getServiceName();
|
||||
List<ServiceInstance> serviceInstances = event
|
||||
.getServiceInstances();
|
||||
String serviceName = getServiceName(subscribedURL);
|
||||
|
||||
if (Objects.equals(sourceServiceName, serviceName)) {
|
||||
logger.debug(
|
||||
"handle serviceInstanceChange of metadata service, serviceName = {}, subscribeUrl={}",
|
||||
event.getServiceName(),
|
||||
subscribedURL.getServiceKey());
|
||||
|
||||
// only update serviceInstances of the specified
|
||||
// serviceName
|
||||
subscribeDubboMetadataServiceURLs(subscribedURL, listener,
|
||||
sourceServiceName);
|
||||
sourceServiceName, serviceInstances);
|
||||
}
|
||||
}
|
||||
|
||||
@ -509,34 +564,25 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
}
|
||||
|
||||
private void subscribeDubboMetadataServiceURLs(URL subscribedURL,
|
||||
NotifyListener listener, String serviceName) {
|
||||
NotifyListener listener, String serviceName,
|
||||
List<ServiceInstance> serviceInstances) {
|
||||
|
||||
String serviceInterface = subscribedURL.getServiceInterface();
|
||||
String version = subscribedURL.getParameter(VERSION_KEY);
|
||||
String protocol = subscribedURL.getParameter(PROTOCOL_KEY);
|
||||
|
||||
List<ServiceInstance> serviceInstances = getServiceInstances(serviceName);
|
||||
|
||||
List<URL> urls = dubboMetadataUtils.getDubboMetadataServiceURLs(serviceInstances,
|
||||
serviceInterface, version, protocol);
|
||||
|
||||
notifyAllSubscribedURLs(subscribedURL, urls, listener);
|
||||
}
|
||||
|
||||
// private void subscribeDubboMetadataServiceURLs(URL subscribedURL,
|
||||
// NotifyListener listener, Set<String> serviceNames) {
|
||||
//
|
||||
// String serviceInterface = subscribedURL.getServiceInterface();
|
||||
// String version = subscribedURL.getParameter(VERSION_KEY);
|
||||
// String protocol = subscribedURL.getParameter(PROTOCOL_KEY);
|
||||
//
|
||||
// List<ServiceInstance> serviceInstances = getServiceInstances(serviceNames);
|
||||
//
|
||||
// List<URL> urls = dubboMetadataUtils.getDubboMetadataServiceURLs(serviceInstances,
|
||||
// serviceInterface, version, protocol);
|
||||
//
|
||||
// notifyAllSubscribedURLs(subscribedURL, urls, listener);
|
||||
// }
|
||||
private void subscribeDubboMetadataServiceURLs(URL subscribedURL,
|
||||
NotifyListener listener, String serviceName) {
|
||||
List<ServiceInstance> serviceInstances = getServiceInstances(serviceName);
|
||||
subscribeDubboMetadataServiceURLs(subscribedURL, listener, serviceName,
|
||||
serviceInstances);
|
||||
}
|
||||
|
||||
private boolean containsProviderCategory(URL subscribedURL) {
|
||||
String category = subscribedURL.getParameter(CATEGORY_KEY);
|
||||
@ -557,6 +603,14 @@ public class DubboCloudRegistry extends FailbackRegistry {
|
||||
return ADMIN_PROTOCOL.equals(url.getProtocol());
|
||||
}
|
||||
|
||||
public Map<URL, NotifyListener> getUrlNotifyListenerMap() {
|
||||
return urlNotifyListenerMap;
|
||||
}
|
||||
|
||||
public Map<String, ReSubscribeMetadataJob> getReConnectJobMap() {
|
||||
return reConnectJobMap;
|
||||
}
|
||||
|
||||
protected boolean isDubboMetadataServiceURL(URL url) {
|
||||
return DUBBO_METADATA_SERVICE_CLASS_NAME.equals(url.getServiceInterface());
|
||||
}
|
||||
|
||||
@ -0,0 +1,84 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.dubbo.registry;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.registry.NotifyListener;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* For re subscribe URL from provider.
|
||||
*
|
||||
* @author <a href="mailto:chenxilzx1@gmail.com">theonefx</a>
|
||||
*/
|
||||
public class ReSubscribeMetadataJob implements Runnable {
|
||||
|
||||
protected final Logger logger = LoggerFactory.getLogger(ReSubscribeMetadataJob.class);
|
||||
|
||||
private final String serviceName;
|
||||
|
||||
private final DubboCloudRegistry dubboCloudRegistry;
|
||||
|
||||
private final int errorCounts;
|
||||
|
||||
public ReSubscribeMetadataJob(String serviceName,
|
||||
DubboCloudRegistry dubboCloudRegistry, int errorCounts) {
|
||||
this.errorCounts = errorCounts;
|
||||
this.serviceName = serviceName;
|
||||
this.dubboCloudRegistry = dubboCloudRegistry;
|
||||
}
|
||||
|
||||
public ReSubscribeMetadataJob(String serviceName,
|
||||
DubboCloudRegistry dubboCloudRegistry) {
|
||||
this(serviceName, dubboCloudRegistry, 0);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
if (dubboCloudRegistry.getReConnectJobMap().get(serviceName) != this) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
logger.info("reSubscribe, serviceName = {}, count = {}", serviceName,
|
||||
errorCounts);
|
||||
for (Map.Entry<URL, NotifyListener> entry : dubboCloudRegistry
|
||||
.getUrlNotifyListenerMap().entrySet()) {
|
||||
doRun(entry.getKey(), entry.getValue());
|
||||
}
|
||||
dubboCloudRegistry.getReConnectJobMap().remove(serviceName);
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.warn(String.format(
|
||||
"reSubscribe failed, serviceName = %s, try refresh again",
|
||||
serviceName), e);
|
||||
dubboCloudRegistry.addReSubscribeMetadataJob(serviceName, errorCounts + 1);
|
||||
}
|
||||
}
|
||||
|
||||
private void doRun(URL url, NotifyListener listener) {
|
||||
Set<String> serviceNames = dubboCloudRegistry.getServices(url);
|
||||
|
||||
if (serviceNames.contains(serviceName)) {
|
||||
dubboCloudRegistry.subscribeURLs(url, serviceNames, listener);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.dubbo.registry;
|
||||
|
||||
import com.alibaba.cloud.dubbo.registry.event.ServiceInstancesChangedEvent;
|
||||
|
||||
import org.springframework.context.ApplicationListener;
|
||||
import org.springframework.core.Ordered;
|
||||
|
||||
/**
|
||||
* The interface of ServiceInstanceChange event Listener.
|
||||
*
|
||||
* @author <a href="mailto:chenxilzx1@gmail.com">theonefx</a>
|
||||
* @see ServiceInstancesChangedEvent
|
||||
* @see Ordered
|
||||
* @see ApplicationListener
|
||||
*/
|
||||
public interface ServiceInstanceChangeListener
|
||||
extends ApplicationListener<ServiceInstancesChangedEvent>, Ordered {
|
||||
|
||||
}
|
||||
@ -100,7 +100,9 @@ public class SpringCloudRegistryFactory extends AbstractRegistryFactory {
|
||||
default:
|
||||
registry = new DubboCloudRegistry(url, discoveryClient,
|
||||
dubboServiceMetadataRepository, dubboMetadataConfigServiceProxy,
|
||||
jsonUtils, dubboGenericServiceFactory, applicationContext);
|
||||
jsonUtils, dubboGenericServiceFactory, applicationContext,
|
||||
dubboCloudProperties.getMaxReSubscribeMetadataTimes(),
|
||||
dubboCloudProperties.getReSubscribeMetadataIntervial());
|
||||
break;
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user