diff --git a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderMinimumBootstrap.java b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderMinimumBootstrap.java index 4e1958bb14..58760e5340 100644 --- a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderMinimumBootstrap.java +++ b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderMinimumBootstrap.java @@ -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(); } diff --git a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java index 1bddcd37e6..aeb8eb7835 100644 --- a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java +++ b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java @@ -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")) diff --git a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java index bd47a7e2a5..bf49d6944c 100644 --- a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java +++ b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java @@ -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()) diff --git a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceConsumerBootstrap.java b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceConsumerBootstrap.java new file mode 100644 index 0000000000..8b45c1f1c1 --- /dev/null +++ b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceConsumerBootstrap.java @@ -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 referenceConfig = configManager.getReference("echo"); + + EchoService echoService = referenceConfig.get(); + + ReferenceConfig 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)); + } + + + } +} diff --git a/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceProviderBootstrap.java b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceProviderBootstrap.java new file mode 100644 index 0000000000..87d548d913 --- /dev/null +++ b/dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/ZookeeperDubboServiceProviderBootstrap.java @@ -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(); + } +} diff --git a/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java b/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java index b82c8e72df..3988d0b626 100644 --- a/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java +++ b/dubbo-configcenter/dubbo-configcenter-nacos/src/main/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfiguration.java @@ -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 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 getConfigKeys(String group) { + // TODO use Nacos Client API to replace HTTP Open API + SortedSet keys = new TreeSet<>(); + try { + List 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 keysStream = toKeysStream(result.content); + keysStream.forEach(keys::add); + } catch (IOException e) { + if (logger.isErrorEnabled()) { + logger.error(e.getMessage(), e); + } + } + return keys; + } + + private Stream 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 listeners = new CopyOnWriteArraySet<>(); @@ -259,5 +323,4 @@ public class NacosDynamicConfiguration implements DynamicConfiguration { return ConfigChangeType.MODIFIED; } } - } diff --git a/dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java b/dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java index 350936c2c0..ac2978d843 100644 --- a/dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java +++ b/dubbo-configcenter/dubbo-configcenter-nacos/src/test/java/org/apache/dubbo/configcenter/support/nacos/NacosDynamicConfigurationTest.java @@ -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 keys = config.getConfigKeys(DynamicConfiguration.DEFAULT_GROUP); + + Assertions.assertFalse(keys.isEmpty()); + + } + private void put(String key, String value) { put(key, DynamicConfiguration.DEFAULT_GROUP, value); } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java index c55696790a..e8908e0e93 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscovery.java @@ -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 *

- * 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); } /** diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java index af7ec91075..d41e4bc582 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java @@ -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 listenedServices = new LinkedHashSet<>(); + private final Set 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>> 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 serviceInstances) { @@ -338,100 +351,107 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry { private List getSubscribedURLs(URL subscribedURL, Collection instances, String serviceName) { - List subscribedURLs = new LinkedList<>(); - // local service instances could be mutable List 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> 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 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 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 subscribedURLs = cloneSubscribedURLs(subscribedURL, serviceInstances); // clear local service instances serviceInstances.clear(); - return subscribedURLs; } - private List cloneSubscribedURLs(URL subscribedURL, Collection serviceInstances, - Map> revisionURLsCache) { - - // If revisionURLsCache is not empty, clone the template URLs to be the subscribed URLs - if (!revisionURLsCache.isEmpty()) { - - List clonedURLs = new LinkedList<>(); - - Iterator iterator = serviceInstances.iterator(); - - while (iterator.hasNext()) { - - ServiceInstance serviceInstance = iterator.next(); - - List templateURLs = getTemplateURLs(subscribedURL, serviceInstance, revisionURLsCache); - // The parameters of URLs that the MetadataService exported - Map> serviceURLsParams = getMetadataServiceURLsParams(serviceInstance); - - templateURLs.forEach(templateURL -> { - - String protocol = templateURL.getProtocol(); - - Map 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 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 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 serviceInstances) { + + if (isEmpty(serviceInstances)) { // if empty, return immediately + return; } - return Collections.emptyList(); + String serviceName = serviceInstances.get(0).getServiceName(); + + synchronized (this) { + + // revisionExportedURLs is mutable + Map> revisionExportedURLs = serviceExportedURLsCache.computeIfAbsent(serviceName, s -> new HashMap<>()); + + if (revisionExportedURLs.isEmpty()) { // if empty, return immediately + return; + } + + Set existedRevisions = revisionExportedURLs.keySet(); // read-only + Set currentRevisions = serviceInstances.stream() + .map(ServiceInstanceMetadataUtils::getExportedServicesRevision) + .collect(Collectors.toSet()); + // staleRevisions = existedRevisions(copy) - currentRevisions + Set staleRevisions = new HashSet<>(existedRevisions); + staleRevisions.removeAll(currentRevisions); + // remove exported URLs if staled + staleRevisions.forEach(revisionExportedURLs::remove); + } + } + + private List cloneSubscribedURLs(URL subscribedURL, Collection serviceInstances) { + + if (isEmpty(serviceInstances)) { + return emptyList(); + } + + List 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 null if serviceInstances is empty. */ private ServiceInstance selectServiceInstance(List 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}. *

