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

增加redisRegistry consumer侧的服务探活逻辑 (#7929)

This commit is contained in:
Roshi 2021-06-10 23:45:49 +08:00 committed by GitHub
parent 149d1f68e5
commit ebd9b4bde2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 128 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
dubbo-registry/dubbo-registry-redis/src/main/java/org/apache/dubbo/registry/redis/RedisRegistry.java
View File

@ -93,6 +93,11 @@ public class RedisRegistry extends FailbackRegistry {
private volatile boolean admin = false;
private final Map<URL, Long> expireCache = new ConcurrentHashMap<>();
// just for unit test
private volatile boolean doExpire = true;
public RedisRegistry(URL url) {
super(url);
String type = url.getParameter(REDIS_CLIENT_KEY, MONO_REDIS);
@ -137,6 +142,15 @@ public class RedisRegistry extends FailbackRegistry {
}
}
}
if (doExpire) {
for (Map.Entry<URL, Long> expireEntry : expireCache.entrySet()) {
if (expireEntry.getValue() < System.currentTimeMillis()) {
doNotify(toCategoryPath(expireEntry.getKey()));
}
}
}
if (admin) {
clean();
}
@ -290,18 +304,28 @@ public class RedisRegistry extends FailbackRegistry {
continue;
}
List<URL> urls = new ArrayList<>();
Set<URL> toDeleteExpireKeys = new HashSet<>(expireCache.keySet());
Map<String, String> values = redisClient.hgetAll(key);
if (CollectionUtils.isNotEmptyMap(values)) {
for (Map.Entry<String, String> entry : values.entrySet()) {
URL u = URL.valueOf(entry.getKey());
long expire = Long.parseLong(entry.getValue());
if (!u.getParameter(DYNAMIC_KEY, true)
|| Long.parseLong(entry.getValue()) >= now) {
|| expire >= now) {
if (UrlUtils.isMatch(url, u)) {
urls.add(u);
expireCache.put(u, expire);
toDeleteExpireKeys.remove(u);
}
}
}
}
if (!toDeleteExpireKeys.isEmpty()) {
for (URL u : toDeleteExpireKeys) {
expireCache.remove(u);
}
}
if (urls.isEmpty()) {
urls.add(URLBuilder.from(url)
.setProtocol(EMPTY_PROTOCOL)
@ -311,6 +335,7 @@ public class RedisRegistry extends FailbackRegistry {
.build());
}
result.addAll(urls);
if (logger.isInfoEnabled()) {
logger.info("redis notify: " + key + " = " + urls);
}

104
dubbo-registry/dubbo-registry-redis/src/test/java/org/apache/dubbo/registry/redis/RedisRegistryTest.java
View File

@ -22,6 +22,7 @@ import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.Registry;
import org.apache.commons.lang3.SystemUtils;
import org.apache.dubbo.registry.support.AbstractRegistry;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@ -30,6 +31,8 @@ import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.embedded.RedisServer;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
@ -44,7 +47,9 @@ import static redis.embedded.RedisServer.newRedisServer;
public class RedisRegistryTest {
private static final String SERVICE = "org.apache.dubbo.test.injvmServie";
private static final URL SERVICE_URL = URL.valueOf("redis://redis/" + SERVICE + "?notify=false&methods=test1,test2");
private static final URL SERVICE_URL = URL.valueOf("redis://redis/" + SERVICE + "?notify=false&methods=test1,test2&category=providers,configurators,routers");
private static final URL PROVIDER_URL_A = URL.valueOf("redis://127.0.0.1:20880/" + SERVICE + "?notify=false&methods=test1,test2");
private static final URL PROVIDER_URL_B = URL.valueOf("redis://127.0.0.1:20881/" + SERVICE + "?notify=false&methods=test1,test2");
private RedisServer redisServer;
private RedisRegistry redisRegistry;
@ -74,7 +79,7 @@ public class RedisRegistryTest {
}
}
Assertions.assertNull(exception);
registryUrl = URL.valueOf("redis://localhost:" + redisPort);
registryUrl = URL.valueOf("redis://localhost:" + redisPort + "?session=4000");
redisRegistry = (RedisRegistry) new RedisRegistryFactory().createRegistry(registryUrl);
}
@ -105,6 +110,101 @@ public class RedisRegistryTest {
});
}
@Test
public void testSubscribeExpireCache() throws Exception {
redisRegistry.register(PROVIDER_URL_A);
redisRegistry.register(PROVIDER_URL_B);
NotifyListener listener = new NotifyListener() {
@Override
public void notify(List<URL> urls) {
}
};
redisRegistry.subscribe(SERVICE_URL, listener);
Field expireCache = RedisRegistry.class.getDeclaredField("expireCache");
expireCache.setAccessible(true);
Map<URL, Long> cacheExpire = (Map<URL, Long>)expireCache.get(redisRegistry);
assertThat(cacheExpire.get(PROVIDER_URL_A) > 0, is(true));
assertThat(cacheExpire.get(PROVIDER_URL_B) > 0, is(true));
redisRegistry.unregister(PROVIDER_URL_A);
boolean success = false;
for (int i = 0; i < 30; i++) {
cacheExpire = (Map<URL, Long>)expireCache.get(redisRegistry);
if (cacheExpire.get(PROVIDER_URL_A) == null) {
success = true;
break;
}
Thread.sleep(500);
}
assertThat(success, is(true));
}
@Test
public void testSubscribeWhenProviderCrash() throws Exception {
// unit test will fail if doExpire=false
// Field doExpireField = RedisRegistry.class.getDeclaredField("doExpire");
// doExpireField.setAccessible(true);
// doExpireField.set(redisRegistry, false);
redisRegistry.register(PROVIDER_URL_A);
redisRegistry.register(PROVIDER_URL_B);
assertThat(redisRegistry.getRegistered().contains(PROVIDER_URL_A), is(true));
assertThat(redisRegistry.getRegistered().contains(PROVIDER_URL_B), is(true));
Set<URL> notifiedUrls = new HashSet<>();
Object lock = new Object();
NotifyListener listener = new NotifyListener() {
@Override
public void notify(List<URL> urls) {
synchronized (lock) {
notifiedUrls.clear();
notifiedUrls.addAll(urls);
}
}
};
redisRegistry.subscribe(SERVICE_URL, listener);
assertThat(redisRegistry.getSubscribed().size(), is(1));
boolean firstOk = false;
boolean secondOk = false;
for (int i = 0; i < 30; i++) {
synchronized (lock) {
if (notifiedUrls.contains(PROVIDER_URL_A) && notifiedUrls.contains(PROVIDER_URL_B)) {
firstOk = true;
break;
}
}
Thread.sleep(500);
}
assertThat(firstOk, is(true));
// kill -9 to providerB
Field registeredField = AbstractRegistry.class.getDeclaredField("registered");
registeredField.setAccessible(true);
((Set<URL>) registeredField.get(redisRegistry)).remove(PROVIDER_URL_B);
for (int i = 0; i < 30; i++) {
synchronized (lock) {
if (notifiedUrls.contains(PROVIDER_URL_A) && notifiedUrls.size() == 1) {
secondOk = true;
break;
}
}
Thread.sleep(500);
}
assertThat(secondOk, is(true));
}
@Test
public void testSubscribeAndUnsubscribe() {