natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[3.0] Service Name Mapping Refactor And Nacos model some support. (#7824)

This commit is contained in:
赵延 2021-06-10 17:18:49 +08:00 committed by GitHub
parent db41d2a5f4
commit 724ec8b554
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
61 changed files with 1172 additions and 702 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/DynamicConfiguration.java
View File

@ -172,11 +172,11 @@ public interface DynamicConfiguration extends Configuration, AutoCloseable {
* @param key
* @param group
* @param content
* @param stat
* @param ticket
* @return
* @throws UnsupportedOperationException
*/
default boolean publishConfigCas(String key, String group, String content, Object stat) throws UnsupportedOperationException {
default boolean publishConfigCas(String key, String group, String content, Object ticket) throws UnsupportedOperationException {
return false;
}
@ -266,13 +266,4 @@ public interface DynamicConfiguration extends Configuration, AutoCloseable {
default boolean removeConfig(String key, String group) {
return true;
}
/**
* support cas or not.
*
* @return
*/
default boolean hasSupportCas() {
return false;
}
}

5
dubbo-common/src/main/java/org/apache/dubbo/common/config/configcenter/wrapper/CompositeDynamicConfiguration.java
View File

@ -40,6 +40,11 @@ public class CompositeDynamicConfiguration implements DynamicConfiguration {
}
}
public Set<DynamicConfiguration> getInnerConfigurations() {
return configurations;
}
@Override
public void addListener(String key, String group, ConfigurationListener listener) {
iterateListenerOperation(configuration -> configuration.addListener(key, group, listener));

2
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -199,8 +199,6 @@ public interface CommonConstants {
String METADATA_KEY = "metadata-type";
String MAPPING_KEY = "mapping-type";
String CONFIG_MAPPING_TYPE = "config";
String METADATA_MAPPING_TYPE = "metadata";

5
dubbo-compatible/pom.xml
View File

@ -28,11 +28,6 @@
<name>${project.artifactId}</name>
<description>The compatible module of dubbo project</description>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-filter-cache</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-spring</artifactId>

9
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ServiceConfig.java
View File

@ -34,9 +34,8 @@ import org.apache.dubbo.config.bootstrap.DubboBootstrap;
import org.apache.dubbo.config.invoker.DelegateProviderMetaDataInvoker;
import org.apache.dubbo.config.support.Parameter;
import org.apache.dubbo.config.utils.ConfigValidationUtils;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.ServiceNameMapping;
import org.apache.dubbo.metadata.ServiceNameMappingHandler;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.metadata.MetadataUtils;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
@ -71,7 +70,6 @@ import static org.apache.dubbo.common.constants.CommonConstants.ANY_VALUE;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_IP_TO_BIND;
import static org.apache.dubbo.common.constants.CommonConstants.LOCALHOST_VALUE;
import static org.apache.dubbo.common.constants.CommonConstants.MAPPING_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METHODS_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY;
@ -219,9 +217,8 @@ public class ServiceConfig<T> extends ServiceConfigBase<T> {
public void exported() {
List<URL> exportedURLs = this.getExportedUrls();
exportedURLs.forEach(url -> {
Map<String, String> parameters = getApplication().getParameters();
ServiceNameMapping serviceNameMapping = ServiceNameMapping.getExtension(parameters != null ? parameters.get(MAPPING_KEY) : null);
ServiceNameMappingHandler.map(serviceNameMapping, url);
ServiceNameMapping serviceNameMapping = ServiceNameMapping.getDefaultExtension();
serviceNameMapping.map(url);
});
}

20
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/mock/MockServiceDiscovery.java
View File

@ -17,38 +17,37 @@
package org.apache.dubbo.config.mock;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.AbstractServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import java.util.HashSet;
import java.util.Set;
public class MockServiceDiscovery implements ServiceDiscovery {
public class MockServiceDiscovery extends AbstractServiceDiscovery {
private URL registryURL;
private ServiceInstance serviceInstance;
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.registryURL = registryURL;
}
@Override
public void destroy() throws Exception {
public void doDestroy() throws Exception {
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = null;
}
@ -61,9 +60,4 @@ public class MockServiceDiscovery implements ServiceDiscovery {
public URL getUrl() {
return registryURL;
}
@Override
public ServiceInstance getLocalInstance() {
return serviceInstance;
}
}

1
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/propertyconfigurer/consumer/dubbo-consumer.xml
View File

@ -28,7 +28,6 @@
</bean>
<dubbo:application name="demo-consumer">
<dubbo:parameter key="mapping-type" value="metadata"/>
</dubbo:application>
<!-- <dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/>-->

1
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/propertyconfigurer/consumer2/dubbo-consumer.xml
View File

@ -28,7 +28,6 @@
</bean>
<dubbo:application name="demo-consumer">
<dubbo:parameter key="mapping-type" value="metadata"/>
</dubbo:application>
<!-- <dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/>-->

1
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/propertyconfigurer/provider/dubbo-provider.xml
View File

@ -31,7 +31,6 @@
</bean>
<dubbo:application name="demo-provider">
<dubbo:parameter key="mapping-type" value="metadata"/>
</dubbo:application>
<dubbo:config-center address="${dubbo.config-center.address}"/>

23
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/registry/MockServiceDiscovery.java
View File

@ -17,39 +17,35 @@
package org.apache.dubbo.config.spring.registry;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.AbstractServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import java.util.HashSet;
import java.util.Set;
public class MockServiceDiscovery implements ServiceDiscovery {
public class MockServiceDiscovery extends AbstractServiceDiscovery {
private URL registryURL;
private ServiceInstance serviceInstance;
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.registryURL = registryURL;
}
@Override
public void destroy() throws Exception {
public void doDestroy() throws Exception {
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = null;
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
}
@Override
@ -61,9 +57,4 @@ public class MockServiceDiscovery implements ServiceDiscovery {
public URL getUrl() {
return registryURL;
}
@Override
public ServiceInstance getLocalInstance() {
return serviceInstance;
}
}

77
dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosConfigServiceWrapper.java Normal file
View File

@ -0,0 +1,77 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.configcenter.support.nacos;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException;
import static org.apache.dubbo.common.utils.StringUtils.HYPHEN_CHAR;
import static org.apache.dubbo.common.utils.StringUtils.SLASH_CHAR;
public class NacosConfigServiceWrapper {
private static final String INNERCLASS_SYMBOL = "$";
private static final String INNERCLASS_COMPATIBLE_SYMBOL = "___";
private static final long DEFAULT_TIMEOUT = 3000L;
private ConfigService configService;
public NacosConfigServiceWrapper(ConfigService configService) {
this.configService = configService;
}
public ConfigService getConfigService() {
return configService;
}
public void addListener(String dataId, String group, Listener listener) throws NacosException {
configService.addListener(handleInnerSymbol(dataId), handleInnerSymbol(group), listener);
}
public String getConfig(String dataId, String group) throws NacosException {
return configService.getConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), DEFAULT_TIMEOUT);
}
public String getConfig(String dataId, String group, long timeout) throws NacosException {
return configService.getConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), timeout);
}
public boolean publishConfig(String dataId, String group, String content) throws NacosException {
return configService.publishConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), content);
}
public boolean publishConfigCas(String dataId, String group, String content, String casMd5) throws NacosException {
return configService.publishConfigCas(handleInnerSymbol(dataId), handleInnerSymbol(group), content, casMd5);
}
public boolean removeConfig(String dataId, String group) throws NacosException {
return configService.removeConfig(handleInnerSymbol(dataId), handleInnerSymbol(group));
}
/**
* see {@link com.alibaba.nacos.client.config.utils.ParamUtils#isValid(java.lang.String)}
*/
private String handleInnerSymbol(String data) {
if (data == null) {
return null;
}
return data.replace(INNERCLASS_SYMBOL, INNERCLASS_COMPATIBLE_SYMBOL).replace(SLASH_CHAR, HYPHEN_CHAR);
}
}

82
dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java
View File

@ -37,27 +37,27 @@ import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.AbstractSharedListener;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.client.config.http.HttpAgent;
import com.alibaba.nacos.common.http.HttpRestResult;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executor;
import java.util.stream.Stream;
import static com.alibaba.nacos.api.PropertyKeyConst.NAMING_LOAD_CACHE_AT_START;
import static com.alibaba.nacos.api.PropertyKeyConst.ENCODE;
import static com.alibaba.nacos.api.PropertyKeyConst.SERVER_ADDR;
import static com.alibaba.nacos.client.naming.utils.UtilAndComs.NACOS_NAMING_LOG_NAME;
import static java.util.Collections.emptyMap;
import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
import static org.apache.dubbo.common.utils.StringConstantFieldValuePredicate.of;
import static org.apache.dubbo.common.utils.StringUtils.HYPHEN_CHAR;
import static org.apache.dubbo.common.utils.StringUtils.SLASH_CHAR;
import static org.apache.dubbo.common.utils.StringUtils.isBlank;
/**
* The nacos implementation of {@link DynamicConfiguration}
@ -77,23 +77,23 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
/**
* The nacos configService
*/
private final ConfigService configService;
private final NacosConfigServiceWrapper configService;
private HttpAgent httpAgent;
/**
* The map store the key to {@link NacosConfigListener} mapping
*/
private final ConcurrentMap<String, NacosConfigListener> watchListenerMap;
private final Map<String, NacosConfigListener> watchListenerMap;
NacosDynamicConfiguration(URL url) {
this.nacosProperties = buildNacosProperties(url);
this.configService = buildConfigService(url);
this.httpAgent = getHttpAgent(configService);
this.httpAgent = getHttpAgent(configService.getConfigService());
watchListenerMap = new ConcurrentHashMap<>();
}
private ConfigService buildConfigService(URL url) {
private NacosConfigServiceWrapper buildConfigService(URL url) {
ConfigService configService = null;
try {
configService = NacosFactory.createConfigService(nacosProperties);
@ -103,7 +103,7 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
}
throw new IllegalStateException(e);
}
return configService;
return new NacosConfigServiceWrapper(configService);
}
private HttpAgent getHttpAgent(ConfigService configService) {
@ -141,14 +141,10 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
}
private static void setProperties(URL url, Properties properties) {
putPropertyIfAbsent(url, properties, NACOS_NAMING_LOG_NAME);
// Get the parameters from constants
Map<String, String> parameters = url.getParameters(of(PropertyKeyConst.class));
// Put all parameters
properties.putAll(parameters);
putPropertyIfAbsent(url, properties, NAMING_LOAD_CACHE_AT_START, "true");
}
private static void putPropertyIfAbsent(URL url, Properties properties, String propertyName) {
@ -182,13 +178,12 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
@Override
public void addListener(String key, String group, ConfigurationListener listener) {
String resolvedGroup = resolveGroup(group);
String listenerKey = buildListenerKey(key, group);
NacosConfigListener nacosConfigListener =
watchListenerMap.computeIfAbsent(listenerKey, k -> createTargetListener(key, resolvedGroup));
watchListenerMap.computeIfAbsent(listenerKey, k -> createTargetListener(key, group));
nacosConfigListener.addListener(listener);
try {
configService.addListener(key, resolvedGroup, nacosConfigListener);
configService.addListener(key, group, nacosConfigListener);
} catch (NacosException e) {
logger.error(e.getMessage());
}
@ -205,13 +200,12 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
@Override
public String getConfig(String key, String group, long timeout) throws IllegalStateException {
String resolvedGroup = resolveGroup(group);
try {
long nacosTimeout = timeout < 0 ? getDefaultTimeout() : timeout;
if (StringUtils.isEmpty(resolvedGroup)) {
resolvedGroup = DEFAULT_GROUP;
if (StringUtils.isEmpty(group)) {
group = DEFAULT_GROUP;
}
return configService.getConfig(key, resolvedGroup, nacosTimeout);
return configService.getConfig(key, group, nacosTimeout);
} catch (NacosException e) {
logger.error(e.getMessage());
}
@ -241,9 +235,8 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
@Override
public boolean publishConfig(String key, String group, String content) {
boolean published = false;
String resolvedGroup = resolveGroup(group);
try {
published = configService.publishConfig(key, resolvedGroup, content);
published = configService.publishConfig(key, group, content);
} catch (NacosException e) {
logger.error(e.getErrMsg(), e);
}
@ -251,18 +244,16 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
}
@Override
public boolean publishConfigCas(String key, String group, String content, Object stat) {
boolean published = false;
String resolvedGroup = resolveGroup(group);
public boolean publishConfigCas(String key, String group, String content, Object ticket) {
try {
if (!(null != stat && stat instanceof String)) {
throw new IllegalArgumentException("nacos publishConfigCas requires stat of string type");
if (!(ticket instanceof String)) {
throw new IllegalArgumentException("nacos publishConfigCas requires string type ticket");
}
published = configService.publishConfigCas(key, resolvedGroup, content, (String) stat);
return configService.publishConfigCas(key, group, content, (String) ticket);
} catch (NacosException e) {
logger.error(e.getErrMsg(), e);
logger.warn("nacos publishConfigCas failed.", e);
return false;
}
return published;
}
@Override
@ -280,9 +271,29 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
public SortedSet<String> getConfigKeys(String group) {
// TODO use Nacos Client API to replace HTTP Open API
SortedSet<String> keys = new TreeSet<>();
try {
Map<String, String> paramsValues = new HashMap<>();
paramsValues.put("search", "accurate");
paramsValues.put("dataId", "");
paramsValues.put("group", group.replace(SLASH_CHAR, HYPHEN_CHAR));
paramsValues.put("pageNo", "1");
paramsValues.put("pageSize", String.valueOf(Integer.MAX_VALUE));
String encoding = getProperty(ENCODE, "UTF-8");
HttpRestResult<String> result = httpAgent.httpGet(GET_CONFIG_KEYS_PATH, emptyMap(), paramsValues, encoding, 5 * 1000);
Stream<String> keysStream = toKeysStream(result.getData());
keysStream.forEach(keys::add);
} catch (Exception e) {
if (logger.isErrorEnabled()) {
logger.error(e.getMessage(), e);
}
}
return keys;
}
@Override
public boolean removeConfig(String key, String group) {
boolean removed = false;
@ -361,15 +372,6 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
}
protected String buildListenerKey(String key, String group) {
return key + HYPHEN_CHAR + resolveGroup(group);
}
protected String resolveGroup(String group) {
return isBlank(group) ? group : group.replace(SLASH_CHAR, HYPHEN_CHAR);
}
@Override
public boolean hasSupportCas() {
return true;
return key + HYPHEN_CHAR + group;
}
}

24
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java
View File

@ -26,6 +26,8 @@ import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
import org.apache.zookeeper.data.Stat;
import java.util.Collection;
import java.util.Set;
import java.util.concurrent.Executor;
@ -49,7 +51,6 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration
private static final int DEFAULT_QUEUE = 10000;
private static final Long THREAD_KEEP_ALIVE_TIME = 0L;
ZookeeperDynamicConfiguration(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
this.url = url;
@ -92,10 +93,18 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration
}
@Override
public boolean publishConfigCas(String key, String group, String content, Object stat) {
String pathKey = buildPathKey(group, key);
zkClient.createOrUpdate(pathKey, content, false, stat);
return true;
public boolean publishConfigCas(String key, String group, String content, Object ticket) {
try {
if (ticket != null && !(ticket instanceof Stat)) {
throw new IllegalArgumentException("zookeeper publishConfigCas requires stat type ticket");
}
String pathKey = buildPathKey(group, key);
zkClient.createOrUpdate(pathKey, content, false, ticket == null ? 0 : ((Stat) ticket).getVersion());
return true;
} catch (Exception e) {
logger.warn("zookeeper publishConfigCas failed.", e);
return false;
}
}
@Override
@ -134,9 +143,4 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration
zkClient.removeDataListener(pathKey, cacheListener);
}
}
@Override
public boolean hasSupportCas() {
return true;
}
}

7
dubbo-configcenter/dubbo-configcenter-zookeeper/src/test/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfigurationTest.java
View File

@ -42,6 +42,7 @@ import java.util.concurrent.CountDownLatch;
import static java.util.Arrays.asList;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@ -150,11 +151,7 @@ public class ZookeeperDynamicConfigurationTest {
configItem = configuration.getConfigItem(key, group);
assertEquals("test", configItem.getContent());
assertTrue(configuration.publishConfigCas(key, group, "newtest", configItem.getStat()));
try {
configuration.publishConfigCas(key, group, "newtest2", configItem.getStat());
} catch (Exception e) {
assertTrue(e.getMessage().contains("KeeperErrorCode = BadVersion"));
}
assertFalse(configuration.publishConfigCas(key, group, "newtest2", configItem.getStat()));
assertEquals("newtest", configuration.getConfigItem(key, group).getContent());
}

1
dubbo-demo/dubbo-demo-generic-call/src/main/java/org/apache/dubbo/demo/consumer/GenericApplication.java
View File

@ -47,7 +47,6 @@ public class GenericApplication {
ApplicationConfig applicationConfig = new ApplicationConfig("demo-consumer");
Map<String, String> parameters = new HashMap<>();
parameters.put("mapping-type", "metadata");
applicationConfig.setParameters(parameters);
MetadataReportConfig metadataReportConfig = new MetadataReportConfig();

12
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/pom.xml
View File

@ -30,10 +30,6 @@
<skip_maven_deploy>true</skip_maven_deploy>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-demo-interface</artifactId>
@ -63,6 +59,14 @@
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-configcenter-nacos</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-spring</artifactId>

3
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-consumer/src/main/resources/spring/dubbo-consumer.xml
View File

@ -22,10 +22,9 @@
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<dubbo:application name="demo-consumer">
<dubbo:parameter key="mapping-type" value="metadata"/>
</dubbo:application>
<!-- <dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/>-->
<dubbo:metadata-report address="zookeeper://127.0.0.1:2181"/>
<dubbo:registry address="zookeeper://127.0.0.1:2181"/>

4
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/pom.xml
View File

@ -66,6 +66,10 @@
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-dubbo</artifactId>

1
dubbo-demo/dubbo-demo-xml/dubbo-demo-xml-provider/src/main/resources/spring/dubbo-provider.xml
View File

@ -22,7 +22,6 @@
http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
<dubbo:application name="demo-provider">
<dubbo:parameter key="mapping-type" value="metadata"/>
</dubbo:application>
<dubbo:config-center address="zookeeper://127.0.0.1:2181"/>

8
dubbo-distribution/dubbo-all/pom.xml
View File

@ -283,6 +283,13 @@
<scope>compile</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-redis</artifactId>
@ -427,6 +434,7 @@
<include>org.apache.dubbo:dubbo-metadata-api</include>
<include>org.apache.dubbo:dubbo-metadata-report-redis</include>
<include>org.apache.dubbo:dubbo-metadata-report-zookeeper</include>
<include>org.apache.dubbo:dubbo-metadata-report-nacos</include>
<include>org.apache.dubbo:dubbo-qos</include>
<include>org.apache.dubbo:dubbo-auth</include>
<include>org.apache.dubbo:dubbo-filter-cache</include>

5
dubbo-distribution/dubbo-bom/pom.xml
View File

@ -233,6 +233,11 @@
<artifactId>dubbo-metadata-report-zookeeper</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-nacos</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-redis</artifactId>

158
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/DynamicConfigurationServiceNameMapping.java
View File

@ -1,158 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import static java.lang.String.valueOf;
import static java.util.Arrays.asList;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SEPARATOR;
import static org.apache.dubbo.common.utils.CollectionUtils.ofSet;
import static org.apache.dubbo.rpc.model.ApplicationModel.getName;
/**
* The {@link ServiceNameMapping} implementation based on {@link DynamicConfiguration}
*/
public class DynamicConfigurationServiceNameMapping implements ServiceNameMapping {
private static final List<String> IGNORED_SERVICE_INTERFACES = asList(MetadataService.class.getName());
private final Logger logger = LoggerFactory.getLogger(getClass());
private static final int PUBLISH_CONFIG_RETRY_TIMES = 6;
@Override
public void map(URL url) {
String serviceInterface = url.getServiceInterface();
String group = url.getGroup();
String version = url.getVersion();
String protocol = url.getProtocol();
if (IGNORED_SERVICE_INTERFACES.contains(serviceInterface)) {
return;
}
DynamicConfiguration dynamicConfiguration = DynamicConfiguration.getDynamicConfiguration();
// the Dubbo Service Key as group
// the service(application) name as key
// It does matter whatever the content is, we just need a record
String key = getName();
String content = valueOf(System.currentTimeMillis());
execute(() -> {
dynamicConfiguration.publishConfig(key, ServiceNameMapping.buildGroup(serviceInterface, group, version, protocol), content);
if (logger.isDebugEnabled()) {
logger.info(String.format("Dubbo service[%s] mapped to interface name[%s].",
group, serviceInterface));
}
});
}
@Override
public void mapWithCas(URL url) {
String serviceInterface = url.getServiceInterface();
if (IGNORED_SERVICE_INTERFACES.contains(serviceInterface)) {
return;
}
execute(() -> {
publishConfigCas(serviceInterface, DEFAULT_MAPPING_GROUP, getName());
});
}
@Override
public Set<String> getAndListen(URL url, MappingListener mappingListener) {
String serviceInterface = url.getServiceInterface();
String group = url.getGroup();
String version = url.getVersion();
String protocol = url.getProtocol();
DynamicConfiguration dynamicConfiguration = DynamicConfiguration.getDynamicConfiguration();
Set<String> serviceNames = new LinkedHashSet<>();
execute(() -> {
Set<String> keys = dynamicConfiguration
.getConfigKeys(ServiceNameMapping.buildGroup(serviceInterface, group, version, protocol));
serviceNames.addAll(keys);
});
return Collections.unmodifiableSet(serviceNames);
}
@Override
public Set<String> getAndListenWithNewStore(URL url, MappingListener mappingListener) {
String serviceInterface = url.getServiceInterface();
DynamicConfiguration dynamicConfiguration = DynamicConfiguration.getDynamicConfiguration();
Set<String> serviceNames = new LinkedHashSet<>();
execute(() -> {
String configContent = dynamicConfiguration.getConfig(serviceInterface, DEFAULT_MAPPING_GROUP);
if (null != configContent) {
String[] split = StringUtils.split(configContent, CommonConstants.COMMA_SEPARATOR_CHAR);
serviceNames.addAll(ofSet(split));
}
});
return Collections.unmodifiableSet(serviceNames);
}
/**
* publish config with cas.
*
* @param key
* @param group
* @param appName
* @return
*/
private boolean publishConfigCas(String key, String group, String appName) {
int currentRetryTimes = 1;
boolean result = false;
DynamicConfiguration dynamicConfiguration = DynamicConfiguration.getDynamicConfiguration();
String newConfigContent = appName;
do {
ConfigItem configItem = dynamicConfiguration.getConfigItem(key, group);
String oldConfigContent = configItem.getContent();
if (StringUtils.isNotEmpty(oldConfigContent)) {
boolean contains = StringUtils.isContains(configItem.getContent(), appName);
if (contains) {
return true;
}
newConfigContent = oldConfigContent + COMMA_SEPARATOR + appName;
}
result = dynamicConfiguration.publishConfigCas(key, group, newConfigContent, configItem.getStat());
} while (!result && currentRetryTimes++ <= PUBLISH_CONFIG_RETRY_TIMES);
return result;
}
private void execute(Runnable runnable) {
try {
runnable.run();
} catch (Throwable e) {
if (logger.isWarnEnabled()) {
logger.warn(e.getMessage(), e);
}
}
}
}

24
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MappingChangedEvent.java
View File

@ -19,30 +19,26 @@ package org.apache.dubbo.metadata;
import java.util.Set;
public class MappingChangedEvent {
private String serviceKey;
private Set<String> apps;
private final String serviceKey;
private final Set<String> apps;
public MappingChangedEvent(String serviceKey, Set<String> apps) {
this.serviceKey = serviceKey;
this.apps = apps;
}
public String getServiceKey() {
return serviceKey;
}
public void setServiceKey(String serviceKey) {
this.serviceKey = serviceKey;
}
public Set<String> getApps() {
return apps;
}
public void setApps(Set<String> apps) {
this.apps = apps;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("{serviceKey: ").append(serviceKey).append(", apps: ");
sb.append(apps.toString()).append("}");
return sb.toString();
return "{serviceKey: " + serviceKey + ", apps: " +
apps.toString() + "}";
}
}

77
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java
View File

@ -18,13 +18,14 @@ package org.apache.dubbo.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.common.utils.StringUtils;
import java.util.Collections;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.CONFIG_MAPPING_TYPE;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static java.util.Collections.emptySet;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SEPARATOR;
import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
import static org.apache.dubbo.common.utils.StringUtils.SLASH;
@ -33,8 +34,9 @@ import static org.apache.dubbo.common.utils.StringUtils.SLASH;
*
* @since 2.7.5
*/
@SPI("config")
@SPI("metadata")
public interface ServiceNameMapping {
String DEFAULT_MAPPING_GROUP = "mapping";
/**
@ -42,15 +44,6 @@ public interface ServiceNameMapping {
*/
void map(URL url);
/**
* Map the specified Dubbo service interface, group, version and protocol to current Dubbo service name with cas.
*
* @param url
*/
default void mapWithCas(URL url) {
}
/**
* Get the service names from the specified Dubbo service interface, group, version and protocol
*
@ -59,51 +52,31 @@ public interface ServiceNameMapping {
Set<String> getAndListen(URL url, MappingListener mappingListener);
/**
* service name mapping new store structure.
* interface(key)
* -- mapping(group)
* --appName1,appName2,appName3(content)
* @param url
* @param mappingListener
* @return
* Get the default extension of {@link ServiceNameMapping}
*
* @return non-null {@link ServiceNameMapping}
*/
default Set<String> getAndListenWithNewStore(URL url, MappingListener mappingListener){
return Collections.emptySet();
};
static ServiceNameMapping getDefaultExtension() {
return getExtensionLoader(ServiceNameMapping.class).getDefaultExtension();
}
default Set<String> get(URL url) {
return getAndListen(url, null);
static String buildMappingKey(URL url) {
return buildGroup(url.getServiceInterface());
}
static String buildGroup(String serviceInterface) {
//the issue : https://github.com/apache/dubbo/issues/4671
return DEFAULT_MAPPING_GROUP + SLASH + serviceInterface;
}
static String toStringKeys(Set<String> serviceNames) {
return serviceNames.toString();
}
/**
* Get the default extension of {@link ServiceNameMapping}
*
* @return non-null {@link ServiceNameMapping}
* @see DynamicConfigurationServiceNameMapping
*/
static ServiceNameMapping getDefaultExtension() {
return getExtensionLoader(ServiceNameMapping.class).getDefaultExtension();
}
static ServiceNameMapping getExtension(String name) {
return getExtensionLoader(ServiceNameMapping.class).getExtension(name == null ? CONFIG_MAPPING_TYPE : name);
}
static String buildMappingKey(URL url) {
return buildGroup(url.getServiceInterface(), url.getGroup(), url.getVersion(), url.getParameter(PROTOCOL_KEY, DUBBO));
}
static String buildGroup(String serviceInterface, String group, String version, String protocol) {
// the issue : https://github.com/apache/dubbo/issues/4671
// StringBuilder groupBuilder = new StringBuilder(serviceInterface)
// .append(KEY_SEPARATOR).append(defaultString(group))
// .append(KEY_SEPARATOR).append(defaultString(version))
// .append(KEY_SEPARATOR).append(defaultString(protocol));
// return groupBuilder.toString();
return DEFAULT_MAPPING_GROUP + SLASH + serviceInterface;
static Set<String> getAppNames(String content) {
if (StringUtils.isBlank(content)) {
return emptySet();
}
return new HashSet<>(Arrays.asList(content.split(COMMA_SEPARATOR)));
}
}

72
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMappingHandler.java
View File

@ -1,72 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
public class ServiceNameMappingHandler {
public static final String DUBBO_SERVICENAME_STORE = "dubbo.application.service-name.store";
private static final Logger logger = LoggerFactory.getLogger(ServiceNameMappingHandler.class);
private static final ServiceNameMappingStoreEnum DEFAULT_STORE_TYPE = ServiceNameMappingStoreEnum.BOTH_STORAGE;
private final ServiceNameMapping serviceNameMapping;
private final URL url;
public ServiceNameMappingHandler(ServiceNameMapping serviceNameMapping, URL url) {
this.serviceNameMapping = serviceNameMapping;
this.url = url;
}
public static void map(ServiceNameMapping serviceNameMapping, URL url) {
new ServiceNameMappingHandler(serviceNameMapping, url).init();
}
public void init() {
DynamicConfiguration dynamicConfiguration = DynamicConfiguration.getDynamicConfiguration();
ServiceNameMappingStoreEnum storeType = DEFAULT_STORE_TYPE;
boolean hasSupportCas = dynamicConfiguration.hasSupportCas();
if (!hasSupportCas) {
storeType = ServiceNameMappingStoreEnum.APPLICANT_INTERFACE_STORAGE;
}
doMap(storeType);
}
public void doMap(ServiceNameMappingStoreEnum storeType) {
if (null == storeType) {
throw new IllegalStateException("storeType of serviceNameMapping cannot be null");
}
switch (storeType) {
case INTERFACE_APPLICATION_STORAGE:
serviceNameMapping.mapWithCas(url);
break;
case APPLICANT_INTERFACE_STORAGE:
serviceNameMapping.map(url);
break;
case BOTH_STORAGE:
serviceNameMapping.map(url);
serviceNameMapping.mapWithCas(url);
break;
default:
serviceNameMapping.map(url);
}
}
}

51
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMappingStoreEnum.java
View File

@ -1,51 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata;
import java.util.HashMap;
import java.util.Map;
public enum ServiceNameMappingStoreEnum {
INTERFACE_APPLICATION_STORAGE("interface_application_storage"),
APPLICANT_INTERFACE_STORAGE("applicant_interface_storage"),
BOTH_STORAGE("both_storage");
private String storeType;
private static final Map<String, ServiceNameMappingStoreEnum> storeMap = new HashMap<>();
static {
for (ServiceNameMappingStoreEnum serviceNameMappingStoreEnum : values()) {
storeMap.put(serviceNameMappingStoreEnum.getStoreType(), serviceNameMappingStoreEnum);
}
}
ServiceNameMappingStoreEnum(String storeType) {
this.storeType = storeType;
}
public String getStoreType() {
return this.storeType;
}
public static ServiceNameMappingStoreEnum getStoreEnumByStoreType(String storeType) {
return storeMap.get(storeType);
}
}

26
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/report/MetadataReport.java
View File

@ -18,6 +18,7 @@ package org.apache.dubbo.metadata.report;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.metadata.MappingListener;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.definition.model.ServiceDefinition;
@ -48,17 +49,6 @@ public interface MetadataReport {
return null;
}
/**
* Service<-->Application Mapping -- START
**/
default Set<String> getServiceAppMapping(String serviceKey, MappingListener listener, URL url) {
return Collections.emptySet();
}
default void registerServiceAppMapping(String serviceKey, String application, URL url) {
return;
}
/**
* deprecated or need triage
**/
@ -76,4 +66,18 @@ public interface MetadataReport {
List<String> getSubscribedURLs(SubscriberMetadataIdentifier subscriberMetadataIdentifier);
default ConfigItem getConfigItem(String key, String group) {
return new ConfigItem();
}
default boolean registerServiceAppMapping(String serviceInterface, String defaultMappingGroup, String newConfigContent, Object ticket) {
return false;
}
/**
* Service<-->Application Mapping -- START
**/
default Set<String> getServiceAppMapping(String serviceKey, MappingListener listener, URL url) {
return Collections.emptySet();
}
}

1
dubbo-metadata/dubbo-metadata-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.metadata.ServiceNameMapping
View File

@ -1 +0,0 @@
config=org.apache.dubbo.metadata.DynamicConfigurationServiceNameMapping

2
dubbo-metadata/dubbo-metadata-api/src/test/java/org/apache/dubbo/metadata/ServiceNameMappingTest.java
View File

@ -34,7 +34,7 @@ import static java.util.Arrays.asList;
import static java.util.Collections.singleton;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.SUBSCRIBED_SERVICE_NAMES_KEY;
import static org.apache.dubbo.metadata.DynamicConfigurationServiceNameMapping.buildGroup;
import static org.apache.dubbo.common.config.configcenter.DynamicConfigurationServiceNameMapping.buildGroup;
import static org.apache.dubbo.metadata.ServiceNameMapping.getDefaultExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;

43
dubbo-metadata/dubbo-metadata-report-nacos/pom.xml Normal file
View File

@ -0,0 +1,43 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You 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
~
~ http://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.
-->
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<parent>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata</artifactId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>dubbo-metadata-report-nacos</artifactId>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-configcenter-nacos</artifactId>
<version>${project.parent.version}</version>
</dependency>
</dependencies>
</project>

78
dubbo-metadata/dubbo-metadata-report-nacos/src/main/java/org/apache/dubbo/metadata/store/nacos/NacosConfigServiceWrapper.java Normal file
View File

@ -0,0 +1,78 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata.store.nacos;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException;
import static org.apache.dubbo.common.utils.StringUtils.HYPHEN_CHAR;
import static org.apache.dubbo.common.utils.StringUtils.SLASH_CHAR;
public class NacosConfigServiceWrapper {
private static final String INNERCLASS_SYMBOL = "$";
private static final String INNERCLASS_COMPATIBLE_SYMBOL = "___";
private static final long DEFAULT_TIMEOUT = 3000L;
private ConfigService configService;
public NacosConfigServiceWrapper(ConfigService configService) {
this.configService = configService;
}
public ConfigService getConfigService() {
return configService;
}
public void addListener(String dataId, String group, Listener listener) throws NacosException {
configService.addListener(handleInnerSymbol(dataId), handleInnerSymbol(group), listener);
}
public String getConfig(String dataId, String group) throws NacosException {
return configService.getConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), DEFAULT_TIMEOUT);
}
public String getConfig(String dataId, String group, long timeout) throws NacosException {
return configService.getConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), timeout);
}
public boolean publishConfig(String dataId, String group, String content) throws NacosException {
return configService.publishConfig(handleInnerSymbol(dataId), handleInnerSymbol(group), content);
}
public boolean publishConfigCas(String dataId, String group, String content, String casMd5) throws NacosException {
return configService.publishConfigCas(handleInnerSymbol(dataId), handleInnerSymbol(group), content, casMd5);
}
public boolean removeConfig(String dataId, String group) throws NacosException {
return configService.removeConfig(handleInnerSymbol(dataId), handleInnerSymbol(group));
}
/**
* see {@link com.alibaba.nacos.client.config.utils.ParamUtils#isValid(java.lang.String)}
*/
private String handleInnerSymbol(String data) {
if (data == null) {
return null;
}
return data.replace(INNERCLASS_SYMBOL, INNERCLASS_COMPATIBLE_SYMBOL).replace(SLASH_CHAR, HYPHEN_CHAR);
}
}

405
dubbo-metadata/dubbo-metadata-report-nacos/src/main/java/org/apache/dubbo/metadata/store/nacos/NacosMetadataReport.java Normal file
View File

@ -0,0 +1,405 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata.store.nacos;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigChangeType;
import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
import org.apache.dubbo.common.utils.MD5Utils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MappingChangedEvent;
import org.apache.dubbo.metadata.MappingListener;
import org.apache.dubbo.metadata.ServiceNameMapping;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.report.identifier.BaseMetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.KeyTypeEnum;
import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.ServiceMetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier;
import org.apache.dubbo.metadata.report.support.AbstractMetadataReport;
import org.apache.dubbo.rpc.RpcException;
import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.PropertyKeyConst;
import com.alibaba.nacos.api.config.listener.AbstractSharedListener;
import com.alibaba.nacos.api.exception.NacosException;
import com.google.gson.Gson;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executor;
import static com.alibaba.nacos.api.PropertyKeyConst.SERVER_ADDR;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
import static org.apache.dubbo.common.utils.StringConstantFieldValuePredicate.of;
import static org.apache.dubbo.common.utils.StringUtils.HYPHEN_CHAR;
import static org.apache.dubbo.metadata.ServiceNameMapping.DEFAULT_MAPPING_GROUP;
import static org.apache.dubbo.metadata.ServiceNameMapping.getAppNames;
/**
* metadata report impl for nacos
*/
public class NacosMetadataReport extends AbstractMetadataReport {
private NacosConfigServiceWrapper configService;
private Gson gson = new Gson();
/**
* The group used to store metadata in Nacos
*/
private String group;
private Map<String, NacosConfigListener> watchListenerMap = new ConcurrentHashMap<>();
private Map<String, MappingDataListener> casListenerMap = new ConcurrentHashMap<>();
public NacosMetadataReport(URL url) {
super(url);
this.configService = buildConfigService(url);
group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
}
public NacosConfigServiceWrapper buildConfigService(URL url) {
Properties nacosProperties = buildNacosProperties(url);
try {
configService = new NacosConfigServiceWrapper(NacosFactory.createConfigService(nacosProperties));
} catch (NacosException e) {
if (logger.isErrorEnabled()) {
logger.error(e.getErrMsg(), e);
}
throw new IllegalStateException(e);
}
return configService;
}
private Properties buildNacosProperties(URL url) {
Properties properties = new Properties();
setServerAddr(url, properties);
setProperties(url, properties);
return properties;
}
private void setServerAddr(URL url, Properties properties) {
StringBuilder serverAddrBuilder =
new StringBuilder(url.getHost()) // Host
.append(":")
.append(url.getPort()); // Port
// Append backup parameter as other servers
String backup = url.getParameter(BACKUP_KEY);
if (backup != null) {
serverAddrBuilder.append(",").append(backup);
}
String serverAddr = serverAddrBuilder.toString();
properties.put(SERVER_ADDR, serverAddr);
}
private static void setProperties(URL url, Properties properties) {
// Get the parameters from constants
Map<String, String> parameters = url.getParameters(of(PropertyKeyConst.class));
// Put all parameters
properties.putAll(parameters);
}
private static void putPropertyIfAbsent(URL url, Properties properties, String propertyName) {
String propertyValue = url.getParameter(propertyName);
if (StringUtils.isNotEmpty(propertyValue)) {
properties.setProperty(propertyName, propertyValue);
}
}
private static void putPropertyIfAbsent(URL url, Properties properties, String propertyName, String defaultValue) {
String propertyValue = url.getParameter(propertyName);
if (StringUtils.isNotEmpty(propertyValue)) {
properties.setProperty(propertyName, propertyValue);
} else {
properties.setProperty(propertyName, defaultValue);
}
}
@Override
public void publishAppMetadata(SubscriberMetadataIdentifier identifier, MetadataInfo metadataInfo) {
String content = gson.toJson(metadataInfo);
try {
configService.publishConfig(identifier.getApplication(), identifier.getRevision(), content);
} catch (NacosException e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
@Override
public MetadataInfo getAppMetadata(SubscriberMetadataIdentifier identifier, Map<String, String> instanceMetadata) {
try {
String content = configService.getConfig(identifier.getApplication(), identifier.getRevision(), 3000L);
return gson.fromJson(content, MetadataInfo.class);
} catch (NacosException e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
@Override
protected void doStoreProviderMetadata(MetadataIdentifier providerMetadataIdentifier, String serviceDefinitions) {
this.storeMetadata(providerMetadataIdentifier, serviceDefinitions);
}
@Override
protected void doStoreConsumerMetadata(MetadataIdentifier consumerMetadataIdentifier, String value) {
this.storeMetadata(consumerMetadataIdentifier, value);
}
@Override
protected void doSaveMetadata(ServiceMetadataIdentifier serviceMetadataIdentifier, URL url) {
storeMetadata(serviceMetadataIdentifier, URL.encode(url.toFullString()));
}
@Override
protected void doRemoveMetadata(ServiceMetadataIdentifier serviceMetadataIdentifier) {
deleteMetadata(serviceMetadataIdentifier);
}
@Override
protected List<String> doGetExportedURLs(ServiceMetadataIdentifier metadataIdentifier) {
String content = getConfig(metadataIdentifier);
if (StringUtils.isEmpty(content)) {
return Collections.emptyList();
}
return new ArrayList<String>(Arrays.asList(URL.decode(content)));
}
@Override
protected void doSaveSubscriberData(SubscriberMetadataIdentifier subscriberMetadataIdentifier, String urlListStr) {
storeMetadata(subscriberMetadataIdentifier, urlListStr);
}
@Override
protected String doGetSubscribedURLs(SubscriberMetadataIdentifier subscriberMetadataIdentifier) {
return getConfig(subscriberMetadataIdentifier);
}
@Override
public String getServiceDefinition(MetadataIdentifier metadataIdentifier) {
return getConfig(metadataIdentifier);
}
@Override
public boolean registerServiceAppMapping(String key, String group, String content, Object ticket) {
try {
if (!(ticket instanceof String)) {
throw new IllegalArgumentException("nacos publishConfigCas requires string type ticket");
}
return configService.publishConfigCas(key, group, content, (String) ticket);
} catch (NacosException e) {
logger.warn("nacos publishConfigCas failed.", e);
return false;
}
}
@Override
public ConfigItem getConfigItem(String key, String group) {
String content = getConfig(key, group);
String casMd5 = "";
if (StringUtils.isNotEmpty(content)) {
casMd5 = MD5Utils.getMd5(content);
}
return new ConfigItem(content, casMd5);
}
@Override
public Set<String> getServiceAppMapping(String serviceKey, MappingListener listener, URL url) {
String group = DEFAULT_MAPPING_GROUP;
if (null == casListenerMap.get(buildListenerKey(serviceKey, group))) {
addCasServiceMappingListener(serviceKey, group, listener);
}
String content = getConfig(serviceKey, group);
return ServiceNameMapping.getAppNames(content);
}
private String getConfig(String dataId, String group) {
try {
return configService.getConfig(dataId, group);
} catch (NacosException e) {
logger.error(e.getMessage());
}
return null;
}
private void addCasServiceMappingListener(String serviceKey, String group, MappingListener listener) {
MappingDataListener mappingDataListener = casListenerMap.computeIfAbsent(buildListenerKey(serviceKey, group), k -> new MappingDataListener(serviceKey, group));
mappingDataListener.addListeners(listener);
addListener(serviceKey, DEFAULT_MAPPING_GROUP, mappingDataListener);
}
public void addListener(String key, String group, ConfigurationListener listener) {
String listenerKey = buildListenerKey(key, group);
NacosConfigListener nacosConfigListener =
watchListenerMap.computeIfAbsent(listenerKey, k -> createTargetListener(key, group));
nacosConfigListener.addListener(listener);
try {
configService.addListener(key, group, nacosConfigListener);
} catch (NacosException e) {
logger.error(e.getMessage());
}
}
private NacosConfigListener createTargetListener(String key, String group) {
NacosConfigListener configListener = new NacosConfigListener();
configListener.fillContext(key, group);
return configListener;
}
private String buildListenerKey(String key, String group) {
return key + HYPHEN_CHAR + group;
}
private void storeMetadata(BaseMetadataIdentifier identifier, String value) {
try {
boolean publishResult = configService.publishConfig(identifier.getUniqueKey(KeyTypeEnum.UNIQUE_KEY), group, value);
if (!publishResult) {
throw new RuntimeException("publish nacos metadata failed");
}
} catch (Throwable t) {
logger.error("Failed to put " + identifier + " to nacos " + value + ", cause: " + t.getMessage(), t);
throw new RpcException("Failed to put " + identifier + " to nacos " + value + ", cause: " + t.getMessage(), t);
}
}
private void deleteMetadata(BaseMetadataIdentifier identifier) {
try {
boolean publishResult = configService.removeConfig(identifier.getUniqueKey(KeyTypeEnum.UNIQUE_KEY), group);
if (!publishResult) {
throw new RuntimeException("remove nacos metadata failed");
}
} catch (Throwable t) {
logger.error("Failed to remove " + identifier + " from nacos , cause: " + t.getMessage(), t);
throw new RpcException("Failed to remove " + identifier + " from nacos , cause: " + t.getMessage(), t);
}
}
private String getConfig(BaseMetadataIdentifier identifier) {
try {
return configService.getConfig(identifier.getUniqueKey(KeyTypeEnum.UNIQUE_KEY), group, 3000L);
} catch (Throwable t) {
logger.error("Failed to get " + identifier + " from nacos , cause: " + t.getMessage(), t);
throw new RpcException("Failed to get " + identifier + " from nacos , cause: " + t.getMessage(), t);
}
}
public class NacosConfigListener extends AbstractSharedListener {
private Set<ConfigurationListener> listeners = new CopyOnWriteArraySet<>();
/**
* cache data to store old value
*/
private Map<String, String> cacheData = new ConcurrentHashMap<>();
@Override
public Executor getExecutor() {
return null;
}
/**
* receive
*
* @param dataId data ID
* @param group group
* @param configInfo content
*/
@Override
public void innerReceive(String dataId, String group, String configInfo) {
String oldValue = cacheData.get(dataId);
ConfigChangedEvent event = new ConfigChangedEvent(dataId, group, configInfo, getChangeType(configInfo, oldValue));
if (configInfo == null) {
cacheData.remove(dataId);
} else {
cacheData.put(dataId, configInfo);
}
listeners.forEach(listener -> listener.process(event));
}
void addListener(ConfigurationListener configurationListener) {
this.listeners.add(configurationListener);
}
void removeListener(ConfigurationListener configurationListener) {
this.listeners.remove(configurationListener);
}
private ConfigChangeType getChangeType(String configInfo, String oldValue) {
if (StringUtils.isBlank(configInfo)) {
return ConfigChangeType.DELETED;
}
if (StringUtils.isBlank(oldValue)) {
return ConfigChangeType.ADDED;
}
return ConfigChangeType.MODIFIED;
}
}
static class MappingDataListener implements ConfigurationListener {
private String dataId;
private String groupId;
private String serviceKey;
private Set<MappingListener> listeners;
public MappingDataListener(String dataId, String groupId) {
this.serviceKey = dataId;
this.dataId = dataId;
this.groupId = groupId;
this.listeners = new HashSet<>();
}
public void addListeners(MappingListener mappingListener) {
listeners.add(mappingListener);
}
@Override
public void process(ConfigChangedEvent event) {
if (ConfigChangeType.DELETED == event.getChangeType()) {
return;
}
if (!dataId.equals(event.getKey()) || !groupId.equals(event.getGroup())) {
return;
}
Set<String> apps = getAppNames(event.getContent());
MappingChangedEvent mappingChangedEvent = new MappingChangedEvent(serviceKey, apps);
listeners.forEach(listener -> listener.onEvent(mappingChangedEvent));
}
}
}

32
dubbo-metadata/dubbo-metadata-report-nacos/src/main/java/org/apache/dubbo/metadata/store/nacos/NacosMetadataReportFactory.java Normal file
View File

@ -0,0 +1,32 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.metadata.store.nacos;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.metadata.report.MetadataReport;
import org.apache.dubbo.metadata.report.support.AbstractMetadataReportFactory;
/**
* metadata report factory impl for nacos
*/
public class NacosMetadataReportFactory extends AbstractMetadataReportFactory {
@Override
protected MetadataReport createMetadataReport(URL url) {
return new NacosMetadataReport(url);
}
}

1
dubbo-metadata/dubbo-metadata-report-nacos/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.metadata.report.MetadataReportFactory Normal file
View File

@ -0,0 +1 @@
nacos=org.apache.dubbo.metadata.store.nacos.NacosMetadataReportFactory

2
dubbo-metadata/dubbo-metadata-report-zookeeper/pom.xml
View File

@ -17,8 +17,8 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>dubbo-metadata</artifactId>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata</artifactId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>

97
dubbo-metadata/dubbo-metadata-report-zookeeper/src/main/java/org/apache/dubbo/metadata/store/zookeeper/ZookeeperMetadataReport.java
View File

@ -17,9 +17,9 @@
package org.apache.dubbo.metadata.store.zookeeper;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MappingChangedEvent;
import org.apache.dubbo.metadata.MappingListener;
@ -30,16 +30,16 @@ import org.apache.dubbo.metadata.report.identifier.MetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.ServiceMetadataIdentifier;
import org.apache.dubbo.metadata.report.identifier.SubscriberMetadataIdentifier;
import org.apache.dubbo.metadata.report.support.AbstractMetadataReport;
import org.apache.dubbo.remoting.zookeeper.ChildListener;
import org.apache.dubbo.remoting.zookeeper.DataListener;
import org.apache.dubbo.remoting.zookeeper.EventType;
import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
import org.apache.dubbo.remoting.zookeeper.ZookeeperTransporter;
import com.google.gson.Gson;
import org.apache.zookeeper.data.Stat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@ -47,6 +47,8 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR;
import static org.apache.dubbo.metadata.ServiceNameMapping.DEFAULT_MAPPING_GROUP;
import static org.apache.dubbo.metadata.ServiceNameMapping.getAppNames;
/**
* ZookeeperMetadataReport
@ -61,7 +63,8 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
private Gson gson = new Gson();
private Map<String, MappingChildListener> listenerMap = new ConcurrentHashMap<>();
private Map<String, MappingDataListener> casListenerMap = new ConcurrentHashMap<>();
public ZookeeperMetadataReport(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
@ -76,7 +79,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
zkClient = zookeeperTransporter.connect(url);
}
String toRootDir() {
protected String toRootDir() {
if (root.equals(PATH_SEPARATOR)) {
return root;
}
@ -109,7 +112,7 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
if (StringUtils.isEmpty(content)) {
return Collections.emptyList();
}
return new ArrayList<String>(Arrays.asList(URL.decode(content)));
return new ArrayList<>(Collections.singletonList(URL.decode(content)));
}
@Override
@ -143,16 +146,6 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
}
}
@Override
public void registerServiceAppMapping(String serviceKey, String application, URL url) {
String path = toRootDir() + serviceKey + PATH_SEPARATOR + application;
if (StringUtils.isBlank(zkClient.getContent(path))) {
Map<String, String> value = new HashMap<>();
value.put("timestamp", String.valueOf(System.currentTimeMillis()));
zkClient.create(path, gson.toJson(value), false);
}
}
@Override
public MetadataInfo getAppMetadata(SubscriberMetadataIdentifier identifier, Map<String, String> instanceMetadata) {
String content = zkClient.getContent(getNodePath(identifier));
@ -161,33 +154,51 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
@Override
public Set<String> getServiceAppMapping(String serviceKey, MappingListener listener, URL url) {
Set<String> appNameSet = new HashSet<>();
String path = toRootDir() + serviceKey;
List<String> appNameList = zkClient.getChildren(path);
if (!CollectionUtils.isEmpty(appNameList)) {
appNameSet.addAll(appNameList);
String path = buildPathKey(DEFAULT_MAPPING_GROUP, serviceKey);
if (null == casListenerMap.get(path)) {
addCasServiceMappingListener(path, serviceKey, listener);
}
if (null == listenerMap.get(path)) {
zkClient.create(path, false);
addServiceMappingListener(path, serviceKey, listener);
}
return appNameSet;
return getAppNames(zkClient.getContent(path));
}
private void addServiceMappingListener(String path, String serviceKey, MappingListener listener) {
MappingChildListener mappingChildListener = listenerMap.computeIfAbsent(path, _k -> new MappingChildListener(serviceKey, path));
mappingChildListener.addListener(listener);
zkClient.addChildListener(path, mappingChildListener);
@Override
public ConfigItem getConfigItem(String serviceKey, String group) {
String path = buildPathKey(group, serviceKey);
return zkClient.getConfigItem(path);
}
private static class MappingChildListener implements ChildListener {
@Override
public boolean registerServiceAppMapping(String key, String group, String content, Object ticket) {
try {
if (ticket != null && !(ticket instanceof Stat)) {
throw new IllegalArgumentException("zookeeper publishConfigCas requires stat type ticket");
}
String pathKey = buildPathKey(group, key);
zkClient.createOrUpdate(pathKey, content, false, ticket == null ? 0 : ((Stat) ticket).getVersion());
return true;
} catch (Exception e) {
logger.warn("zookeeper publishConfigCas failed.", e);
return false;
}
}
private String buildPathKey(String group, String serviceKey) {
return toRootDir() + group + PATH_SEPARATOR + serviceKey;
}
private void addCasServiceMappingListener(String path, String serviceKey, MappingListener listener) {
MappingDataListener mappingDataListener = casListenerMap.computeIfAbsent(path, _k -> new MappingDataListener(serviceKey, path));
mappingDataListener.addListener(listener);
zkClient.addDataListener(path, mappingDataListener);
}
private static class MappingDataListener implements DataListener {
private String serviceKey;
private String path;
private Set<MappingListener> listeners;
public MappingChildListener(String serviceKey, String path) {
public MappingDataListener(String serviceKey, String path) {
this.serviceKey = serviceKey;
this.path = path;
this.listeners = new HashSet<>();
@ -198,10 +209,18 @@ public class ZookeeperMetadataReport extends AbstractMetadataReport {
}
@Override
public void childChanged(String path, List<String> children) {
MappingChangedEvent event = new MappingChangedEvent();
event.setServiceKey(serviceKey);
event.setApps(null != children ? new HashSet<>(children) : null);
public void dataChanged(String path, Object value, EventType eventType) {
if (!this.path.equals(path)) {
return;
}
if (EventType.NodeCreated != eventType && EventType.NodeDataChanged != eventType) {
return;
}
Set<String> apps = getAppNames((String) value);
MappingChangedEvent event = new MappingChangedEvent(serviceKey, apps);
listeners.forEach(mappingListener -> mappingListener.onEvent(event));
}
}

1
dubbo-metadata/pom.xml
View File

@ -33,6 +33,7 @@
<module>dubbo-metadata-processor</module>
<module>dubbo-metadata-report-zookeeper</module>
<module>dubbo-metadata-report-redis</module>
<module>dubbo-metadata-report-nacos</module>
</modules>
</project>

62
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscovery.java
View File

@ -16,27 +16,75 @@
*/
package org.apache.dubbo.registry.client;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.isInstanceUpdated;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.resetInstanceUpdateKey;
public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
private volatile boolean isDestroy;
protected ServiceInstance serviceInstance;
@Override
public ServiceInstance getLocalInstance() {
public final ServiceInstance getLocalInstance() {
return serviceInstance;
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public final void initialize(URL registryURL) throws Exception {
doInitialize(registryURL);
}
public abstract void doInitialize(URL registryURL) throws Exception;
@Override
public final void register(ServiceInstance serviceInstance) throws RuntimeException {
if (ServiceInstanceMetadataUtils.getExportedServicesRevision(serviceInstance) == null) {
ServiceInstanceMetadataUtils.calInstanceRevision(this, serviceInstance);
}
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
doRegister(serviceInstance);
this.serviceInstance = serviceInstance;
}
public abstract void doRegister(ServiceInstance serviceInstance) throws RuntimeException;
@Override
public final void update(ServiceInstance serviceInstance) throws RuntimeException {
if (this.serviceInstance == null) {
this.register(serviceInstance);
return;
}
if (!isInstanceUpdated(serviceInstance)) {
return;
}
doUpdate(serviceInstance);
resetInstanceUpdateKey(serviceInstance);
this.serviceInstance = serviceInstance;
}
public abstract void doUpdate(ServiceInstance serviceInstance) throws RuntimeException;
@Override
public final void unregister(ServiceInstance serviceInstance) throws RuntimeException {
doUnregister(serviceInstance);
}
public abstract void doUnregister(ServiceInstance serviceInstance);
@Override
public final void destroy() throws Exception {
isDestroy = true;
doDestroy();
}
public abstract void doDestroy() throws Exception;
@Override
public final boolean isDestroy() {
return isDestroy;
}
}

18
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/FileSystemServiceDiscovery.java
View File

@ -52,7 +52,7 @@ import static org.apache.dubbo.common.config.configcenter.file.FileSystemDynamic
* @see FileSystemDynamicConfiguration
* @since 2.7.5
*/
public class FileSystemServiceDiscovery implements ServiceDiscovery {
public class FileSystemServiceDiscovery extends AbstractServiceDiscovery {
private final Logger logger = LoggerFactory.getLogger(getClass());
@ -60,10 +60,8 @@ public class FileSystemServiceDiscovery implements ServiceDiscovery {
private FileSystemDynamicConfiguration dynamicConfiguration;
private ServiceInstance serviceInstance;
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
dynamicConfiguration = createDynamicConfiguration(registryURL);
registerDubboShutdownHook();
registerListener();
@ -86,7 +84,7 @@ public class FileSystemServiceDiscovery implements ServiceDiscovery {
}
@Override
public void destroy() throws Exception {
public void doDestroy() throws Exception {
dynamicConfiguration.close();
releaseAndRemoveRegistrationFiles();
}
@ -128,13 +126,9 @@ public class FileSystemServiceDiscovery implements ServiceDiscovery {
return null;
}
@Override
public ServiceInstance getLocalInstance() {
return serviceInstance;
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
String serviceInstanceId = getServiceInstanceId(serviceInstance);
String serviceName = getServiceName(serviceInstance);
@ -168,12 +162,12 @@ public class FileSystemServiceDiscovery implements ServiceDiscovery {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
register(serviceInstance);
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
String key = getServiceInstanceId(serviceInstance);
String group = getServiceName(serviceInstance);
releaseFileLock(key, group);

8
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/SelfHostMetaServiceDiscovery.java
View File

@ -49,6 +49,8 @@ import java.util.concurrent.TimeUnit;
public abstract class SelfHostMetaServiceDiscovery implements ServiceDiscovery {
private volatile boolean isDestroy;
private final Logger logger = LoggerFactory.getLogger(getClass());
private URL registryURL;
@ -124,12 +126,18 @@ public abstract class SelfHostMetaServiceDiscovery implements ServiceDiscovery {
@Override
public void destroy() throws Exception {
isDestroy = true;
doDestroy();
metadataMap.clear();
serviceInstanceRevisionMap.clear();
echoCheckExecutor.shutdown();
}
@Override
public boolean isDestroy() {
return isDestroy;
}
private void updateMetadata(ServiceInstance serviceInstance) {
WritableMetadataService metadataService = WritableMetadataService.getDefaultExtension();
String metadataString = JSONObject.toJSONString(serviceInstance.getMetadata());

2
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java
View File

@ -61,6 +61,8 @@ public interface ServiceDiscovery extends Prioritized {
*/
void destroy() throws Exception;
boolean isDestroy();
// ==================================================================================== //
// =================================== Registration =================================== //

8
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
View File

@ -56,7 +56,6 @@ import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.MAPPING_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
@ -234,8 +233,6 @@ public class ServiceDiscoveryRegistry implements Registry {
logger.warn("Cannot find app mapping for service " + url.getServiceInterface() + ", will not migrate.", e);
}
Set<String> serviceNames = writableMetadataService.getCachedMapping(url);
if (CollectionUtils.isEmpty(subscribedServices)) {
if (check) {
throw new IllegalStateException("Should has at least one way to know which services this interface belongs to, subscription url: " + url);
@ -292,7 +289,7 @@ public class ServiceDiscoveryRegistry implements Registry {
@Override
public boolean isAvailable() {
return !serviceDiscovery.getServices().isEmpty();
return !serviceDiscovery.isDestroy() && !serviceDiscovery.getServices().isEmpty();
}
@Override
@ -468,9 +465,8 @@ public class ServiceDiscoveryRegistry implements Registry {
protected Set<String> findMappedServices(URL registryURL, URL subscribedURL, MappingListener listener) {
Set<String> result = new LinkedHashSet<>();
ServiceNameMapping serviceNameMapping = ServiceNameMapping.getExtension(registryURL.getParameter(MAPPING_KEY));
ServiceNameMapping serviceNameMapping = ServiceNameMapping.getDefaultExtension();
result.addAll(serviceNameMapping.getAndListen(subscribedURL, listener));
result.addAll(serviceNameMapping.getAndListenWithNewStore(subscribedURL, listener));
return result;
}

12
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
View File

@ -169,7 +169,7 @@ public class ServiceInstancesChangedListener {
public synchronized void addListenerAndNotify(String serviceKey, NotifyListener listener) {
this.listeners.put(serviceKey, listener);
List<URL> urls = getAddresses(serviceKey, listener.getConsumerUrl());
List<URL> urls = getAddresses(serviceKey);
if (CollectionUtils.isNotEmpty(urls)) {
listener.notify(urls);
}
@ -253,14 +253,12 @@ public class ServiceInstancesChangedListener {
if (revisionToMetadata == null) {
return false;
}
boolean result = false;
for (Map.Entry<String, MetadataInfo> entry : revisionToMetadata.entrySet()) {
if (entry.getValue() == MetadataInfo.EMPTY) {
result = true;
break;
return true;
}
}
return result;
return false;
}
protected MetadataInfo getRemoteMetadata(ServiceInstance instance, String revision, Map<ServiceInfo, Set<String>> localServiceToRevisions, List<ServiceInstance> subInstances) {
@ -351,14 +349,14 @@ public class ServiceInstancesChangedListener {
return urls;
}
protected List<URL> getAddresses(String serviceProtocolKey, URL consumerURL) {
protected List<URL> getAddresses(String serviceProtocolKey) {
return (List<URL>) serviceUrls.get(serviceProtocolKey);
}
protected void notifyAddressChanged() {
listeners.forEach((key, notifyListener) -> {
//FIXME, group wildcard match
List<URL> urls = toUrlsWithEmpty(getAddresses(key, notifyListener.getConsumerUrl()));
List<URL> urls = toUrlsWithEmpty(getAddresses(key));
logger.info("Notify service " + key + " with urls " + urls.size());
notifyListener.notify(urls);
});

81
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceNameMapping.java
View File

@ -17,59 +17,72 @@
package org.apache.dubbo.registry.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.config.configcenter.ConfigItem;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.MappingListener;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.ServiceNameMapping;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.report.MetadataReport;
import org.apache.dubbo.metadata.report.MetadataReportInstance;
import org.apache.dubbo.registry.client.RegistryClusterIdentifier;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import static java.util.Arrays.asList;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
import static org.apache.dubbo.rpc.model.ApplicationModel.getName;
public class MetadataServiceNameMapping implements ServiceNameMapping {
private static final List<String> IGNORED_SERVICE_INTERFACES = asList(MetadataService.class.getName());
private final Logger logger = LoggerFactory.getLogger(getClass());
private static final List<String> IGNORED_SERVICE_INTERFACES = Collections.singletonList(MetadataService.class.getName());
private static final int CAS_RETRY_TIMES = 6;
@Override
public void map(URL url) {
String serviceInterface = url.getServiceInterface();
String group = url.getGroup();
String version = url.getVersion();
String protocol = url.getProtocol();
execute(() -> {
String serviceInterface = url.getServiceInterface();
if (IGNORED_SERVICE_INTERFACES.contains(serviceInterface)) {
return;
}
String registryCluster = getRegistryCluster(url);
MetadataReport metadataReport = MetadataReportInstance.getMetadataReport(registryCluster);
if (IGNORED_SERVICE_INTERFACES.contains(serviceInterface)) {
return;
}
String registryCluster = getRegistryCluster(url);
MetadataReport metadataReport = MetadataReportInstance.getMetadataReport(registryCluster);
metadataReport.registerServiceAppMapping(ServiceNameMapping.buildGroup(serviceInterface, group, version, protocol), getName(), url);
int currentRetryTimes = 1;
boolean success;
String newConfigContent = getName();
do {
ConfigItem configItem = metadataReport.getConfigItem(serviceInterface, DEFAULT_MAPPING_GROUP);
String oldConfigContent = configItem.getContent();
if (StringUtils.isNotEmpty(oldConfigContent)) {
boolean contains = StringUtils.isContains(oldConfigContent, getName());
if (contains) {
break;
}
newConfigContent = oldConfigContent + COMMA_SEPARATOR + getName();
}
success = metadataReport.registerServiceAppMapping(serviceInterface, DEFAULT_MAPPING_GROUP, newConfigContent, configItem.getStat());
} while (!success && currentRetryTimes++ <= CAS_RETRY_TIMES);
});
}
@Override
public Set<String> getAndListen(URL url, MappingListener mappingListener) {
String serviceInterface = url.getServiceInterface();
String group = url.getGroup();
String version = url.getVersion();
String protocol = url.getProtocol();
String mappingKey = ServiceNameMapping.buildGroup(serviceInterface, group, version, protocol);
Set<String> serviceNames = new LinkedHashSet<>();
String registryCluster = getRegistryCluster(url);
MetadataReport metadataReport = MetadataReportInstance.getMetadataReport(registryCluster);
Set<String> apps = metadataReport.getServiceAppMapping(
mappingKey,
mappingListener,
url);
if (CollectionUtils.isNotEmpty(apps)) {
execute(() -> {
String serviceInterface = url.getServiceInterface();
String registryCluster = getRegistryCluster(url);
MetadataReport metadataReport = MetadataReportInstance.getMetadataReport(registryCluster);
Set<String> apps = metadataReport.getServiceAppMapping(serviceInterface, mappingListener, url);
serviceNames.addAll(apps);
}
});
return serviceNames;
}
@ -84,4 +97,14 @@ public class MetadataServiceNameMapping implements ServiceNameMapping {
}
return registryCluster;
}
private void execute(Runnable runnable) {
try {
runnable.run();
} catch (Throwable e) {
if (logger.isWarnEnabled()) {
logger.warn(e.getMessage(), e);
}
}
}
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java
View File

@ -217,7 +217,7 @@ public class ServiceInstanceMetadataUtils {
* @return if not found, return <code>null</code>
*/
public static Endpoint getEndpoint(ServiceInstance serviceInstance, String protocol) {
List<Endpoint> endpoints = ((DefaultServiceInstance)serviceInstance).getEndpoints();
List<Endpoint> endpoints = ((DefaultServiceInstance) serviceInstance).getEndpoints();
if (endpoints != null) {
for (Endpoint endpoint : endpoints) {
if (endpoint.getProtocol().equals(protocol)) {
@ -229,14 +229,14 @@ public class ServiceInstanceMetadataUtils {
}
public static void calInstanceRevision(ServiceDiscovery serviceDiscovery, ServiceInstance instance) {
String registryCluster = serviceDiscovery.getUrl().getParameter(REGISTRY_CLUSTER_KEY);
String registryCluster = serviceDiscovery.getUrl() == null ? DEFAULT_KEY : serviceDiscovery.getUrl().getParameter(REGISTRY_CLUSTER_KEY);
if (registryCluster == null) {
registryCluster = DEFAULT_KEY;
}
WritableMetadataService writableMetadataService = WritableMetadataService.getDefaultExtension();
MetadataInfo metadataInfo = writableMetadataService.getMetadataInfos().get(registryCluster);
if (metadataInfo == null) {
metadataInfo = ((InMemoryWritableMetadataService)writableMetadataService).getDefaultMetadataInfo();
metadataInfo = ((InMemoryWritableMetadataService) writableMetadataService).getDefaultMetadataInfo();
}
if (metadataInfo != null) {
String existingInstanceRevision = instance.getMetadata().get(EXPORTED_SERVICES_REVISION_PROPERTY_NAME);
@ -253,6 +253,10 @@ public class ServiceInstanceMetadataUtils {
return "true".equals(instance.getExtendParams().get(INSTANCE_REVISION_UPDATED_KEY));
}
public static void resetInstanceUpdateKey(ServiceInstance instance) {
instance.getExtendParams().remove(INSTANCE_REVISION_UPDATED_KEY);
}
public static void refreshMetadataAndInstance(ServiceInstance serviceInstance) {
RemoteMetadataServiceImpl remoteMetadataService = MetadataUtils.getRemoteMetadataService();
remoteMetadataService.publishMetadata(ApplicationModel.getName());

6
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/store/InMemoryWritableMetadataService.java
View File

@ -153,9 +153,7 @@ public class InMemoryWritableMetadataService implements WritableMetadataService
String[] clusters = getRegistryCluster(url).split(",");
for (String cluster : clusters) {
MetadataInfo metadataInfo = metadataInfos.computeIfAbsent(cluster, k -> {
return new MetadataInfo(ApplicationModel.getName());
});
MetadataInfo metadataInfo = metadataInfos.computeIfAbsent(cluster, k -> new MetadataInfo(ApplicationModel.getName()));
metadataInfo.addService(new ServiceInfo(url));
}
metadataSemaphore.release();
@ -182,7 +180,7 @@ public class InMemoryWritableMetadataService implements WritableMetadataService
return removeURL(exportedServiceURLs, url);
}
private String getRegistryCluster(URL url){
private String getRegistryCluster(URL url) {
String registryCluster = RegistryClusterIdentifier.getExtension(url).providerKey(url);
if (StringUtils.isEmpty(registryCluster)) {
registryCluster = DEFAULT_KEY;

2
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
View File

@ -239,7 +239,7 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
registry.unsubscribe(getSubscribeUrl(), this);
}
} catch (Throwable t) {
logger.warn("unexpected error when unsubscribe service " + serviceKey + "from registry" + registry.getUrl(), t);
logger.warn("unexpected error when unsubscribe service " + serviceKey + "from registry " + registry.getUrl(), t);
}
ExtensionLoader<AddressListener> addressListenerExtensionLoader = ExtensionLoader.getExtensionLoader(AddressListener.class);

20
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/InMemoryServiceDiscovery.java
View File

@ -36,12 +36,10 @@ import static java.util.Collections.emptyList;
*
* @since 2.7.5
*/
public class InMemoryServiceDiscovery implements ServiceDiscovery {
public class InMemoryServiceDiscovery extends AbstractServiceDiscovery {
private Map<String, List<ServiceInstance>> repository = new HashMap<>();
private ServiceInstance serviceInstance;
private URL registryURL;
@Override
@ -75,18 +73,12 @@ public class InMemoryServiceDiscovery implements ServiceDiscovery {
return registryURL;
}
@Override
public ServiceInstance getLocalInstance() {
return serviceInstance;
}
public String toString() {
return "InMemoryServiceDiscovery";
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
String serviceName = serviceInstance.getServiceName();
List<ServiceInstance> serviceInstances = repository.computeIfAbsent(serviceName, s -> new LinkedList<>());
if (!serviceInstances.contains(serviceInstance)) {
@ -95,24 +87,24 @@ public class InMemoryServiceDiscovery implements ServiceDiscovery {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
unregister(serviceInstance);
register(serviceInstance);
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
String serviceName = serviceInstance.getServiceName();
List<ServiceInstance> serviceInstances = repository.computeIfAbsent(serviceName, s -> new LinkedList<>());
serviceInstances.remove(serviceInstance);
}
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.registryURL = registryURL;
}
@Override
public void destroy() {
public void doDestroy() {
}
}

27
dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
View File

@ -20,8 +20,8 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.AbstractServiceDiscovery;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
@ -52,15 +52,13 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
public class KubernetesServiceDiscovery implements ServiceDiscovery {
public class KubernetesServiceDiscovery extends AbstractServiceDiscovery {
private final Logger logger = LoggerFactory.getLogger(getClass());
private KubernetesClient kubernetesClient;
private String currentHostname;
private ServiceInstance localServiceInstance;
private URL registryURL;
private String namespace;
@ -78,7 +76,7 @@ public class KubernetesServiceDiscovery implements ServiceDiscovery {
private final static ConcurrentHashMap<String, AtomicLong> SERVICE_UPDATE_TIME = new ConcurrentHashMap<>(64);
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
Config config = KubernetesConfigUtils.createKubernetesConfig(registryURL);
this.kubernetesClient = new DefaultKubernetesClient(config);
this.currentHostname = System.getenv("HOSTNAME");
@ -102,7 +100,7 @@ public class KubernetesServiceDiscovery implements ServiceDiscovery {
}
@Override
public void destroy() throws Exception {
public void doDestroy() throws Exception {
SERVICE_WATCHER.forEach((k, v) -> v.close());
SERVICE_WATCHER.clear();
@ -116,15 +114,13 @@ public class KubernetesServiceDiscovery implements ServiceDiscovery {
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
localServiceInstance = serviceInstance;
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
if (enableRegister) {
kubernetesClient
.pods()
.inNamespace(namespace)
.withName(currentHostname)
.edit(pod->
.edit(pod ->
new PodBuilder(pod)
.editOrNewMetadata()
.addToAnnotations(KUBERNETES_PROPERTIES_KEY, JSONObject.toJSONString(serviceInstance.getMetadata()))
@ -138,14 +134,12 @@ public class KubernetesServiceDiscovery implements ServiceDiscovery {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
register(serviceInstance);
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
localServiceInstance = null;
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
if (enableRegister) {
kubernetesClient
.pods()
@ -175,11 +169,6 @@ public class KubernetesServiceDiscovery implements ServiceDiscovery {
.collect(Collectors.toSet());
}
@Override
public ServiceInstance getLocalInstance() {
return localServiceInstance;
}
@Override
public List<ServiceInstance> getInstances(String serviceName) throws NullPointerException {
Endpoints endpoints =

2
dubbo-registry/dubbo-registry-kubernetes/src/test/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryTest.java
View File

@ -162,7 +162,7 @@ public class KubernetesServiceDiscoveryTest {
Thread.sleep(5000);
ArgumentCaptor<ServiceInstancesChangedEvent> eventArgumentCaptor =
ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
Mockito.verify(mockListener, Mockito.times(2)).onEvent(eventArgumentCaptor.capture());
Mockito.verify(mockListener, Mockito.times(1)).onEvent(eventArgumentCaptor.capture());
Assertions.assertEquals(1, eventArgumentCaptor.getValue().getServiceInstances().size());
serviceDiscovery.unregister(serviceInstance);

21
dubbo-registry/dubbo-registry-multicast/src/main/java/org/apache/dubbo/registry/multicast/MulticastServiceDiscovery.java
View File

@ -17,7 +17,7 @@
package org.apache.dubbo.registry.multicast;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.AbstractServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import java.util.Collections;
@ -26,32 +26,29 @@ import java.util.Set;
/**
* TODO: make multicast protocol support Service Discovery
*/
public class MulticastServiceDiscovery implements ServiceDiscovery {
public class MulticastServiceDiscovery extends AbstractServiceDiscovery {
private URL registryURL;
private ServiceInstance serviceInstance;
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.registryURL = registryURL;
}
@Override
public void destroy() throws Exception {
public void doDestroy() throws Exception {
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;
public void doUpdate(ServiceInstance serviceInstance) throws RuntimeException {
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = null;
}
@ -65,8 +62,4 @@ public class MulticastServiceDiscovery implements ServiceDiscovery {
return registryURL;
}
@Override
public ServiceInstance getLocalInstance() {
return serviceInstance;
}
}

7
dubbo-registry/dubbo-registry-multiple/src/main/java/org/apache/dubbo/registry/multiple/MultipleServiceDiscovery.java
View File

@ -40,6 +40,7 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
private URL registryURL;
private ServiceInstance serviceInstance;
private String applicationName;
private volatile boolean isDestroy;
@Override
public void initialize(URL registryURL) throws Exception {
@ -65,11 +66,17 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
@Override
public void destroy() throws Exception {
this.isDestroy = true;
for (ServiceDiscovery serviceDiscovery : serviceDiscoveries.values()) {
serviceDiscovery.destroy();
}
}
@Override
public boolean isDestroy() {
return isDestroy;
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
this.serviceInstance = serviceInstance;

91
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosNamingServiceWrapper.java Normal file
View File

@ -0,0 +1,91 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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 org.apache.dubbo.registry.nacos;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ListView;
import java.util.List;
public class NacosNamingServiceWrapper {
private static final String INNERCLASS_SYMBOL = "$";
private static final String INNERCLASS_COMPATIBLE_SYMBOL = "___";
private NamingService namingService;
public NacosNamingServiceWrapper(NamingService namingService) {
this.namingService = namingService;
}
public String getServerStatus() {
return namingService.getServerStatus();
}
public void subscribe(String serviceName, EventListener eventListener) throws NacosException {
namingService.subscribe(handleInnerSymbol(serviceName), eventListener);
}
public void subscribe(String serviceName, String group, EventListener eventListener) throws NacosException {
namingService.subscribe(handleInnerSymbol(serviceName), group, eventListener);
}
public List<Instance> getAllInstances(String serviceName, String group) throws NacosException {
return namingService.getAllInstances(handleInnerSymbol(serviceName), group);
}
public void registerInstance(String serviceName, String group, Instance instance) throws NacosException {
namingService.registerInstance(handleInnerSymbol(serviceName), group, instance);
}
public void deregisterInstance(String serviceName, String group, String ip, int port) throws NacosException {
namingService.deregisterInstance(handleInnerSymbol(serviceName), group, ip, port);
}
public void deregisterInstance(String serviceName, String group, Instance instance) throws NacosException {
namingService.deregisterInstance(handleInnerSymbol(serviceName), group, instance);
}
public ListView<String> getServicesOfServer(int pageNo, int pageSize, String parameter) throws NacosException {
return namingService.getServicesOfServer(pageNo, pageSize, parameter);
}
public List<Instance> selectInstances(String serviceName, boolean healthy) throws NacosException {
return namingService.selectInstances(handleInnerSymbol(serviceName), healthy);
}
public void shutdown() throws NacosException {
this.namingService.shutDown();
}
/**
* see https://github.com/apache/dubbo/issues/7129
* nacos service name just support `0-9a-zA-Z-._:`, grpc interface is inner interface, need compatible.
*/
private String handleInnerSymbol(String serviceName) {
if (serviceName == null) {
return null;
}
return serviceName.replace(INNERCLASS_SYMBOL, INNERCLASS_COMPATIBLE_SYMBOL);
}
}

6
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosRegistry.java
View File

@ -125,9 +125,9 @@ public class NacosRegistry extends FailbackRegistry {
private final Logger logger = LoggerFactory.getLogger(getClass());
private final NamingService namingService;
private final NacosNamingServiceWrapper namingService;
public NacosRegistry(URL url, NamingService namingService) {
public NacosRegistry(URL url, NacosNamingServiceWrapper namingService) {
super(url);
this.namingService = namingService;
}
@ -607,7 +607,7 @@ public class NacosRegistry extends FailbackRegistry {
* @param namingService {@link NamingService}
* @throws NacosException
*/
void callback(NamingService namingService) throws NacosException;
void callback(NacosNamingServiceWrapper namingService) throws NacosException;
}

27
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/NacosServiceDiscovery.java
View File

@ -27,7 +27,6 @@ import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedLi
import org.apache.dubbo.registry.nacos.util.NacosNamingServiceUtils;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ListView;
@ -54,25 +53,24 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
private String group;
private NamingService namingService;
private NacosNamingServiceWrapper namingService;
private URL registryURL;
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.namingService = createNamingService(registryURL);
this.group = getGroup(registryURL);
this.registryURL = registryURL;
}
@Override
public void destroy() {
this.namingService = null;
public void doDestroy() throws Exception {
this.namingService.shutdown();
}
@Override
public void register(ServiceInstance serviceInstance) throws RuntimeException {
super.register(serviceInstance);
public void doRegister(ServiceInstance serviceInstance) {
execute(namingService, service -> {
Instance instance = toInstance(serviceInstance);
service.registerInstance(instance.getServiceName(), group, instance);
@ -80,19 +78,14 @@ public class NacosServiceDiscovery extends AbstractServiceDiscovery {
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
// TODO: Nacos should support
if (this.serviceInstance == null) {
register(serviceInstance);
} else {
unregister(serviceInstance);
register(serviceInstance);
this.serviceInstance = serviceInstance;
}
public void doUpdate(ServiceInstance serviceInstance) {
ServiceInstance oldInstance = this.serviceInstance;
unregister(oldInstance);
register(serviceInstance);
}
@Override
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
execute(namingService, service -> {
Instance instance = toInstance(serviceInstance);
service.deregisterInstance(instance.getServiceName(), group, instance);

8
dubbo-registry/dubbo-registry-nacos/src/main/java/org/apache/dubbo/registry/nacos/util/NacosNamingServiceUtils.java
View File

@ -22,6 +22,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.nacos.NacosNamingServiceWrapper;
import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.PropertyKeyConst;
@ -36,6 +37,7 @@ import java.util.Properties;
import static com.alibaba.nacos.api.PropertyKeyConst.NAMING_LOAD_CACHE_AT_START;
import static com.alibaba.nacos.api.PropertyKeyConst.SERVER_ADDR;
import static com.alibaba.nacos.api.common.Constants.DEFAULT_GROUP;
import static com.alibaba.nacos.client.naming.utils.UtilAndComs.NACOS_NAMING_LOG_NAME;
import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
import static org.apache.dubbo.common.utils.StringConstantFieldValuePredicate.of;
@ -99,7 +101,7 @@ public class NacosNamingServiceUtils {
* @return {@link NamingService}
* @since 2.7.5
*/
public static NamingService createNamingService(URL connectionURL) {
public static NacosNamingServiceWrapper createNamingService(URL connectionURL) {
Properties nacosProperties = buildNacosProperties(connectionURL);
NamingService namingService;
try {
@ -110,7 +112,7 @@ public class NacosNamingServiceUtils {
}
throw new IllegalStateException(e);
}
return namingService;
return new NacosNamingServiceWrapper(namingService);
}
private static Properties buildNacosProperties(URL url) {
@ -137,6 +139,8 @@ public class NacosNamingServiceUtils {
}
private static void setProperties(URL url, Properties properties) {
putPropertyIfAbsent(url, properties, NACOS_NAMING_LOG_NAME);
// @since 2.7.8 : Refactoring
// Get the parameters from constants
Map<String, String> parameters = url.getParameters(of(PropertyKeyConst.class));

1
dubbo-registry/dubbo-registry-xds/pom.xml
View File

@ -95,6 +95,7 @@
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.6.1</version>
<configuration>
<protocArtifact>com.google.protobuf:protoc:3.5.1-1:exe:${os.detected.classifier}</protocArtifact>
<pluginId>grpc-java</pluginId>

37
dubbo-registry/dubbo-registry-zookeeper/src/main/java/org/apache/dubbo/registry/zookeeper/ZookeeperServiceDiscovery.java
View File

@ -44,7 +44,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import static org.apache.dubbo.common.function.ThrowableFunction.execute;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.isInstanceUpdated;
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkParams.ROOT_PATH;
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkUtils.build;
import static org.apache.dubbo.registry.zookeeper.util.CuratorFrameworkUtils.buildCuratorFramework;
@ -73,7 +72,7 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
private final Map<String, ZookeeperServiceDiscoveryChangeWatcher> watcherCaches = new ConcurrentHashMap<>();
@Override
public void initialize(URL registryURL) throws Exception {
public void doInitialize(URL registryURL) throws Exception {
this.registryURL = registryURL;
this.curatorFramework = buildCuratorFramework(registryURL);
this.rootPath = ROOT_PATH.getParameterValue(registryURL);
@ -86,12 +85,12 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
return registryURL;
}
public void destroy() throws Exception {
public void doDestroy() throws Exception {
serviceDiscovery.close();
}
public void register(ServiceInstance serviceInstance) throws RuntimeException {
super.register(serviceInstance);
@Override
public void doRegister(ServiceInstance serviceInstance) {
try {
serviceDiscovery.registerService(build(serviceInstance));
} catch (Exception e) {
@ -99,19 +98,18 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
}
}
public void update(ServiceInstance serviceInstance) throws RuntimeException {
if (this.serviceInstance == null) {
this.register(serviceInstance);
} else if (isInstanceUpdated(serviceInstance)) {
this.unregister(this.serviceInstance);
this.register(serviceInstance);
this.serviceInstance = serviceInstance;
}
@Override
public void doUpdate(ServiceInstance serviceInstance) {
ServiceInstance oldInstance = this.serviceInstance;
this.unregister(oldInstance);
this.register(serviceInstance);
}
public void unregister(ServiceInstance serviceInstance) throws RuntimeException {
@Override
public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
doInServiceRegistry(serviceDiscovery -> serviceDiscovery.unregisterService(build(serviceInstance)));
}
};
@Override
public Set<String> getServices() {
@ -180,11 +178,14 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
@Override
public void removeServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws IllegalArgumentException {
listener.getServiceNames().forEach(serviceName -> {
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.remove(serviceName);
watcher.stopWatching();
ZookeeperServiceDiscoveryChangeWatcher watcher = watcherCaches.remove(buildServicePath(serviceName));
if (watcher != null) {
watcher.stopWatching();
}
});
}
private void doInServiceRegistry(ThrowableConsumer<org.apache.curator.x.discovery.ServiceDiscovery> consumer) {
ThrowableConsumer.execute(serviceDiscovery, s -> consumer.accept(s));
}
@ -194,7 +195,7 @@ public class ZookeeperServiceDiscovery extends AbstractServiceDiscovery {
}
protected void registerServiceWatcher(String serviceName, ServiceInstancesChangedListener listener) {
String path = buildServicePath(serviceName);
String path = buildServicePath(serviceName);
try {
curatorFramework.create().creatingParentsIfNeeded().forPath(path);
} catch (KeeperException.NodeExistsException e) {

2
dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/ZookeeperClient.java
View File

@ -61,7 +61,7 @@ public interface ZookeeperClient {
void create(String path, String content, boolean ephemeral);
void createOrUpdate(String path, String content, boolean ephemeral, Object stat);
void createOrUpdate(String path, String content, boolean ephemeral, int ticket);
String getContent(String path);

27
dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/curator/CuratorZookeeperClient.java
View File

@ -58,7 +58,7 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
protected static final Logger logger = LoggerFactory.getLogger(CuratorZookeeperClient.class);
private static final String ZK_SESSION_EXPIRE_KEY = "zk.session.expire";
static final Charset CHARSET = StandardCharsets.UTF_8;
private static final Charset CHARSET = StandardCharsets.UTF_8;
private final CuratorFramework client;
private static Map<String, NodeCache> nodeCacheMap = new ConcurrentHashMap<>();
@ -149,25 +149,20 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
}
@Override
protected void update(String path, String data, Object stat) {
protected void update(String path, String data, int version) {
byte[] dataBytes = data.getBytes(CHARSET);
try {
if (null == stat || !(stat instanceof Stat)) {
throw new IllegalArgumentException("unable to get the version information of zookeeper data");
}
client.setData().withVersion(((Stat) stat).getVersion()).forPath(path, dataBytes);
} catch (NoNodeException e) {
logger.warn("ZNode " + path + "does not exists.", e);
client.setData().withVersion(version).forPath(path, dataBytes);
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
@Override
protected void createOrUpdatePersistent(String path, String data, Object stat) {
protected void createOrUpdatePersistent(String path, String data, int version) {
try {
if (checkExists(path)) {
update(path, data, stat);
update(path, data, version);
} else {
createPersistent(path, data);
}
@ -177,10 +172,10 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
}
@Override
protected void createOrUpdateEphemeral(String path, String data, Object stat) {
protected void createOrUpdateEphemeral(String path, String data, int version) {
try {
if (checkExists(path)) {
update(path, data, stat);
update(path, data, version);
} else {
createEphemeral(path, data);
}
@ -216,7 +211,7 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
if (client.checkExists().forPath(path) != null) {
return true;
}
} catch (Exception e) {
} catch (Exception ignored) {
}
return false;
}
@ -241,14 +236,14 @@ public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorZooke
@Override
public ConfigItem doGetConfigItem(String path) {
String content = null;
Stat stat = null;
String content;
Stat stat;
try {
stat = new Stat();
byte[] dataBytes = client.getData().storingStatIn(stat).forPath(path);
content = (dataBytes == null || dataBytes.length == 0) ? null : new String(dataBytes, CHARSET);
} catch (NoNodeException e) {
// ignore NoNode Exception.
return new ConfigItem();
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}

15
dubbo-remoting/dubbo-remoting-zookeeper/src/main/java/org/apache/dubbo/remoting/zookeeper/support/AbstractZookeeperClient.java
View File

@ -185,15 +185,15 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
}
@Override
public void createOrUpdate(String path, String content, boolean ephemeral, Object stat) {
public void createOrUpdate(String path, String content, boolean ephemeral, int version) {
int i = path.lastIndexOf('/');
if (i > 0) {
create(path.substring(0, i), false);
}
if (ephemeral) {
createOrUpdateEphemeral(path, content, stat);
createOrUpdateEphemeral(path, content, version);
} else {
createOrUpdatePersistent(path, content, stat);
createOrUpdatePersistent(path, content, version);
}
}
@ -207,9 +207,6 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
@Override
public ConfigItem getConfigItem(String path) {
if (!checkExists(path)) {
return new ConfigItem();
}
return doGetConfigItem(path);
}
@ -223,11 +220,11 @@ public abstract class AbstractZookeeperClient<TargetDataListener, TargetChildLis
protected abstract void createEphemeral(String path, String data);
protected abstract void update(String path, String data, Object stat);
protected abstract void update(String path, String data, int version);
protected abstract void createOrUpdatePersistent(String path, String data, Object stat);
protected abstract void createOrUpdatePersistent(String path, String data, int version);
protected abstract void createOrUpdateEphemeral(String path, String data, Object stat);
protected abstract void createOrUpdateEphemeral(String path, String data, int version);
@Override
public abstract boolean checkExists(String path);