+ * First, put the revision {@link ServiceInstance service instance} + * associating {@link #getExportedURLs(URL, ServiceInstance) exported URLs} into cache. + *

+ * 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. + *

+ * 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 getTemplateURLs(URL subscribedURL, ServiceInstance selectedInstance) { + + List exportedURLs = getRevisionExportedURLs(selectedInstance); + + if (isEmpty(exportedURLs)) { + return emptyList(); + } + + return filterSubscribedURLs(subscribedURL, exportedURLs); + } + + private List filterSubscribedURLs(URL subscribedURL, List 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 + *

* 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}. - *

- * First, put the revision {@link ServiceInstance service instance} - * associating {@link #getProviderExportedURLs(URL, ServiceInstance) exported URLs} into cache. - *

- * 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. - *

- * 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 getTemplateURLs(URL subscribedURL, ServiceInstance selectedInstance, - Map> revisionURLsCache) { - // get the revision from the specified {@link ServiceInstance} - String revision = getExportedServicesRevision(selectedInstance); - // try to get templateURLs from cache - List templateURLs = revisionURLsCache.get(revision); + private List 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 exportedURLs = null; + + synchronized (this) { // It's required to lock here because it may run in the sync or async mode + + Map> 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 : + *

    + *
  1. The normal result
    2. + *
  2. The empty result if the {@link ServiceInstance service instance} did not export yet
    3. + *
  3. null if there is an invocation error on {@link MetadataService} proxy
    4. + *
+ */ + private List getExportedURLs(ServiceInstance providerServiceInstance) { + + List exportedURLs = null; + + String metadataStorageType = getMetadataStorageType(providerServiceInstance); + + try { + MetadataService metadataService = MetadataServiceProxyFactory + .getExtension(metadataStorageType == null ? DEFAULT_EXTENSION : metadataStorageType) + .getProxy(providerServiceInstance); + SortedSet 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 getProviderExportedURLs(URL subscribedURL, ServiceInstance providerInstance) { + protected List getExportedURLs(URL subscribedURL, ServiceInstance providerServiceInstance) { List 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 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)); } -} +} \ No newline at end of file diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/CompositeMetadataServiceURLBuilder.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/CompositeMetadataServiceURLBuilder.java index 8006c852a7..ef19d8666f 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/CompositeMetadataServiceURLBuilder.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/CompositeMetadataServiceURLBuilder.java @@ -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 diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ProtocolPortsMetadataCustomizer.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ProtocolPortsMetadataCustomizer.java new file mode 100644 index 0000000000..f7f8821005 --- /dev/null +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ProtocolPortsMetadataCustomizer.java @@ -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()); + }); + } +} diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java index 98f9d14b14..31a3e1d58d 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java @@ -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 null - */ - public static Integer getProviderPort(ServiceInstance serviceInstance, String protocol) { - Map params = getMetadataServiceURLParams(serviceInstance, protocol); - return getProviderPort(params); - } - - public static String getProviderHost(ServiceInstance serviceInstance, String protocol) { - Map params = getMetadataServiceURLParams(serviceInstance, protocol); - return getProviderHost(params); - } - public static String getMetadataServiceParameter(List urls) { + Map> 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 getParams(URL providerURL) { Map 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 params) { - return valueOf(params.get(PROVIDER_HOST_PROPERTY_NAME)); - } - - public static Integer getProviderPort(Map 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 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 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 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 null + */ + public static Integer getProtocolPort(ServiceInstance serviceInstance, String protocol) { + Map metadata = serviceInstance.getMetadata(); + String propertyName = createProtocolPortPropertyName(protocol); + String propertyValue = metadata.get(propertyName); + return propertyValue == null ? null : Integer.valueOf(propertyValue); } /** diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/StandardMetadataServiceURLBuilder.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/StandardMetadataServiceURLBuilder.java index 48022af20a..15660cbbd2 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/StandardMetadataServiceURLBuilder.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/StandardMetadataServiceURLBuilder.java @@ -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 urls = new ArrayList<>(paramsMap.size()); - for (Map.Entry> entry : paramsMap.entrySet()) { + String serviceName = serviceInstance.getServiceName(); - URLBuilder urlBuilder = new URLBuilder(); + String host = serviceInstance.getHost(); + + for (Map.Entry> entry : paramsMap.entrySet()) { String protocol = entry.getKey(); - Map 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()); } diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/URLRevisionResolver.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/URLRevisionResolver.java index a8acd3d33c..d8b994d0f0 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/URLRevisionResolver.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/URLRevisionResolver.java @@ -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 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 urlsList = toURLsList(urls); + + SortedSet methodSignatures = resolveMethodSignatures(urlsList); + + SortedSet urlParameters = resolveURLParameters(urlsList); + + SortedSet 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 toURLsList(Collection urls) { + return urls.stream() + .map(URL::valueOf) // String to URL + .filter(url -> isNotMetadataService(url.getServiceInterface())) // filter not MetadataService interface + .collect(Collectors.toList()); + } + + private SortedSet resolveMethodSignatures(List 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 to be Stream + .map(Object::toString) // Method to String + .collect(TreeSet::new, Set::add, Set::addAll); // sort and remove the duplicate + } + + private SortedSet resolveURLParameters(Collection 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++) { diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/proxy/MetadataServiceProxy.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/proxy/MetadataServiceProxy.java deleted file mode 100644 index a5b73840f4..0000000000 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/proxy/MetadataServiceProxy.java +++ /dev/null @@ -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 urls; - - private final Protocol protocol; - - public MetadataServiceProxy(ServiceInstance serviceInstance, Protocol protocol) { - this(composite().build(serviceInstance), protocol); - } - - public MetadataServiceProxy(List urls, Protocol protocol) { - this.urls = urls; - this.protocol = protocol; - } - - @Override - public String serviceName() { - return doInMetadataService(MetadataService::serviceName); - } - - @Override - public SortedSet 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 doInMetadataService(Function callback) { - - T result = null; // execution result - - Throwable exception = null; // exception maybe present - - Iterator iterator = urls.iterator(); - - while (iterator.hasNext()) { // Executes MetadataService's method until success - URL url = iterator.next(); - Invoker 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; - } -} diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/selector/RandomServiceInstanceSelector.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/selector/RandomServiceInstanceSelector.java index ff5030469e..605575f01c 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/selector/RandomServiceInstanceSelector.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/selector/RandomServiceInstanceSelector.java @@ -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); } diff --git a/dubbo-registry/dubbo-registry-api/src/main/resources/META-INF/services/org.apache.dubbo.registry.client.ServiceInstanceCustomizer b/dubbo-registry/dubbo-registry-api/src/main/resources/META-INF/services/org.apache.dubbo.registry.client.ServiceInstanceCustomizer index 7b068938bf..a83cb72415 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/resources/META-INF/services/org.apache.dubbo.registry.client.ServiceInstanceCustomizer +++ b/dubbo-registry/dubbo-registry-api/src/main/resources/META-INF/services/org.apache.dubbo.registry.client.ServiceInstanceCustomizer @@ -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 diff --git a/dubbo-registry/dubbo-registry-eureka/pom.xml b/dubbo-registry/dubbo-registry-eureka/pom.xml index cd4181da7f..79bbc22c31 100644 --- a/dubbo-registry/dubbo-registry-eureka/pom.xml +++ b/dubbo-registry/dubbo-registry-eureka/pom.xml @@ -27,7 +27,20 @@ com.netflix.eureka eureka-client + + + javax.ws.rs + jsr311-api + + + + + javax.ws.rs + javax.ws.rs-api + 2.1 + + javax.inject javax.inject