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

Store old revisions (#11241)

This commit is contained in:
Albumen Kevin 2023-01-06 16:36:59 +08:00 committed by GitHub
parent f16febe70d
commit 4a1aa856ab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 196 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -17,13 +17,13 @@
package org.apache.dubbo.common.constants;
import org.apache.dubbo.common.URL;
import java.net.NetworkInterface;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.regex.Pattern;
import org.apache.dubbo.common.URL;
public interface CommonConstants {
String DUBBO = "dubbo";
@ -361,6 +361,13 @@ public interface CommonConstants {
*/
String REGISTRY_LOCAL_FILE_CACHE_ENABLED = "file.cache";
String METADATA_INFO_CACHE_EXPIRE_KEY = "metadata-info-cache.expire";
int DEFAULT_METADATA_INFO_CACHE_EXPIRE = 10 * 60 * 1000;
String METADATA_INFO_CACHE_SIZE_KEY = "metadata-info-cache.size";
int DEFAULT_METADATA_INFO_CACHE_SIZE = 16;
/**
* The limit of callback service instances for one interface on every client

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

@ -19,6 +19,10 @@ package org.apache.dubbo.registry.client;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
@ -37,9 +41,14 @@ import org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils;
import org.apache.dubbo.registry.client.metadata.store.MetaCacheManager;
import org.apache.dubbo.rpc.model.ApplicationModel;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_INFO_CACHE_EXPIRE;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_INFO_CACHE_SIZE;
import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_METADATA_STORAGE_TYPE;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_EXPIRE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_SIZE_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_LOCAL_FILE_CACHE_ENABLED;
import static org.apache.dubbo.common.constants.CommonConstants.REMOTE_METADATA_STORAGE_TYPE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_FETCH_INSTANCE;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_LOAD_METADATA;
import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_CLUSTER_KEY;
@ -59,6 +68,8 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
protected final String serviceName;
protected volatile ServiceInstance serviceInstance;
protected volatile MetadataInfo metadataInfo;
protected final ConcurrentHashMap<String, MetadataInfoStat> metadataInfos = new ConcurrentHashMap<>();
protected final ScheduledFuture<?> refreshCacheFuture;
protected MetadataReport metadataReport;
protected String metadataType;
protected final MetaCacheManager metaCacheManager;
@ -88,6 +99,30 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
this.metaCacheManager = new MetaCacheManager(localCacheEnabled, getCacheNameSuffix(),
applicationModel.getFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getCacheRefreshingScheduledExecutor());
int metadataInfoCacheExpireTime = registryURL.getParameter(METADATA_INFO_CACHE_EXPIRE_KEY, DEFAULT_METADATA_INFO_CACHE_EXPIRE);
int metadataInfoCacheSize = registryURL.getParameter(METADATA_INFO_CACHE_SIZE_KEY, DEFAULT_METADATA_INFO_CACHE_SIZE);
this.refreshCacheFuture = applicationModel.getFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor()
.scheduleAtFixedRate(() -> {
try {
while (metadataInfos.size() > metadataInfoCacheSize) {
AtomicReference<String> oldestRevision = new AtomicReference<>();
AtomicReference<MetadataInfoStat> oldestStat = new AtomicReference<>();
metadataInfos.forEach((k, v) -> {
if (System.currentTimeMillis() - v.getUpdateTime() > metadataInfoCacheExpireTime &&
(oldestStat.get() == null || oldestStat.get().getUpdateTime() > v.getUpdateTime())) {
oldestRevision.set(k);
oldestStat.set(v);
}
});
if (oldestStat.get() != null) {
metadataInfos.remove(oldestRevision.get(), oldestStat.get());
}
}
} catch (Throwable t) {
logger.error(INTERNAL_ERROR, "", "", "Error occurred when clean up metadata info cache.", t);
}
}, metadataInfoCacheExpireTime / 2, metadataInfoCacheExpireTime / 2, TimeUnit.MILLISECONDS);
}
@ -163,6 +198,16 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
return this.metadataInfo;
}
@Override
public MetadataInfo getLocalMetadata(String revision) {
MetadataInfoStat metadataInfoStat = metadataInfos.get(revision);
if (metadataInfoStat != null) {
return metadataInfoStat.getMetadataInfo();
} else {
return null;
}
}
@Override
public MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances) {
MetadataInfo metadata = metaCacheManager.get(revision);
@ -217,6 +262,7 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
public final void destroy() throws Exception {
isDestroy = true;
metaCacheManager.destroy();
refreshCacheFuture.cancel(true);
doDestroy();
}
@ -291,12 +337,17 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
}
protected void reportMetadata(MetadataInfo metadataInfo) {
if (metadataInfo == null) {
return;
}
if (metadataReport != null) {
SubscriberMetadataIdentifier identifier = new SubscriberMetadataIdentifier(serviceName, metadataInfo.getRevision());
if ((DEFAULT_METADATA_STORAGE_TYPE.equals(metadataType) && metadataReport.shouldReportMetadata()) || REMOTE_METADATA_STORAGE_TYPE.equals(metadataType)) {
metadataReport.publishAppMetadata(identifier, metadataInfo);
}
}
MetadataInfo clonedMetadataInfo = metadataInfo.clone();
metadataInfos.put(metadataInfo.getRevision(), new MetadataInfoStat(clonedMetadataInfo));
}
protected void unReportMetadata(MetadataInfo metadataInfo) {
@ -328,4 +379,21 @@ public abstract class AbstractServiceDiscovery implements ServiceDiscovery {
}
return stringBuilder.toString();
}
private static class MetadataInfoStat {
private final MetadataInfo metadataInfo;
private final long updateTime = System.currentTimeMillis();
public MetadataInfoStat(MetadataInfo metadataInfo) {
this.metadataInfo = metadataInfo;
}
public MetadataInfo getMetadataInfo() {
return metadataInfo;
}
public long getUpdateTime() {
return updateTime;
}
}
}

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

@ -16,15 +16,15 @@
*/
package org.apache.dubbo.registry.client;
import java.util.List;
import java.util.Set;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.lang.Prioritized;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.RegistryService;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
import java.util.List;
import java.util.Set;
import static org.apache.dubbo.common.constants.CommonConstants.REGISTRY_DELAY_NOTIFICATION_KEY;
/**
@ -69,6 +69,10 @@ public interface ServiceDiscovery extends RegistryService, Prioritized {
MetadataInfo getLocalMetadata();
default MetadataInfo getLocalMetadata(String revision) {
return getLocalMetadata();
}
MetadataInfo getRemoteMetadata(String revision);
MetadataInfo getRemoteMetadata(String revision, List<ServiceInstance> instances);

28
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/MetadataServiceDelegation.java
View File

@ -16,18 +16,6 @@
*/
package org.apache.dubbo.registry.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.InstanceMetadataChangedListener;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.support.RegistryManager;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
@ -40,6 +28,18 @@ import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.resource.Disposable;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metadata.InstanceMetadataChangedListener;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.support.RegistryManager;
import org.apache.dubbo.rpc.model.ApplicationModel;
import static java.util.Collections.emptySortedSet;
import static java.util.Collections.unmodifiableSortedSet;
import static org.apache.dubbo.common.URL.buildKey;
@ -182,8 +182,8 @@ public class MetadataServiceDelegation implements MetadataService, Disposable {
}
for (ServiceDiscovery sd : registryManager.getServiceDiscoveries()) {
MetadataInfo metadataInfo = sd.getLocalMetadata();
if (revision.equals(metadataInfo.getRevision())) {
MetadataInfo metadataInfo = sd.getLocalMetadata(revision);
if (metadataInfo != null && revision.equals(metadataInfo.getRevision())) {
return metadataInfo;
}
}

85
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/ServiceDiscoveryCacheTest.java Normal file
View File

@ -0,0 +1,85 @@
/*
* 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;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.metadata.MetadataInfo;
import org.apache.dubbo.registry.client.support.MockServiceDiscovery;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_INFO_CACHE_EXPIRE_KEY;
import static org.awaitility.Awaitility.await;
class ServiceDiscoveryCacheTest {
@Test
void test() throws InterruptedException {
ApplicationModel applicationModel = FrameworkModel.defaultModel().newApplication();
applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig("Test"));
URL registryUrl = URL.valueOf("mock://127.0.0.1:12345").addParameter(METADATA_INFO_CACHE_EXPIRE_KEY, 10);
MockServiceDiscovery mockServiceDiscovery = Mockito.spy(new MockServiceDiscovery(applicationModel, registryUrl));
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService"));
mockServiceDiscovery.register();
ServiceInstance localInstance = mockServiceDiscovery.getLocalInstance();
Assertions.assertEquals(localInstance.getServiceMetadata(), mockServiceDiscovery.getLocalMetadata(localInstance.getServiceMetadata().getRevision()));
List<MetadataInfo> instances = new LinkedList<>();
instances.add(localInstance.getServiceMetadata().clone());
for (int i = 0; i < 15; i++) {
Thread.sleep(1);
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService" + i));
mockServiceDiscovery.update();
instances.add(mockServiceDiscovery.getLocalInstance().getServiceMetadata().clone());
}
for (MetadataInfo instance : instances) {
Assertions.assertEquals(instance, mockServiceDiscovery.getLocalMetadata(instance.getRevision()));
}
for (int i = 0; i < 5; i++) {
Thread.sleep(1);
mockServiceDiscovery.register(URL.valueOf("mock://127.0.0.1:12345")
.setServiceInterface("org.apache.dubbo.registry.service.DemoService-new" + i));
mockServiceDiscovery.update();
instances.add(mockServiceDiscovery.getLocalInstance().getServiceMetadata().clone());
}
await().until(() -> Objects.isNull(mockServiceDiscovery.getLocalMetadata(instances.get(4).getRevision())));
for (int i = 0; i < 5; i++) {
Assertions.assertNull(mockServiceDiscovery.getLocalMetadata(instances.get(i).getRevision()));
}
applicationModel.destroy();
}
}

5
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/support/MockServiceDiscovery.java
View File

@ -40,11 +40,6 @@ public class MockServiceDiscovery extends AbstractServiceDiscovery {
}
@Override
public void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance) throws RuntimeException {
}
@Override
public void doUnregister(ServiceInstance serviceInstance) {

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

@ -16,6 +16,14 @@
*/
package org.apache.dubbo.registry.multiple;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.metadata.MetadataInfo;
@ -26,14 +34,6 @@ import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
public class MultipleServiceDiscovery implements ServiceDiscovery {
public static final String REGISTRY_PREFIX_KEY = "child.";
private static final String REGISTRY_TYPE = "registry-type";
@ -137,6 +137,11 @@ public class MultipleServiceDiscovery implements ServiceDiscovery {
throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method.");
}
@Override
public MetadataInfo getLocalMetadata(String revision) {
throw new UnsupportedOperationException("Multiple registry implementation does not support getLocalMetadata() method.");
}
@Override
public MetadataInfo getRemoteMetadata(String revision) {
throw new UnsupportedOperationException("Multiple registry implementation does not support getMetadata() method.");