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

Merge remote-tracking branch 'origin/cloud-native' into cloud-native 

# Conflicts:
#	dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceConsumerBootstrap.java
#	dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
#	dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java
This commit is contained in:
ken.lj 2019-08-26 21:15:16 +08:00
parent 5817ac705f 27ba1c7014
commit 0b0b83f962
18 changed files with 653 additions and 342 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderMinimumBootstrap.java
View File

@ -16,6 +16,9 @@
*/
package org.apache.dubbo.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.bootstrap.rest.UserServiceImpl;
/**
* TODO
*/
@ -24,10 +27,11 @@ public class DubboServiceProviderMinimumBootstrap {
public static void main(String[] args) {
new DubboBootstrap()
.application("dubbo-provider-demo")
// .registry(builder -> builder.address("zookeeper://127.0.0.1:2181?registry.type=service"))
.registry(builder -> builder.address("eureka://127.0.0.1:8761?registry-type=service"))
.registry(builder -> builder.address("zookeeper://127.0.0.1:2181?registry-type=service"))
// .registry(builder -> builder.address("eureka://127.0.0.1:8761?registry-type=service"))
.protocol(builder -> builder.port(-1).name("dubbo"))
.service(builder -> builder.interfaceClass(EchoService.class).ref(new EchoServiceImpl()))
.service("echo", builder -> builder.interfaceClass(EchoService.class).ref(new EchoServiceImpl()))
.service("user", builder -> builder.interfaceClass(UserService.class).ref(new UserServiceImpl()))
.start()
.await();
}

4
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java
View File

@ -32,11 +32,11 @@ public class NacosDubboServiceConsumerBootstrap {
public static void main(String[] args) throws Exception {
ApplicationConfig applicationConfig = new ApplicationConfig("dubbo-nacos-consumer-demo");
applicationConfig.setMetadataType("remote");
// applicationConfig.setMetadataType("remote");
new DubboBootstrap()
.application(applicationConfig)
// Zookeeper
.registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service&subscribed.services=dubbo-nacos-provider-demo"))
.registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry-type=service&subscribed-services=dubbo-nacos-provider-demo"))
.metadataReport(new MetadataReportConfig("nacos://127.0.0.1:8848"))
// Nacos
// .registry("consul", builder -> builder.address("consul://127.0.0.1:8500?registry.type=service&subscribed.services=dubbo-provider-demo").group("namespace1"))

4
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java
View File

@ -30,11 +30,11 @@ public class NacosDubboServiceProviderBootstrap {
public static void main(String[] args) {
ApplicationConfig applicationConfig = new ApplicationConfig("dubbo-nacos-provider-demo");
applicationConfig.setMetadataType("remote");
// applicationConfig.setMetadataType("remote");
new DubboBootstrap()
.application(applicationConfig)
// Zookeeper in service registry type
.registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service"))
.registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry-type=service"))
// Nacos
// .registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service"))
// .registry(RegistryBuilder.newBuilder().address("etcd3://127.0.0.1:2379?registry.type=service").build())

58
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceConsumerBootstrap.java Normal file
View File

@ -0,0 +1,58 @@
/*
* 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.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.context.ConfigManager;
/**
* Dubbo Provider Bootstrap
*
* @since 2.7.4
*/
public class ZookeeperDubboServiceConsumerBootstrap {
public static void main(String[] args) throws Exception {
new DubboBootstrap()
.application("zookeeper-dubbo-consumer")
.registry("zookeeper", builder -> builder.address("zookeeper://127.0.0.1:2181?registry-type=service&subscribed-services=zookeeper-dubbo-provider"))
.reference("echo", builder -> builder.interfaceClass(EchoService.class).protocol("dubbo"))
.reference("user", builder -> builder.interfaceClass(UserService.class).protocol("rest"))
.start()
.await();
ConfigManager configManager = ConfigManager.getInstance();
ReferenceConfig<EchoService> referenceConfig = configManager.getReference("echo");
EchoService echoService = referenceConfig.get();
ReferenceConfig<UserService> referenceConfig2 = configManager.getReference("user");
UserService userService = referenceConfig2.get();
for (int i = 0; i < 500; i++) {
Thread.sleep(2000L);
System.out.println(echoService.echo("Hello,World"));
System.out.println(userService.getUser(i * 1L));
}
}
}

38
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceProviderBootstrap.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.bootstrap.rest.UserServiceImpl;
/**
* TODO
*/
public class ZookeeperDubboServiceProviderBootstrap {
public static void main(String[] args) {
new DubboBootstrap()
.application("zookeeper-dubbo-provider")
.registry(builder -> builder.address("zookeeper://127.0.0.1:2181?registry-type=service"))
.protocol("dubbo", builder -> builder.port(-1).name("dubbo"))
.protocol("rest", builder -> builder.port(8082).name("rest"))
.service("echo", builder -> builder.interfaceClass(EchoService.class).ref(new EchoServiceImpl()).protocolIds("dubbo"))
.service("user", builder -> builder.interfaceClass(UserService.class).ref(new UserServiceImpl()).protocolIds("rest"))
.start()
.await();
}
}

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

@ -26,26 +26,40 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.api.NacosFactory;
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.client.config.impl.HttpSimpleClient;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.List;
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.ACCESS_KEY;
import static com.alibaba.nacos.api.PropertyKeyConst.CLUSTER_NAME;
import static com.alibaba.nacos.api.PropertyKeyConst.ENCODE;
import static com.alibaba.nacos.api.PropertyKeyConst.ENDPOINT;
import static com.alibaba.nacos.api.PropertyKeyConst.NAMESPACE;
import static com.alibaba.nacos.api.PropertyKeyConst.SECRET_KEY;
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.Arrays.asList;
import static java.util.Collections.emptyList;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPERATOR;
import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
@ -54,17 +68,22 @@ import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
*/
public class NacosDynamicConfiguration implements DynamicConfiguration {
private static final String GET_CONFIG_KEYS_PATH = "/v1/cs/configs";
private final Logger logger = LoggerFactory.getLogger(getClass());
/**
* the default timeout in millis to get config from nacos
*/
private static final long DEFAULT_TIMEOUT = 5000L;
private Properties nacosProperties;
/**
* The nacos configService
*/
private final ConfigService configService;
private ConfigService configService;
private HttpAgent httpAgent;
/**
* The map store the key to {@link NacosConfigListener} mapping
@ -72,12 +91,14 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
private final ConcurrentMap<String, NacosConfigListener> watchListenerMap;
NacosDynamicConfiguration(URL url) {
buildConfigService(url);
this.nacosProperties = buildNacosProperties(url);
this.configService = buildConfigService(url);
this.httpAgent = getHttpAgent(configService);
watchListenerMap = new ConcurrentHashMap<>();
}
private ConfigService buildConfigService(URL url) {
Properties nacosProperties = buildNacosProperties(url);
ConfigService configService = null;
try {
configService = NacosFactory.createConfigService(nacosProperties);
} catch (NacosException e) {
@ -89,6 +110,18 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
return configService;
}
private HttpAgent getHttpAgent(ConfigService configService) {
HttpAgent agent = null;
try {
Field field = configService.getClass().getDeclaredField("agent");
field.setAccessible(true);
agent = (HttpAgent) field.get(configService);
} catch (Exception e) {
throw new IllegalStateException(e);
}
return agent;
}
public void publishNacosConfig(String key, String value) {
String[] keyAndGroup = getKeyAndGroup(key);
publishConfig(keyAndGroup[0], keyAndGroup[1], value);
@ -143,6 +176,7 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
putPropertyIfAbsent(url, properties, ACCESS_KEY);
putPropertyIfAbsent(url, properties, SECRET_KEY);
putPropertyIfAbsent(url, properties, CLUSTER_NAME);
putPropertyIfAbsent(url, properties, ENCODE);
}
private void putPropertyIfAbsent(URL url, Properties properties, String propertyName) {
@ -208,6 +242,36 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
return null;
}
@Override
public SortedSet<String> getConfigKeys(String group) {
// TODO use Nacos Client API to replace HTTP Open API
SortedSet<String> keys = new TreeSet<>();
try {
List<String> paramsValues = asList("search", "accurate", "dataId", "", "group", group, "pageNo", "1", "pageSize", String.valueOf(Integer.MAX_VALUE));
String encoding = getProperty(ENCODE, "UTF-8");
HttpSimpleClient.HttpResult result = httpAgent.httpGet(GET_CONFIG_KEYS_PATH, emptyList(), paramsValues, encoding, 5 * 1000);
Stream<String> keysStream = toKeysStream(result.content);
keysStream.forEach(keys::add);
} catch (IOException e) {
if (logger.isErrorEnabled()) {
logger.error(e.getMessage(), e);
}
}
return keys;
}
private Stream<String> toKeysStream(String content) {
JSONObject jsonObject = JSON.parseObject(content);
JSONArray pageItems = jsonObject.getJSONArray("pageItems");
return pageItems.stream()
.map(object -> (JSONObject) object)
.map(json -> json.getString("dataId"));
}
private String getProperty(String name, String defaultValue) {
return nacosProperties.getProperty(name, defaultValue);
}
public class NacosConfigListener extends AbstractSharedListener {
private Set<ConfigurationListener> listeners = new CopyOnWriteArraySet<>();
@ -259,5 +323,4 @@ public class NacosDynamicConfiguration implements DynamicConfiguration {
return ConfigChangeType.MODIFIED;
}
}
}

13
dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java
View File

@ -33,6 +33,7 @@ import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
import java.util.SortedSet;
import java.util.concurrent.CountDownLatch;
@ -99,6 +100,18 @@ public class NacosDynamicConfigurationTest {
}
@Test
public void testGetConfigKeys() {
put("key1", "a");
put("key2", "b");
SortedSet<String> keys = config.getConfigKeys(DynamicConfiguration.DEFAULT_GROUP);
Assertions.assertFalse(keys.isEmpty());
}
private void put(String key, String value) {
put(key, DynamicConfiguration.DEFAULT_GROUP, value);
}

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

@ -23,6 +23,7 @@ import org.apache.dubbo.common.utils.Page;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.event.EventDispatcher;
import org.apache.dubbo.event.EventListener;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
@ -198,16 +199,22 @@ public interface ServiceDiscovery extends Prioritized {
/**
* Add an instance of {@link ServiceInstancesChangedListener} for specified service
* <p>
* Default, the ServiceInstancesChangedListener will be {@link EventDispatcher#addEventListener(EventListener) added}
* into {@link EventDispatcher}
* Default, Current method will be invoked by {@link ServiceDiscoveryRegistry#subscribe(URL, NotifyListener)
* the ServiceDiscoveryRegistry on the subscription}, and it's mandatory to
* {@link EventDispatcher#addEventListener(EventListener) add} the {@link ServiceInstancesChangedListener} argument
* into {@link EventDispatcher} whether the subclass implements same approach or not, thus this method is used to
* trigger or adapt the vendor's change notification mechanism typically, like Zookeeper Watcher,
* Nacos EventListener. If the registry observes the change, It's suggested that the implementation could invoke
* {@link #dispatchServiceInstancesChangedEvent(String)} method or variants
*
* @param listener an instance of {@link ServiceInstancesChangedListener}
* @throws NullPointerException
* @throws IllegalArgumentException
* @see EventPublishingServiceDiscovery
* @see EventDispatcher
*/
default void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
throws NullPointerException, IllegalArgumentException {
getDefaultExtension().addEventListener(listener);
}
/**

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

@ -37,14 +37,14 @@ import org.apache.dubbo.registry.support.FailbackRegistry;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.stream.Collectors;
@ -52,15 +52,17 @@ import java.util.stream.Collectors;
import static java.lang.String.format;
import static java.util.Collections.emptyList;
import static java.util.stream.Stream.of;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
import static org.apache.dubbo.common.URLBuilder.from;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_PROTOCOL;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPERATOR;
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.PID_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.SIDE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_TYPE_KEY;
import static org.apache.dubbo.common.constants.RegistryConstants.SERVICE_REGISTRY_TYPE;
@ -75,10 +77,8 @@ import static org.apache.dubbo.common.utils.StringUtils.isBlank;
import static org.apache.dubbo.metadata.WritableMetadataService.DEFAULT_EXTENSION;
import static org.apache.dubbo.registry.client.ServiceDiscoveryFactory.getExtension;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getExportedServicesRevision;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getMetadataServiceURLsParams;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getMetadataStorageType;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProviderHost;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProviderPort;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProtocolPort;
/**
* {@link ServiceDiscoveryRegistry} is the service-oriented {@link Registry} and dislike the traditional one that
@ -109,7 +109,14 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
private final WritableMetadataService writableMetadataService;
private final Set<String> listenedServices = new LinkedHashSet<>();
private final Set<String> registeredListeners = new LinkedHashSet<>();
/**
* A cache for all URLs of services that the subscribed services exported
* The key is the service name
* The value is a nested {@link Map} whose key is the revision and value is all URLs of services
*/
private final Map<String, Map<String, List<URL>>> serviceExportedURLsCache = new LinkedHashMap<>();
public ServiceDiscoveryRegistry(URL registryURL) {
super(registryURL);
@ -303,7 +310,7 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
subscribeURLs(url, listener, serviceName, serviceInstances);
// register ServiceInstancesChangedListener
registerServiceInstancesChangedListener(new ServiceInstancesChangedListener(serviceName, subscribedServices.get(serviceName)) {
registerServiceInstancesChangedListener(url, new ServiceInstancesChangedListener(serviceName, subscribedServices.get(serviceName)) {
@Override
public void onEvent(ServiceInstancesChangedEvent event) {
@ -315,14 +322,20 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
/**
* Register the {@link ServiceInstancesChangedListener} If absent
*
* @param url {@link URL}
* @param listener the {@link ServiceInstancesChangedListener}
*/
private void registerServiceInstancesChangedListener(ServiceInstancesChangedListener listener) {
if (listenedServices.add(listener.getServiceName())) {
private void registerServiceInstancesChangedListener(URL url, ServiceInstancesChangedListener listener) {
String listenerId = createListenerId(url, listener);
if (registeredListeners.add(listenerId)) {
serviceDiscovery.addServiceInstancesChangedListener(listener);
}
}
private String createListenerId(URL url, ServiceInstancesChangedListener listener) {
return listener.getServiceName() + ":" + url.toString(VERSION_KEY, GROUP_KEY, PROTOCOL_KEY);
}
protected void subscribeURLs(URL subscribedURL, NotifyListener listener, String serviceName,
Collection<ServiceInstance> serviceInstances) {
@ -338,100 +351,107 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
private List<URL> getSubscribedURLs(URL subscribedURL, Collection<ServiceInstance> instances, String serviceName) {
List<URL> subscribedURLs = new LinkedList<>();
// local service instances could be mutable
List<ServiceInstance> serviceInstances = instances.stream()
.filter(ServiceInstance::isEnabled)
.filter(ServiceInstance::isHealthy)
.collect(Collectors.toList());
/**
* A caches all revisions of exported services in different {@link ServiceInstance}s
* associating with the {@link URL urls}
*/
Map<String, List<URL>> revisionURLsCache = new HashMap<>();
int size = serviceInstances.size();
if (ServiceInstanceMetadataUtils.isDubboServiceInstance(serviceInstances.get(0))) {
// try to get the exported URLs from every instance until it's successful.
for (int i = 0; i < serviceInstances.size(); i++) {
// select a instance of {@link ServiceInstance}
ServiceInstance selectedInstance = selectServiceInstance(serviceInstances);
List<URL> templateURLs = getTemplateURLs(subscribedURL, selectedInstance, revisionURLsCache);
if (isNotEmpty(templateURLs)) {
// add templateURLs into subscribedURLs
subscribedURLs.addAll(templateURLs);
// remove the selected ServiceInstance in this time, it remains N - 1 elements.
serviceInstances.remove(selectedInstance);
break;
}
}
// Clone the subscribed URLs from the template URLs
List<URL> clonedURLs = cloneSubscribedURLs(subscribedURL, serviceInstances, revisionURLsCache);
// Add all cloned URLs into subscribedURLs
subscribedURLs.addAll(clonedURLs);
// clear all revisions
revisionURLsCache.clear();
} else {
for (ServiceInstance instance : serviceInstances) {
URLBuilder builder = new URLBuilder(
subscribedServices.get(serviceName),
instance.getHost(),
instance.getPort(),
subscribedURL.getServiceInterface(),
instance.getMetadata()
);
builder.addParameter(APPLICATION_KEY, serviceName);
subscribedURLs.add(builder.build());
}
if (size == 0) {
return emptyList();
}
expungeStaleRevisionExportedURLs(serviceInstances);
initTemplateURLs(subscribedURL, serviceInstances);
// Clone the subscribed URLs from the template URLs
List<URL> subscribedURLs = cloneSubscribedURLs(subscribedURL, serviceInstances);
// clear local service instances
serviceInstances.clear();
return subscribedURLs;
}
private List<URL> cloneSubscribedURLs(URL subscribedURL, Collection<ServiceInstance> serviceInstances,
Map<String, List<URL>> revisionURLsCache) {
// If revisionURLsCache is not empty, clone the template URLs to be the subscribed URLs
if (!revisionURLsCache.isEmpty()) {
List<URL> clonedURLs = new LinkedList<>();
Iterator<ServiceInstance> iterator = serviceInstances.iterator();
while (iterator.hasNext()) {
ServiceInstance serviceInstance = iterator.next();
List<URL> templateURLs = getTemplateURLs(subscribedURL, serviceInstance, revisionURLsCache);
// The parameters of URLs that the MetadataService exported
Map<String, Map<String, Object>> serviceURLsParams = getMetadataServiceURLsParams(serviceInstance);
templateURLs.forEach(templateURL -> {
String protocol = templateURL.getProtocol();
Map<String, Object> serviceURLParams = serviceURLsParams.get(protocol);
String host = getProviderHost(serviceURLParams);
Integer port = getProviderPort(serviceURLParams);
/**
* clone the subscribed {@link URL urls} based on the template {@link URL url}
*/
URL newSubscribedURL = new URL(protocol, host, port, templateURL.getParameters());
clonedURLs.add(newSubscribedURL);
});
private void initTemplateURLs(URL subscribedURL, List<ServiceInstance> serviceInstances) {
// Try to get the template URLs until success
for (int i = 0; i < serviceInstances.size(); i++) {
// select a instance of {@link ServiceInstance}
ServiceInstance selectedInstance = selectServiceInstance(serviceInstances);
// try to get the template URLs
List<URL> templateURLs = getTemplateURLs(subscribedURL, selectedInstance);
if (isNotEmpty(templateURLs)) { // If the result is valid
break; // break the loop
} else {
serviceInstances.remove(selectedInstance); // remove if the service instance is not available
// There may be one or more service instances from the "serviceInstances"
}
}
}
return clonedURLs;
private void expungeStaleRevisionExportedURLs(List<ServiceInstance> serviceInstances) {
if (isEmpty(serviceInstances)) { // if empty, return immediately
return;
}
return Collections.emptyList();
String serviceName = serviceInstances.get(0).getServiceName();
synchronized (this) {
// revisionExportedURLs is mutable
Map<String, List<URL>> revisionExportedURLs = serviceExportedURLsCache.computeIfAbsent(serviceName, s -> new HashMap<>());
if (revisionExportedURLs.isEmpty()) { // if empty, return immediately
return;
}
Set<String> existedRevisions = revisionExportedURLs.keySet(); // read-only
Set<String> currentRevisions = serviceInstances.stream()
.map(ServiceInstanceMetadataUtils::getExportedServicesRevision)
.collect(Collectors.toSet());
// staleRevisions = existedRevisions(copy) - currentRevisions
Set<String> staleRevisions = new HashSet<>(existedRevisions);
staleRevisions.removeAll(currentRevisions);
// remove exported URLs if staled
staleRevisions.forEach(revisionExportedURLs::remove);
}
}
private List<URL> cloneSubscribedURLs(URL subscribedURL, Collection<ServiceInstance> serviceInstances) {
if (isEmpty(serviceInstances)) {
return emptyList();
}
List<URL> clonedURLs = new LinkedList<>();
serviceInstances.forEach(serviceInstance -> {
String host = serviceInstance.getHost();
getTemplateURLs(subscribedURL, serviceInstance)
.stream()
.map(templateURL -> templateURL.removeParameter(TIMESTAMP_KEY))
.map(templateURL -> templateURL.removeParameter(PID_KEY))
.map(templateURL -> {
String protocol = templateURL.getProtocol();
int port = getProtocolPort(serviceInstance, protocol);
if (Objects.equals(templateURL.getHost(), host)
&& Objects.equals(templateURL.getPort(), port)) { // use templateURL if equals
return templateURL;
}
URLBuilder clonedURLBuilder = from(templateURL) // remove the parameters from the template URL
.setHost(host) // reset the host
.setPort(port); // reset the port
return clonedURLBuilder.build();
})
.forEach(clonedURLs::add);
});
return clonedURLs;
}
@ -442,74 +462,187 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
* @return <code>null</code> if <code>serviceInstances</code> is empty.
*/
private ServiceInstance selectServiceInstance(List<ServiceInstance> serviceInstances) {
int size = serviceInstances.size();
if (size == 0) {
return null;
} else if (size == 1) {
return serviceInstances.get(0);
}
ServiceInstanceSelector selector = getExtensionLoader(ServiceInstanceSelector.class).getAdaptiveExtension();
return selector.select(getUrl(), serviceInstances);
}
/**
* Get the template {@link URL urls} from the specified {@link ServiceInstance}.
* <p>
* First, put the revision {@link ServiceInstance service instance}
* associating {@link #getExportedURLs(URL, ServiceInstance) exported URLs} into cache.
* <p>
* And then compare a new {@link ServiceInstance service instances'} revision with cached one,If they are equal,
* return the cached template {@link URL urls} immediately, or to get template {@link URL urls} that the provider
* {@link ServiceInstance instance} exported via executing {@link #getExportedURLs(URL, ServiceInstance)}
* method.
* <p>
* Eventually, the retrieving result will be cached and returned.
*
* @param subscribedURL the subscribed {@link URL url}
* @param selectedInstance the {@link ServiceInstance}
* associating with the {@link URL urls}
* @return non-null {@link List} of {@link URL urls}
*/
protected List<URL> getTemplateURLs(URL subscribedURL, ServiceInstance selectedInstance) {
List<URL> exportedURLs = getRevisionExportedURLs(selectedInstance);
if (isEmpty(exportedURLs)) {
return emptyList();
}
return filterSubscribedURLs(subscribedURL, exportedURLs);
}
private List<URL> filterSubscribedURLs(URL subscribedURL, List<URL> exportedURLs) {
return exportedURLs.stream()
.filter(url -> isSameServiceInterface(subscribedURL, url))
.filter(url -> isSameParameter(subscribedURL, url, VERSION_KEY))
.filter(url -> isSameParameter(subscribedURL, url, GROUP_KEY))
.filter(url -> isCompatibleProtocol(subscribedURL, url))
.collect(Collectors.toList());
}
private boolean isSameServiceInterface(URL one, URL another) {
return Objects.equals(one.getServiceInterface(), another.getServiceInterface());
}
private boolean isSameParameter(URL one, URL another, String key) {
return Objects.equals(one.getParameter(key), another.getParameter(key));
}
private boolean isCompatibleProtocol(URL one, URL another) {
String protocol = one.getParameter(PROTOCOL_KEY);
return isCompatibleProtocol(protocol, another);
}
private boolean isCompatibleProtocol(String protocol, URL targetURL) {
return protocol == null || Objects.equals(protocol, targetURL.getParameter(PROTOCOL_KEY))
|| Objects.equals(protocol, targetURL.getProtocol());
}
/**
* Get all services {@link URL URLs} that the specified {@link ServiceInstance service instance} exported with cache
* <p>
* Typically, the revisions of all {@link ServiceInstance instances} in one service are same. However,
* if one service is upgrading one or more Dubbo service interfaces, one of them may have the multiple declarations
* is deploying in the different {@link ServiceInstance service instances}, thus, it has to compare the interface
* contract one by one, the "revision" that is the number is introduced to identify all Dubbo exported interfaces in
* one {@link ServiceInstance service instance}.
* <p>
* First, put the revision {@link ServiceInstance service instance}
* associating {@link #getProviderExportedURLs(URL, ServiceInstance) exported URLs} into cache.
* <p>
* And then compare a new {@link ServiceInstance service instances'} revision with cached one,If they are equal,
* return the cached template {@link URL urls} immediately, or to get template {@link URL urls} that the provider
* {@link ServiceInstance instance} exported via executing {@link #getProviderExportedURLs(URL, ServiceInstance)}
* method.
* <p>
* Eventually, the retrieving result will be cached and returned.
*
* @param subscribedURL the subscribed {@link URL url}
* @param selectedInstance the {@link ServiceInstance}
* @param revisionURLsCache A caches all revisions of exported services in different {@link ServiceInstance}s
* associating with the {@link URL urls}
* @return non-null {@link List} of {@link URL urls}
* @param providerServiceInstance the {@link ServiceInstance} provides the Dubbo Services
* @return the same as {@link #getExportedURLs(ServiceInstance)}
*/
protected List<URL> getTemplateURLs(URL subscribedURL, ServiceInstance selectedInstance,
Map<String, List<URL>> revisionURLsCache) {
// get the revision from the specified {@link ServiceInstance}
String revision = getExportedServicesRevision(selectedInstance);
// try to get templateURLs from cache
List<URL> templateURLs = revisionURLsCache.get(revision);
private List<URL> getRevisionExportedURLs(ServiceInstance providerServiceInstance) {
if (isEmpty(templateURLs)) { // not exists or getting failed last time
if (!revisionURLsCache.isEmpty()) { // it's not first time
if (logger.isWarnEnabled()) {
logger.warn(format("The ServiceInstance[id: %s, host : %s , port : %s] has different revision : %s" +
", please make sure the service [name : %s] is changing or not.",
selectedInstance.getId(),
selectedInstance.getHost(),
selectedInstance.getPort(),
revision,
selectedInstance.getServiceName()
));
}
}
// get or get again
templateURLs = getProviderExportedURLs(subscribedURL, selectedInstance);
// put into cache
revisionURLsCache.put(revision, templateURLs);
if (providerServiceInstance == null) {
return emptyList();
}
return templateURLs;
String serviceName = providerServiceInstance.getServiceName();
// get the revision from the specified {@link ServiceInstance}
String revision = getExportedServicesRevision(providerServiceInstance);
List<URL> exportedURLs = null;
synchronized (this) { // It's required to lock here because it may run in the sync or async mode
Map<String, List<URL>> exportedURLsMap = serviceExportedURLsCache.computeIfAbsent(serviceName, s -> new LinkedHashMap());
exportedURLs = exportedURLsMap.get(revision);
boolean firstGet = false;
if (exportedURLs == null) { // The hit is missing in cache
if (!exportedURLsMap.isEmpty()) { // The case is that current ServiceInstance with the different revision
if (logger.isWarnEnabled()) {
logger.warn(format("The ServiceInstance[id: %s, host : %s , port : %s] has different revision : %s" +
", please make sure the service [name : %s] is changing or not.",
providerServiceInstance.getId(),
providerServiceInstance.getHost(),
providerServiceInstance.getPort(),
revision,
providerServiceInstance.getServiceName()
));
}
} else {// Else, it's the first time to get the exported URLs
firstGet = true;
}
exportedURLs = getExportedURLs(providerServiceInstance);
if (exportedURLs != null) { // just allow the valid result into exportedURLsMap
exportedURLsMap.put(revision, exportedURLs);
if (logger.isDebugEnabled()) {
logger.debug(format("Getting the exported URLs[size : %s, first : %s] from the target service " +
"instance [id: %s , service : %s , host : %s , port : %s , revision : %s]",
exportedURLs.size(), firstGet,
providerServiceInstance.getId(),
providerServiceInstance.getServiceName(),
providerServiceInstance.getHost(),
providerServiceInstance.getPort(),
revision
));
}
}
}
}
// Get a copy from source in order to prevent the caller trying to change the cached data
return exportedURLs != null ? new ArrayList<>(exportedURLs) : null;
}
/**
* Get all services {@link URL URLs} that the specified {@link ServiceInstance service instance} exported
* from {@link MetadataService} proxy
*
* @param providerServiceInstance the {@link ServiceInstance} provides the Dubbo Services
* @return The possible result :
* <ol>
* <li>The normal result</li>
* <li>The empty result if the {@link ServiceInstance service instance} did not export yet</li>
* <li><code>null</code> if there is an invocation error on {@link MetadataService} proxy</li>
* </ol>
*/
private List<URL> getExportedURLs(ServiceInstance providerServiceInstance) {
List<URL> exportedURLs = null;
String metadataStorageType = getMetadataStorageType(providerServiceInstance);
try {
MetadataService metadataService = MetadataServiceProxyFactory
.getExtension(metadataStorageType == null ? DEFAULT_EXTENSION : metadataStorageType)
.getProxy(providerServiceInstance);
SortedSet<String> urls = metadataService.getExportedURLs();
exportedURLs = urls.stream().map(URL::valueOf).collect(Collectors.toList());
} catch (Throwable e) {
if (logger.isErrorEnabled()) {
logger.error(format("It's failed to get the exported URLs from the target service instance[%s]",
providerServiceInstance), e);
}
exportedURLs = null; // set the result to be null if failed to get
}
return exportedURLs;
}
/**
* Get the exported {@link URL urls} from the specified provider {@link ServiceInstance instance}
*
* @param subscribedURL the subscribed {@link URL url}
* @param providerInstance the target provider {@link ServiceInstance instance}
* @param subscribedURL the subscribed {@link URL url}
* @param providerServiceInstance the target provider {@link ServiceInstance instance}
* @return non-null {@link List} of {@link URL urls}
*/
protected List<URL> getProviderExportedURLs(URL subscribedURL, ServiceInstance providerInstance) {
protected List<URL> getExportedURLs(URL subscribedURL, ServiceInstance providerServiceInstance) {
List<URL> exportedURLs = emptyList();
@ -518,12 +651,12 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
String version = subscribedURL.getParameter(VERSION_KEY);
// The subscribed protocol may be null
String protocol = subscribedURL.getParameter(PROTOCOL_KEY);
String metadataStorageType = getMetadataStorageType(providerInstance);
String metadataStorageType = getMetadataStorageType(providerServiceInstance);
try {
MetadataService metadataService = MetadataServiceProxyFactory
.getExtension(metadataStorageType == null ? DEFAULT_EXTENSION : metadataStorageType)
.getProxy(providerInstance);
.getProxy(providerServiceInstance);
SortedSet<String> urls = metadataService.getExportedURLs(serviceInterface, group, version, protocol);
exportedURLs = urls.stream().map(URL::valueOf).collect(Collectors.toList());
} catch (Throwable e) {
@ -592,4 +725,4 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
public static boolean supports(URL registryURL) {
return SERVICE_REGISTRY_TYPE.equalsIgnoreCase(registryURL.getParameter(REGISTRY_TYPE_KEY));
}
}
}

4
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/CompositeMetadataServiceURLBuilder.java
View File

@ -19,7 +19,6 @@ package org.apache.dubbo.registry.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.metadata.proxy.MetadataServiceProxy;
import java.util.Iterator;
import java.util.LinkedList;
@ -34,10 +33,9 @@ import static org.apache.dubbo.common.utils.CollectionUtils.isNotEmpty;
/**
* The implementation of {@link MetadataServiceURLBuilder} composites the multiple {@link MetadataServiceURLBuilder}
* instances are loaded by Java standard {@link ServiceLoader} will aggregate {@link URL URLs} for
* {@link MetadataServiceProxy}
* {@link MetadataService}
*
* @see MetadataServiceURLBuilder
* @see MetadataServiceProxy
* @see MetadataService
* @see URL
* @see ServiceLoader

51
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ProtocolPortsMetadataCustomizer.java Normal file
View File

@ -0,0 +1,51 @@
/*
* 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.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.metadata.WritableMetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.ServiceInstanceCustomizer;
import org.apache.dubbo.rpc.Protocol;
import static org.apache.dubbo.metadata.WritableMetadataService.getExtension;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getMetadataStorageType;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.setProtocolPort;
/**
* A Class to customize the ports of {@link Protocol protocols} into
* {@link ServiceInstance#getMetadata() the metadata of service instance}
*
* @since 2.7.4
*/
public class ProtocolPortsMetadataCustomizer implements ServiceInstanceCustomizer {
@Override
public void customize(ServiceInstance serviceInstance) {
String metadataStoredType = getMetadataStorageType(serviceInstance);
WritableMetadataService writableMetadataService = getExtension(metadataStoredType);
writableMetadataService.getExportedURLs()
.stream()
.map(URL::valueOf)
.forEach(url -> {
setProtocolPort(serviceInstance, url.getProtocol(), url.getPort());
});
}
}

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

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.WritableMetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.rpc.Protocol;
import com.alibaba.fastjson.JSON;
@ -28,15 +29,19 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import static java.lang.String.valueOf;
import static java.util.Collections.emptyMap;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_DEFAULT;
import static org.apache.dubbo.common.constants.CommonConstants.APPLICATION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
import static org.apache.dubbo.common.utils.StringUtils.isBlank;
import static org.apache.dubbo.registry.integration.RegistryProtocol.DEFAULT_REGISTER_PROVIDER_KEYS;
import static org.apache.dubbo.rpc.Constants.DEPRECATED_KEY;
/**
* The Utilities class for the {@link ServiceInstance#getMetadata() metadata of the service instance}
*
* @see StandardMetadataServiceURLBuilder
* @see ServiceInstance#getMetadata()
* @see MetadataService
* @see URL
@ -49,6 +54,16 @@ public class ServiceInstanceMetadataUtils {
*/
public static final String METADATA_SERVICE_PREFIX = "dubbo.metadata-service.";
/**
* The prefix of {@link Protocol} : "dubbo.protocols."
*/
public static final String DUBBO_PROTOCOLS_PREFIX = "dubbo.protocols.";
/**
* The suffix of port : ".port";
*/
public static final String DUBBO_PORT_SUFFIX = ".port";
/**
* The property name of metadata JSON of {@link MetadataService}'s {@link URL}
*/
@ -76,16 +91,6 @@ public class ServiceInstanceMetadataUtils {
*/
public static String METADATA_STORAGE_TYPE_PROPERTY_NAME = "dubbo.metadata.storage-type";
/**
* The property name of {@link URL url's} parameter name of Dubbo Provider host
*/
public static final String PROVIDER_HOST_PROPERTY_NAME = "provider.host";
/**
* The {@link URL url's} parameter name of Dubbo Provider port
*/
public static final String PROVIDER_PORT_PROPERTY_NAME = "provider.port";
/**
* Get the multiple {@link URL urls'} parameters of {@link MetadataService MetadataService's} Metadata
*
@ -110,30 +115,23 @@ public class ServiceInstanceMetadataUtils {
return params.getOrDefault(protocol, emptyMap());
}
/**
* The provider port from {@link ServiceInstance the specified service instance}
*
* @param serviceInstance {@link ServiceInstance the specified service instance}
* @param protocol the protocol name
* @return The protocol port if found, or <code>null</code>
*/
public static Integer getProviderPort(ServiceInstance serviceInstance, String protocol) {
Map<String, Object> params = getMetadataServiceURLParams(serviceInstance, protocol);
return getProviderPort(params);
}
public static String getProviderHost(ServiceInstance serviceInstance, String protocol) {
Map<String, Object> params = getMetadataServiceURLParams(serviceInstance, protocol);
return getProviderHost(params);
}
public static String getMetadataServiceParameter(List<URL> urls) {
Map<String, Map<String, String>> params = new HashMap<>();
urls.forEach(url -> {
String protocol = url.getProtocol();
params.put(protocol, getParams(url));
});
urls.stream()
// remove APPLICATION_KEY because service name must be present
.map(url -> url.removeParameter(APPLICATION_KEY))
// remove GROUP_KEY because service name must be present
.map(url -> url.removeParameter(GROUP_KEY))
// remove DEPRECATED_KEY because it's always false
.map(url -> url.removeParameter(DEPRECATED_KEY))
// remove TIMESTAMP_KEY because it's nonsense
.map(url -> url.removeParameter(TIMESTAMP_KEY))
.forEach(url -> {
String protocol = url.getProtocol();
params.put(protocol, getParams(url));
});
if (params.isEmpty()) {
return null;
@ -145,21 +143,9 @@ public class ServiceInstanceMetadataUtils {
private static Map<String, String> getParams(URL providerURL) {
Map<String, String> params = new LinkedHashMap<>();
setDefaultParams(params, providerURL);
// set provider host
setProviderHostParam(params, providerURL);
// set provider port
setProviderPortParam(params, providerURL);
return params;
}
public static String getProviderHost(Map<String, Object> params) {
return valueOf(params.get(PROVIDER_HOST_PROPERTY_NAME));
}
public static Integer getProviderPort(Map<String, Object> params) {
return Integer.valueOf(valueOf(params.get(PROVIDER_PORT_PROPERTY_NAME)));
}
/**
* The revision for all exported Dubbo services from the specified {@link ServiceInstance}.
*
@ -226,12 +212,42 @@ public class ServiceInstanceMetadataUtils {
|| metadata.containsKey(METADATA_SERVICE_URLS_PROPERTY_NAME);
}
private static void setProviderHostParam(Map<String, String> params, URL providerURL) {
params.put(PROVIDER_HOST_PROPERTY_NAME, providerURL.getHost());
/**
* Create the property name of Dubbo protocol port
*
* @param protocol the name of protocol, e.g, dubbo, rest, and so on
* @return e.g, "dubbo.protocols.dubbo.port"
*/
public static String createProtocolPortPropertyName(String protocol) {
return DUBBO_PROTOCOLS_PREFIX + protocol + DUBBO_PORT_SUFFIX;
}
private static void setProviderPortParam(Map<String, String> params, URL providerURL) {
params.put(PROVIDER_PORT_PROPERTY_NAME, valueOf(providerURL.getPort()));
/**
* Set the protocol port into {@link ServiceInstance#getMetadata() the metadata of service instance}
*
* @param serviceInstance {@link ServiceInstance service instance}
* @param protocol the name of protocol, e.g, dubbo, rest, and so on
* @param port the port of protocol
*/
public static void setProtocolPort(ServiceInstance serviceInstance, String protocol, int port) {
Map<String, String> metadata = serviceInstance.getMetadata();
String propertyName = createProtocolPortPropertyName(protocol);
metadata.put(propertyName, Integer.toString(port));
}
/**
* Get the property value of port by the specified {@link ServiceInstance#getMetadata() the metadata of
* service instance} and protocol
*
* @param serviceInstance {@link ServiceInstance service instance}
* @param protocol the name of protocol, e.g, dubbo, rest, and so on
* @return if not found, return <code>null</code>
*/
public static Integer getProtocolPort(ServiceInstance serviceInstance, String protocol) {
Map<String, String> metadata = serviceInstance.getMetadata();
String propertyName = createProtocolPortPropertyName(protocol);
String propertyValue = metadata.get(propertyName);
return propertyValue == null ? null : Integer.valueOf(propertyValue);
}
/**

20
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/StandardMetadataServiceURLBuilder.java
View File

@ -26,9 +26,9 @@ import java.util.List;
import java.util.Map;
import static java.lang.String.valueOf;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getMetadataServiceURLsParams;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProviderHost;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProviderPort;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getProtocolPort;
/**
* The {@link MetadataServiceURLBuilder} implementation for The standard Dubbo scenario
@ -51,14 +51,15 @@ public class StandardMetadataServiceURLBuilder implements MetadataServiceURLBuil
List<URL> urls = new ArrayList<>(paramsMap.size());
for (Map.Entry<String, Map<String, Object>> entry : paramsMap.entrySet()) {
String serviceName = serviceInstance.getServiceName();
URLBuilder urlBuilder = new URLBuilder();
String host = serviceInstance.getHost();
for (Map.Entry<String, Map<String, Object>> entry : paramsMap.entrySet()) {
String protocol = entry.getKey();
Map<String, Object> urlParams = entry.getValue();
String host = getProviderHost(urlParams);
Integer port = getProviderPort(urlParams);
urlBuilder.setHost(host)
Integer port = getProtocolPort(serviceInstance, protocol);
URLBuilder urlBuilder = new URLBuilder()
.setHost(host)
.setPort(port)
.setProtocol(protocol)
.setPath(MetadataService.class.getName());
@ -66,6 +67,9 @@ public class StandardMetadataServiceURLBuilder implements MetadataServiceURLBuil
// add parameters
entry.getValue().forEach((name, value) -> urlBuilder.addParameter(name, valueOf(value)));
// add the default parameters
urlBuilder.addParameter(GROUP_KEY, serviceName);
urls.add(urlBuilder.build());
}

58
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/URLRevisionResolver.java
View File

@ -22,7 +22,14 @@ import org.apache.dubbo.metadata.MetadataService;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.stream.Collectors;
import static org.apache.dubbo.common.constants.CommonConstants.PID_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.TIMESTAMP_KEY;
import static org.apache.dubbo.common.utils.CollectionUtils.isEmpty;
/**
@ -46,23 +53,50 @@ public class URLRevisionResolver {
return NO_REVISION;
}
return urls.stream()
.map(URL::valueOf) // String to URL
.map(URL::getServiceInterface) // get the service interface
.filter(this::isNotMetadataService) // filter not MetadataService interface
.map(ClassUtils::forName) // load business interface class
.map(Class::getMethods) // get all public methods from business interface
.map(Arrays::asList) // Array to List
.flatMap(Collection::stream) // flat Stream<Stream> to be Stream
.map(Object::toString) // Method to String
.sorted() // sort methods marking sure the calculation of reversion is stable
.map(URLRevisionResolver::hashCode) // generate Long hashCode
List<URL> urlsList = toURLsList(urls);
SortedSet<String> methodSignatures = resolveMethodSignatures(urlsList);
SortedSet<String> urlParameters = resolveURLParameters(urlsList);
SortedSet<String> values = new TreeSet<>(methodSignatures);
values.addAll(urlParameters);
return values.stream()
.map(this::hashCode) // generate Long hashCode
.reduce(Long::sum) // sum hashCode
.map(String::valueOf) // Long to String
.orElse(NO_REVISION); // NO_REVISION as default
}
private static long hashCode(String value) {
private List<URL> toURLsList(Collection<String> urls) {
return urls.stream()
.map(URL::valueOf) // String to URL
.filter(url -> isNotMetadataService(url.getServiceInterface())) // filter not MetadataService interface
.collect(Collectors.toList());
}
private SortedSet<String> resolveMethodSignatures(List<URL> urls) {
return urls.stream()
.map(URL::getServiceInterface) // get the service interface
.map(ClassUtils::forName) // load business interface class
.map(Class::getMethods) // get all public methods from business interface
.map(Arrays::asList) // Array to List
.flatMap(Collection::stream) // flat Stream<Stream> to be Stream
.map(Object::toString) // Method to String
.collect(TreeSet::new, Set::add, Set::addAll); // sort and remove the duplicate
}
private SortedSet<String> resolveURLParameters(Collection<URL> urls) {
return urls.stream()
.map(url -> url.removeParameter(PID_KEY))
.map(url -> url.removeParameter(TIMESTAMP_KEY))
.map(URL::toParameterString)
.collect(TreeSet::new, Set::add, Set::addAll); // sort and remove the duplicate
}
private long hashCode(String value) {
long h = 0;
char[] chars = value.toCharArray();
for (int i = 0; i < chars.length; i++) {

122
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/proxy/MetadataServiceProxy.java
View File

@ -1,122 +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.registry.client.metadata.proxy;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.proxy.InvokerInvocationHandler;
import java.util.Iterator;
import java.util.List;
import java.util.SortedSet;
import java.util.function.Function;
import static java.lang.reflect.Proxy.newProxyInstance;
import static org.apache.dubbo.registry.client.metadata.MetadataServiceURLBuilder.composite;
/**
* The Proxy object for the {@link MetadataService} whose {@link ServiceInstance} providers may export multiple
* {@link Protocol protocols} at the same time.
*
* @see ServiceInstance
* @see MetadataService
* @since 2.7.4
*/
public class MetadataServiceProxy implements MetadataService {
private final Logger logger = LoggerFactory.getLogger(getClass());
private final List<URL> urls;
private final Protocol protocol;
public MetadataServiceProxy(ServiceInstance serviceInstance, Protocol protocol) {
this(composite().build(serviceInstance), protocol);
}
public MetadataServiceProxy(List<URL> urls, Protocol protocol) {
this.urls = urls;
this.protocol = protocol;
}
@Override
public String serviceName() {
return doInMetadataService(MetadataService::serviceName);
}
@Override
public SortedSet<String> getExportedURLs(String serviceInterface, String group, String version, String protocol) {
return doInMetadataService(metadataService ->
metadataService.getExportedURLs(serviceInterface, group, version, protocol));
}
@Override
public String getServiceDefinition(String interfaceName, String version, String group) {
return doInMetadataService(metadataService ->
metadataService.getServiceDefinition(interfaceName, version, group));
}
@Override
public String getServiceDefinition(String serviceKey) {
return doInMetadataService(metadataService ->
metadataService.getServiceDefinition(serviceKey));
}
protected <T> T doInMetadataService(Function<MetadataService, T> callback) {
T result = null; // execution result
Throwable exception = null; // exception maybe present
Iterator<URL> iterator = urls.iterator();
while (iterator.hasNext()) { // Executes MetadataService's method until success
URL url = iterator.next();
Invoker<MetadataService> invoker = null;
try {
invoker = this.protocol.refer(MetadataService.class, url);
MetadataService proxy = (MetadataService) newProxyInstance(getClass().getClassLoader(),
new Class[]{MetadataService.class}, new InvokerInvocationHandler(invoker));
result = callback.apply(proxy);
exception = null;
} catch (Throwable e) {
exception = e;
// If met with some error, invoke next
if (logger.isErrorEnabled()) {
logger.error(e.getMessage(), e);
}
} finally {
if (invoker != null) {
// to destroy the Invoker finally
invoker.destroy();
invoker = null;
}
}
}
if (exception != null) { // If all executions were failed
throw new RuntimeException(exception.getMessage(), exception);
}
return result;
}
}

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

@ -37,7 +37,7 @@ public class RandomServiceInstanceSelector implements ServiceInstanceSelector {
if (size < 1) {
return null;
}
int index = selectIndexRandomly(size);
int index = size == 1 ? 0 : selectIndexRandomly(size);
return serviceInstances.get(index);
}

1
dubbo-registry/dubbo-registry-api/src/main/resources/META-INF/services/org.apache.dubbo.registry.client.ServiceInstanceCustomizer
View File

@ -2,3 +2,4 @@ org.apache.dubbo.registry.client.metadata.MetadataServiceURLParamsMetadataCustom
org.apache.dubbo.registry.client.metadata.ExportedServicesRevisionMetadataCustomizer
org.apache.dubbo.registry.client.metadata.SubscribedServicesRevisionMetadataCustomizer
org.apache.dubbo.registry.client.metadata.RefreshServiceMetadataCustomizer
org.apache.dubbo.registry.client.metadata.ProtocolPortsMetadataCustomizer

13
dubbo-registry/dubbo-registry-eureka/pom.xml
View File

@ -27,7 +27,20 @@
<dependency>
<groupId>com.netflix.eureka</groupId>
<artifactId>eureka-client</artifactId>
<exclusions>
<exclusion>
<groupId>javax.ws.rs</groupId>
<artifactId>jsr311-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>javax.ws.rs</groupId>
<artifactId>javax.ws.rs-api</artifactId>
<version>2.1</version>
</dependency>
<dependency>
<groupId>javax.inject</groupId>
<artifactId>javax.inject</artifactId>