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

remove notification retry task (#6401) 

fixes #5961
This commit is contained in:
ken.lj 2020-08-28 16:00:35 +08:00 committed by GitHub
parent 859c9ba47d
commit dc297d0439
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 2 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-compatible/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java
View File

@ -51,10 +51,6 @@ public abstract class FailbackRegistry implements org.apache.dubbo.registry.Regi
failbackRegistry.removeFailedUnsubscribedTask(url.getOriginalURL(), new NotifyListener.ReverseCompatibleNotifyListener(listener));
}
public void removeFailedNotifiedTask(URL url, NotifyListener listener) {
failbackRegistry.removeFailedNotifiedTask(url.getOriginalURL(), new NotifyListener.ReverseCompatibleNotifyListener(listener));
}
@Override
public void register(URL url) {
failbackRegistry.register(url.getOriginalURL());

67
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/retry/FailedNotifiedTask.java
View File

@ -1,67 +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.retry;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.timer.Timeout;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.support.FailbackRegistry;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
/**
* FailedNotifiedTask
*/
public final class FailedNotifiedTask extends AbstractRetryTask {
private static final String NAME = "retry notify";
private final NotifyListener listener;
private final List<URL> urls = new CopyOnWriteArrayList<>();
public FailedNotifiedTask(URL url, NotifyListener listener) {
super(url, null, NAME);
if (listener == null) {
throw new IllegalArgumentException();
}
this.listener = listener;
}
public void addUrlToRetry(List<URL> urls) {
if (CollectionUtils.isEmpty(urls)) {
return;
}
this.urls.addAll(urls);
}
public void removeRetryUrl(List<URL> urls) {
this.urls.removeAll(urls);
}
@Override
protected void doRetry(URL url, FailbackRegistry registry, Timeout timeout) {
if (CollectionUtils.isNotEmpty(urls)) {
listener.notify(urls);
urls.clear();
}
reput(timeout, retryPeriod);
}
}

39
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/FailbackRegistry.java
View File

@ -21,7 +21,6 @@ import org.apache.dubbo.common.timer.HashedWheelTimer;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.retry.FailedNotifiedTask;
import org.apache.dubbo.registry.retry.FailedRegisteredTask;
import org.apache.dubbo.registry.retry.FailedSubscribedTask;
import org.apache.dubbo.registry.retry.FailedUnregisteredTask;
@ -57,8 +56,6 @@ public abstract class FailbackRegistry extends AbstractRegistry {
private final ConcurrentMap<Holder, FailedUnsubscribedTask> failedUnsubscribed = new ConcurrentHashMap<Holder, FailedUnsubscribedTask>();
private final ConcurrentMap<Holder, FailedNotifiedTask> failedNotified = new ConcurrentHashMap<Holder, FailedNotifiedTask>();
/**
* The time in milliseconds the retryExecutor will wait
*/
@ -93,11 +90,6 @@ public abstract class FailbackRegistry extends AbstractRegistry {
failedUnsubscribed.remove(h);
}
public void removeFailedNotifiedTask(URL url, NotifyListener listener) {
Holder h = new Holder(url, listener);
failedNotified.remove(h);
}
private void addFailedRegistered(URL url) {
FailedRegisteredTask oldOne = failedRegistered.get(url);
if (oldOne != null) {
@ -159,7 +151,6 @@ public abstract class FailbackRegistry extends AbstractRegistry {
f.cancel();
}
removeFailedUnsubscribed(url, listener);
removeFailedNotified(url, listener);
}
private void addFailedUnsubscribed(URL url, NotifyListener listener) {
@ -184,28 +175,6 @@ public abstract class FailbackRegistry extends AbstractRegistry {
}
}
private void addFailedNotified(URL url, NotifyListener listener, List<URL> urls) {
Holder h = new Holder(url, listener);
FailedNotifiedTask newTask = new FailedNotifiedTask(url, listener);
FailedNotifiedTask f = failedNotified.putIfAbsent(h, newTask);
if (f == null) {
// never has a retry task. then start a new task for retry.
newTask.addUrlToRetry(urls);
retryTimer.newTimeout(newTask, retryPeriod, TimeUnit.MILLISECONDS);
} else {
// just add urls which needs retry.
newTask.addUrlToRetry(urls);
}
}
private void removeFailedNotified(URL url, NotifyListener listener) {
Holder h = new Holder(url, listener);
FailedNotifiedTask f = failedNotified.remove(h);
if (f != null) {
f.cancel();
}
}
ConcurrentMap<URL, FailedRegisteredTask> getFailedRegistered() {
return failedRegistered;
}
@ -222,9 +191,6 @@ public abstract class FailbackRegistry extends AbstractRegistry {
return failedUnsubscribed;
}
ConcurrentMap<Holder, FailedNotifiedTask> getFailedNotified() {
return failedNotified;
}
@Override
public void register(URL url) {
@ -397,9 +363,8 @@ public abstract class FailbackRegistry extends AbstractRegistry {
try {
doNotify(url, listener, urls);
} catch (Exception t) {
// Record a failed registration request to a failed list, retry regularly
addFailedNotified(url, listener, urls);
logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
// Record a failed registration request to a failed list
logger.error("Failed to notify addresses for subscribe " + url + ", cause: " + t.getMessage(), t);
}
}

31
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/support/FailbackRegistryTest.java
View File

@ -27,7 +27,6 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.dubbo.registry.Constants.CONSUMER_PROTOCOL;
@ -153,36 +152,6 @@ public class FailbackRegistryTest {
assertEquals(true, notified.get());
}
@Test
public void testDoRetry_nofify() throws Exception {
//Initial value 0
final AtomicInteger count = new AtomicInteger(0);
NotifyListener listner = new NotifyListener() {
@Override
public void notify(List<URL> urls) {
count.incrementAndGet();
//The exception is thrown for the first time to see if the back will be called again to incrementAndGet
if (count.get() == 1L) {
throw new RuntimeException("test exception please ignore");
}
}
};
registry = new MockRegistry(registryUrl, new CountDownLatch(0));
registry.subscribe(serviceUrl.setProtocol(CONSUMER_PROTOCOL).addParameters(CollectionUtils.toStringMap("check", "false")), listner);
assertEquals(1, count.get()); //Make sure that the subscribe call has just been called once count.incrementAndGet after the call is completed
//Wait for the timer.
for (int i = 0; i < trytimes; i++) {
System.out.println("failback notify retry ,times:" + i);
if (count.get() == 2)
break;
Thread.sleep(sleeptime);
}
assertEquals(2, count.get());
}
@Test
public void testRecover() throws Exception {
CountDownLatch countDownLatch = new CountDownLatch(4);