diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java index d74d57dd2d..2939eb352c 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java @@ -112,6 +112,8 @@ public interface CommonConstants { String EXECUTOR_SERVICE_COMPONENT_KEY = ExecutorService.class.getName(); + String INTERNAL_EXECUTOR_SERVICE_COMPONENT_KEY = "INTERNAL_SERVICE_EXECUTOR"; + String THREADPOOL_KEY = "threadpool"; String THREAD_NAME_KEY = "threadname"; diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java index 78621c2213..0942d88650 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/manager/DefaultExecutorRepository.java @@ -30,6 +30,7 @@ import org.apache.dubbo.config.ModuleConfig; import org.apache.dubbo.config.ProviderConfig; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.ModuleModel; +import org.apache.dubbo.rpc.model.ServiceDescriptor; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -43,6 +44,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_EXPORT_THREAD_NUM; import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_REFER_THREAD_NUM; import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.INTERNAL_EXECUTOR_SERVICE_COMPONENT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY; import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY; import static org.apache.dubbo.common.constants.CommonConstants.THREAD_NAME_KEY; @@ -78,7 +80,7 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA */ @Override public synchronized ExecutorService createExecutorIfAbsent(URL url) { - Map executors = data.computeIfAbsent(EXECUTOR_SERVICE_COMPONENT_KEY, k -> new ConcurrentHashMap<>()); + Map executors = data.computeIfAbsent(getExecutorKey(url), k -> new ConcurrentHashMap<>()); // Consumer's executor is sharing globally, key=Integer.MAX_VALUE. Provider's executor is sharing by protocol. Integer portKey = CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(SIDE_KEY)) ? Integer.MAX_VALUE : url.getPort(); if (url.getParameter(THREAD_NAME_KEY) == null) { @@ -95,13 +97,30 @@ public class DefaultExecutorRepository implements ExecutorRepository, ExtensionA return executor; } + /** + * Return the executor key based on the type (internal or biz) of the current service. + * + * @param url + * @return + */ + private String getExecutorKey(URL url) { + String executorKey = INTERNAL_EXECUTOR_SERVICE_COMPONENT_KEY; + ServiceDescriptor serviceDescriptor = applicationModel.getInternalModule().getServiceRepository().lookupService(url.getServiceInterface()); + // if not found in internal service repository, then it's biz service defined by user. + if (serviceDescriptor == null) { + executorKey = EXECUTOR_SERVICE_COMPONENT_KEY; + + } + return executorKey; + } + private ExecutorService createExecutor(URL url) { return (ExecutorService) extensionAccessor.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url); } @Override public ExecutorService getExecutor(URL url) { - Map executors = data.get(EXECUTOR_SERVICE_COMPONENT_KEY); + Map executors = data.get(getExecutorKey(url)); /* * It's guaranteed that this method is called after {@link #createExecutorIfAbsent(URL)}, so data should already diff --git a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleModel.java b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleModel.java index 163273cde6..89929a9b5b 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleModel.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleModel.java @@ -16,6 +16,7 @@ */ package org.apache.dubbo.rpc.model; +import org.apache.dubbo.common.URL; import org.apache.dubbo.common.config.ModuleEnvironment; import org.apache.dubbo.common.context.ModuleExt; import org.apache.dubbo.common.deploy.ApplicationDeployer; @@ -28,6 +29,7 @@ import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.common.utils.Assert; import org.apache.dubbo.config.context.ModuleConfigManager; +import java.util.HashMap; import java.util.Set; /** @@ -173,4 +175,22 @@ public class ModuleModel extends ScopeModel { public void setModuleEnvironment(ModuleEnvironment moduleEnvironment) { this.moduleEnvironment = moduleEnvironment; } + + public ConsumerModel registerInternalConsumer(Class internalService, URL url) { + ServiceMetadata serviceMetadata = new ServiceMetadata(); + serviceMetadata.setVersion(url.getVersion()); + serviceMetadata.setGroup(url.getGroup()); + serviceMetadata.setDefaultGroup(url.getGroup()); + serviceMetadata.setServiceInterfaceName(internalService.getName()); + serviceMetadata.setServiceType(internalService); + String servyceKey = URL.buildKey(internalService.getName(), url.getGroup(), url.getVersion()); + serviceMetadata.setServiceKey(servyceKey); + + ConsumerModel consumerModel = new ConsumerModel(serviceMetadata.getServiceKey(), "jdk", serviceRepository.lookupService(serviceMetadata.getServiceInterfaceName()), null, + this, serviceMetadata, new HashMap<>()); + + logger.info("Dynamically registering consumer model " + servyceKey + " into model " + this.getDesc()); + serviceRepository.registerConsumer(consumerModel); + return consumerModel; + } } diff --git a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java index 38b620fe4f..5a24a0bf8a 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ModuleServiceRepository.java @@ -205,7 +205,7 @@ public class ModuleServiceRepository { } public ServiceDescriptor lookupService(String interfaceName) { - if (services.containsKey(interfaceName)) { + if (interfaceName != null && services.containsKey(interfaceName)) { List serviceDescriptors = services.get(interfaceName); return serviceDescriptors.size() > 0 ? serviceDescriptors.get(0) : null; } else { diff --git a/dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepositoryTest.java b/dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepositoryTest.java index ce81e11922..657d21eaeb 100644 --- a/dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepositoryTest.java +++ b/dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/manager/ExecutorRepositoryTest.java @@ -46,8 +46,8 @@ public class ExecutorRepositoryTest { @Test public void testGetExecutor() { - testGet(URL.valueOf("dubbo://127.0.0.1:23456")); - testGet(URL.valueOf("dubbo://127.0.0.1:23456?side=consumer")); + testGet(URL.valueOf("dubbo://127.0.0.1:23456/TestService")); + testGet(URL.valueOf("dubbo://127.0.0.1:23456/TestService?side=consumer")); Assertions.assertNotNull(executorRepository.getSharedExecutor()); Assertions.assertNotNull(executorRepository.getServiceExportExecutor()); @@ -68,7 +68,7 @@ public class ExecutorRepositoryTest { @Test public void testUpdateExecutor() { - URL url = URL.valueOf("dubbo://127.0.0.1:23456?threads=5"); + URL url = URL.valueOf("dubbo://127.0.0.1:23456/TestService?threads=5"); ThreadPoolExecutor executorService = (ThreadPoolExecutor) executorRepository.createExecutorIfAbsent(url); executorService.setCorePoolSize(3); diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporter.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporter.java index 68117d18b0..568aae63dd 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporter.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporter.java @@ -33,14 +33,18 @@ import org.apache.dubbo.rpc.ProtocolServer; import org.apache.dubbo.rpc.model.ApplicationModel; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import static java.util.Collections.emptyList; +import static org.apache.dubbo.common.constants.CommonConstants.CORE_THREADS_KEY; import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_PROTOCOL; import static org.apache.dubbo.common.constants.CommonConstants.METADATA_SERVICE_PORT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.METADATA_SERVICE_PROTOCOL_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY; /** * Export metadata service @@ -178,6 +182,11 @@ public class ConfigurableMetadataServiceExporter { serviceConfig.setMethods(generateMethodConfig()); serviceConfig.setConnections(1); // separate connection serviceConfig.setExecutes(100); // max tasks running at the same time + Map threadParams = new HashMap<>(); + threadParams.put(THREADPOOL_KEY, "cached"); + threadParams.put(THREADS_KEY, "100"); + threadParams.put(CORE_THREADS_KEY, "2"); + serviceConfig.setParameters(threadParams); return serviceConfig; } diff --git a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/AbstractServiceNameMapping.java b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/AbstractServiceNameMapping.java index 305319f1d4..3e47b8e417 100644 --- a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/AbstractServiceNameMapping.java +++ b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/AbstractServiceNameMapping.java @@ -248,6 +248,7 @@ public abstract class AbstractServiceNameMapping implements ServiceNameMapping, mappingListeners.clear(); mappingLocks.clear(); mappingInitStatus.clear(); + } private class AsyncMappingTask implements Callable> { diff --git a/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataServiceDetector.java b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataServiceDetector.java new file mode 100644 index 0000000000..a992cdc586 --- /dev/null +++ b/dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/MetadataServiceDetector.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.metadata; + +import org.apache.dubbo.rpc.model.BuiltinServiceDetector; + +public class MetadataServiceDetector implements BuiltinServiceDetector { + + @Override + public Class getService() { + return MetadataService.class; + } + +} diff --git a/dubbo-metadata/dubbo-metadata-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.model.BuiltinServiceDetector b/dubbo-metadata/dubbo-metadata-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.model.BuiltinServiceDetector new file mode 100644 index 0000000000..804f1f0718 --- /dev/null +++ b/dubbo-metadata/dubbo-metadata-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.model.BuiltinServiceDetector @@ -0,0 +1 @@ +metadata=org.apache.dubbo.metadata.MetadataServiceDetector diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java index bc6ddc13e0..2e44b5d65f 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataUtils.java @@ -34,7 +34,8 @@ import org.apache.dubbo.rpc.Invoker; import org.apache.dubbo.rpc.Protocol; import org.apache.dubbo.rpc.ProxyFactory; import org.apache.dubbo.rpc.model.ApplicationModel; -import org.apache.dubbo.rpc.model.ScopeModel; +import org.apache.dubbo.rpc.model.ConsumerModel; +import org.apache.dubbo.rpc.model.ModuleModel; import org.apache.dubbo.rpc.model.ServiceDescriptor; import org.apache.dubbo.rpc.service.Destroyable; @@ -105,7 +106,7 @@ public class MetadataUtils { } } - public static MetadataService referProxy(ServiceInstance instance) { + public static ProxyHolder referProxy(ServiceInstance instance) { MetadataServiceURLBuilder builder; ExtensionLoader loader = instance.getApplicationModel() .getExtensionLoader(MetadataServiceURLBuilder.class); @@ -122,16 +123,19 @@ public class MetadataUtils { List urls = builder.build(instance); if (CollectionUtils.isEmpty(urls)) { throw new IllegalStateException("Introspection service discovery mode is enabled " - + instance + ", but no metadata service can build from it."); + + instance + ", but no metadata service can build from it."); } // Simply rely on the first metadata url, as stated in MetadataServiceURLBuilder. - ScopeModel scopeModel = instance.getApplicationModel(); - Protocol protocol = scopeModel.getExtensionLoader(Protocol.class).getAdaptiveExtension(); + ApplicationModel applicationModel = instance.getApplicationModel(); + ModuleModel internalModel = applicationModel.getInternalModule(); + ConsumerModel consumerModel = applicationModel.getInternalModule().registerInternalConsumer(MetadataService.class, urls.get(0)); + + Protocol protocol = applicationModel.getExtensionLoader(Protocol.class).getAdaptiveExtension(); Invoker invoker = protocol.refer(MetadataService.class, urls.get(0)); - ProxyFactory proxyFactory = scopeModel.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension(); - return proxyFactory.getProxy(invoker); + ProxyFactory proxyFactory = applicationModel.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension(); + return new ProxyHolder(consumerModel, proxyFactory.getProxy(invoker), internalModel); } public static MetadataInfo getRemoteMetadata(String revision, List instances, MetadataReport metadataReport) { @@ -147,13 +151,12 @@ public class MetadataUtils { } else { // change the instance used to communicate to avoid all requests route to the same instance MetadataService metadataServiceProxy = null; + ProxyHolder proxyHolder = null; try { - metadataServiceProxy = MetadataUtils.referProxy(instance); - metadataInfo = metadataServiceProxy.getMetadataInfo(ServiceInstanceMetadataUtils.getExportedServicesRevision(instance)); + proxyHolder = MetadataUtils.referProxy(instance); + metadataInfo = proxyHolder.getProxy().getMetadataInfo(ServiceInstanceMetadataUtils.getExportedServicesRevision(instance)); } finally { - if (metadataServiceProxy instanceof Destroyable) { - ((Destroyable)metadataServiceProxy).$destroy(); - } + MetadataUtils.destroyProxy(proxyHolder); } } } catch (Exception e) { @@ -167,6 +170,12 @@ public class MetadataUtils { return metadataInfo; } + public static void destroyProxy(ProxyHolder proxyHolder) { + if (proxyHolder != null) { + proxyHolder.destroy(); + } + } + public static MetadataInfo getMetadata(String revision, ServiceInstance instance, MetadataReport metadataReport) { SubscriberMetadataIdentifier identifier = new SubscriberMetadataIdentifier(instance.getServiceName(), revision); @@ -194,4 +203,35 @@ public class MetadataUtils { return instances.get(ThreadLocalRandom.current().nextInt(0, instances.size())); } + private static class ProxyHolder { + private final ConsumerModel consumerModel; + private final MetadataService proxy; + private final ModuleModel internalModel; + + public ProxyHolder(ConsumerModel consumerModel, MetadataService proxy, ModuleModel internalModel) { + this.consumerModel = consumerModel; + this.proxy = proxy; + this.internalModel = internalModel; + } + + public void destroy() { + if (proxy instanceof Destroyable) { + ((Destroyable) proxy).$destroy(); + } + internalModel.getServiceRepository().unregisterConsumer(consumerModel); + } + + public ConsumerModel getConsumerModel() { + return consumerModel; + } + + public MetadataService getProxy() { + return proxy; + } + + public ModuleModel getInternalModel() { + return internalModel; + } + } + } 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 0094b0a2bd..f2cc886331 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 @@ -33,12 +33,15 @@ import java.util.Map; import static java.util.Collections.emptyMap; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER; +import static org.apache.dubbo.common.constants.CommonConstants.CORE_THREADS_KEY; import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_PROTOCOL; import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PORT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY; import static org.apache.dubbo.common.constants.CommonConstants.RETRIES_KEY; import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY; +import static org.apache.dubbo.common.constants.CommonConstants.THREADS_KEY; import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY; import static org.apache.dubbo.common.utils.StringUtils.isBlank; @@ -98,13 +101,16 @@ public class StandardMetadataServiceURLBuilder implements MetadataServiceURLBuil String protocol = params.get(PROTOCOL_KEY); int port = Integer.parseInt(params.get(PORT_KEY)); URLBuilder urlBuilder = new URLBuilder() - .setHost(host) - .setPort(port) - .setProtocol(protocol) - .setPath(MetadataService.class.getName()) - .addParameter(TIMEOUT_KEY, ConfigurationUtils.get(applicationModel, METADATA_PROXY_TIMEOUT_KEY, DEFAULT_METADATA_TIMEOUT_VALUE)) - .addParameter(SIDE_KEY, CONSUMER) - .addParameter(CONNECTIONS_KEY, 1) + .setHost(host) + .setPort(port) + .setProtocol(protocol) + .setPath(MetadataService.class.getName()) + .addParameter(TIMEOUT_KEY, ConfigurationUtils.get(applicationModel, METADATA_PROXY_TIMEOUT_KEY, DEFAULT_METADATA_TIMEOUT_VALUE)) + .addParameter(SIDE_KEY, CONSUMER) + .addParameter(CONNECTIONS_KEY, 1) + .addParameter(THREADPOOL_KEY, "cached") + .addParameter(THREADS_KEY, "100") + .addParameter(CORE_THREADS_KEY, "2") .addParameter(RETRIES_KEY, 0); diff --git a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java index 65061d65ec..734d173413 100644 --- a/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java +++ b/dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java @@ -132,19 +132,19 @@ public class TripleProtocol extends AbstractProtocol { public Invoker refer(Class type, URL url) throws RpcException { optimizeSerialization(url); ExecutorService streamExecutor = getOrCreateStreamExecutor( - url.getOrDefaultApplicationModel()); + url.getOrDefaultApplicationModel(), url); TripleInvoker invoker = new TripleInvoker<>(type, url, acceptEncodings, connectionManager, invokers, streamExecutor); invokers.add(invoker); return invoker; } - private ExecutorService getOrCreateStreamExecutor(ApplicationModel applicationModel) { + private ExecutorService getOrCreateStreamExecutor(ApplicationModel applicationModel, URL url) { ExecutorService executor = applicationModel.getExtensionLoader(ExecutorRepository.class) .getDefaultExtension() - .createExecutorIfAbsent(THREAD_POOL_URL); + .createExecutorIfAbsent(url); Objects.requireNonNull(executor, - String.format("No available executor found in %s", THREAD_POOL_URL)); + String.format("No available executor found in %s", url)); return executor; }