register MetadataService as internal service (#10148)
* register MetadataService as internal service * fix ut * register consumer model when generating proxy for internal service * check model type directly * check by looking up service for model type here is not clear * unify thread configuration
This commit is contained in:
ken.lj
GitHub
parent
14d8931339
commit
2face28201
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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<Integer, ExecutorService> executors = data.computeIfAbsent(EXECUTOR_SERVICE_COMPONENT_KEY, k -> new ConcurrentHashMap<>());
|
||||
Map<Integer, ExecutorService> 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<Integer, ExecutorService> executors = data.get(EXECUTOR_SERVICE_COMPONENT_KEY);
|
||||
Map<Integer, ExecutorService> executors = data.get(getExecutorKey(url));
|
||||
|
||||
/*
|
||||
* It's guaranteed that this method is called after {@link #createExecutorIfAbsent(URL)}, so data should already
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -205,7 +205,7 @@ public class ModuleServiceRepository {
|
|||
}
|
||||
|
||||
public ServiceDescriptor lookupService(String interfaceName) {
|
||||
if (services.containsKey(interfaceName)) {
|
||||
if (interfaceName != null && services.containsKey(interfaceName)) {
|
||||
List<ServiceDescriptor> serviceDescriptors = services.get(interfaceName);
|
||||
return serviceDescriptors.size() > 0 ? serviceDescriptors.get(0) : null;
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<String, String> threadParams = new HashMap<>();
|
||||
threadParams.put(THREADPOOL_KEY, "cached");
|
||||
threadParams.put(THREADS_KEY, "100");
|
||||
threadParams.put(CORE_THREADS_KEY, "2");
|
||||
serviceConfig.setParameters(threadParams);
|
||||
|
||||
return serviceConfig;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -248,6 +248,7 @@ public abstract class AbstractServiceNameMapping implements ServiceNameMapping,
|
|||
mappingListeners.clear();
|
||||
mappingLocks.clear();
|
||||
mappingInitStatus.clear();
|
||||
|
||||
}
|
||||
|
||||
private class AsyncMappingTask implements Callable<Set<String>> {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
metadata=org.apache.dubbo.metadata.MetadataServiceDetector
|
||||
|
|
@ -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<MetadataServiceURLBuilder> loader = instance.getApplicationModel()
|
||||
.getExtensionLoader(MetadataServiceURLBuilder.class);
|
||||
|
|
@ -122,16 +123,19 @@ public class MetadataUtils {
|
|||
List<URL> 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<MetadataService> 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<ServiceInstance> 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;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -132,19 +132,19 @@ public class TripleProtocol extends AbstractProtocol {
|
|||
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
|
||||
optimizeSerialization(url);
|
||||
ExecutorService streamExecutor = getOrCreateStreamExecutor(
|
||||
url.getOrDefaultApplicationModel());
|
||||
url.getOrDefaultApplicationModel(), url);
|
||||
TripleInvoker<T> 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;
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue