dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java - dubbo - Git at Google

 /*
  * Copyright 1999-2011 Alibaba Group.
  *
  * Licensed 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 com.alibaba.dubbo.registry.support;

 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;

 import com.alibaba.dubbo.common.Constants;
 import com.alibaba.dubbo.common.URL;
 import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
 import com.alibaba.dubbo.common.utils.NamedThreadFactory;
 import com.alibaba.dubbo.registry.NotifyListener;

 /**
  * FailbackRegistry
  *
  * @author william.liangf
  */
 public abstract class FailbackRegistry extends AbstractRegistry {

     // 定时任务执行器
     private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryFailedRetryTimer", true));

     // 失败重试定时器，定时检查是否有请求失败，如有，无限次重试
     private final ScheduledFuture<?> retryFuture;

     private final Set<String> failedRegistered = new ConcurrentHashSet<String>();

     private final Set<String> failedUnregistered = new ConcurrentHashSet<String>();

     private final ConcurrentMap<String, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<String, Set<NotifyListener>>();

     private final ConcurrentMap<String, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<String, Set<NotifyListener>>();

     private final ConcurrentMap<String, Map<NotifyListener, List<URL>>> failedNotified = new ConcurrentHashMap<String, Map<NotifyListener, List<URL>>>();

     public FailbackRegistry(URL url) {
         super(url);
         int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
         this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
             public void run() {
                 // 检测并连接注册中心
                 try {
                     retry();
                 } catch (Throwable t) { // 防御性容错
                     logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
                 }
             }
         }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
     }

     // 重试失败的动作
     private void retry() throws Exception {
         if (! failedRegistered.isEmpty()) {
             Set<String> failed = new HashSet<String>(failedRegistered);
             if (failed.size() > 0) {
                 if (logger.isInfoEnabled()) {
                     logger.info("Retry register " + failed);
                 }
                 try {
                     for (String url : failed) {
                         try {
                             doRegister(URL.valueOf(url));
                             failedRegistered.remove(url);
                         } catch (Throwable t) { // 忽略所有异常，等待下次重试
                             logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                         }
                     }
                 } catch (Throwable t) { // 忽略所有异常，等待下次重试
                     logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                 }
             }
         }
         if(! failedUnregistered.isEmpty()) {
             Set<String> failed = new HashSet<String>(failedUnregistered);
             if (failed.size() > 0) {
                 if (logger.isInfoEnabled()) {
                     logger.info("Retry unregister " + failed);
                 }
                 try {
                     for (String url : failed) {
                         try {
                             doUnregister(URL.valueOf(url));
                             failedUnregistered.remove(url);
                         } catch (Throwable t) { // 忽略所有异常，等待下次重试
                             logger.warn("Failed to retry unregister  " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                         }
                     }
                 } catch (Throwable t) { // 忽略所有异常，等待下次重试
                     logger.warn("Failed to retry unregister  " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                 }
             }
         }
         if (! failedSubscribed.isEmpty()) {
             Map<String, Set<NotifyListener>> failed = new HashMap<String, Set<NotifyListener>>(failedSubscribed);
             for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
                 if (entry.getValue() == null || entry.getValue().size() == 0) {
                     failed.remove(entry.getKey());
                 }
             }
             if (failed.size() > 0) {
                 if (logger.isInfoEnabled()) {
                     logger.info("Retry subscribe " + failed);
                 }
                 try {
                     for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
                         URL url = URL.valueOf(entry.getKey());
                         Set<NotifyListener> listeners = entry.getValue();
                         for (NotifyListener listener : listeners) {
                             try {
                                 doSubscribe(url, listener);
                                 listeners.remove(listener);
                             } catch (Throwable t) { // 忽略所有异常，等待下次重试
                                 logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                             }
                         }
                     }
                 } catch (Throwable t) { // 忽略所有异常，等待下次重试
                     logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                 }
             }
         }
         if (! failedUnsubscribed.isEmpty()) {
             Map<String, Set<NotifyListener>> failed = new HashMap<String, Set<NotifyListener>>(failedUnsubscribed);
             for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
                 if (entry.getValue() == null || entry.getValue().size() == 0) {
                     failed.remove(entry.getKey());
                 }
             }
             if (failed.size() > 0) {
                 if (logger.isInfoEnabled()) {
                     logger.info("Retry unsubscribe " + failed);
                 }
                 try {
                     for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
                         URL url = URL.valueOf(entry.getKey());
                         Set<NotifyListener> listeners = entry.getValue();
                         for (NotifyListener listener : listeners) {
                             try {
                                 doUnsubscribe(url, listener);
                                 listeners.remove(listener);
                             } catch (Throwable t) { // 忽略所有异常，等待下次重试
                                 logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                             }
                         }
                     }
                 } catch (Throwable t) { // 忽略所有异常，等待下次重试
                     logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                 }
             }
         }
         if (! failedNotified.isEmpty()) {
             Map<String, Map<NotifyListener, List<URL>>> failed = new HashMap<String, Map<NotifyListener, List<URL>>>(failedNotified);
             for (Map.Entry<String, Map<NotifyListener, List<URL>>> entry : failed.entrySet()) {
                 if (entry.getValue() == null || entry.getValue().size() == 0) {
                     failed.remove(entry.getKey());
                 }
             }
             if (failed.size() > 0) {
                 if (logger.isInfoEnabled()) {
                     logger.info("Retry notify " + failed);
                 }
                 try {
                     for (Map<NotifyListener, List<URL>> values : failed.values()) {
                         for (Map.Entry<NotifyListener, List<URL>> entry : values.entrySet()) {
                             try {
                                 NotifyListener listener = entry.getKey();
                                 List<URL> urls = entry.getValue();
                                 listener.notify(urls);
                                 values.remove(listener);
                             } catch (Throwable t) { // 忽略所有异常，等待下次重试
                                 logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                             }
                         }
                     }
                 } catch (Throwable t) { // 忽略所有异常，等待下次重试
                     logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                 }
             }
         }
         doRetry();
     }

     public void destroy() {
         super.destroy();
         try {
             retryFuture.cancel(true);
         } catch (Throwable t) {
             logger.warn(t.getMessage(), t);
         }
     }

     public void register(URL url, NotifyListener listener) {
         super.register(url, listener);
         try {
             // 向服务器端发送注册请求
             doRegister(url);
             removeFailedRegistered(url);
         } catch (Exception t) {
             if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
                 throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
             }
             // 否则，将失败的注册请求记录到失败列表，定时重试
             failedRegistered.add(url.toFullString());
             logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
         }
     }

     public void unregister(URL url, NotifyListener listener) {
         super.unregister(url, listener);
         try {
             // 向服务器端发送取消注册请求
             doUnregister(url);
             removeFailedRegistered(url);
         } catch (Exception t) {
             if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
                 throw new IllegalStateException("Failed to unregister " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
             }
             // 否则，将失败的取消注册请求记录到失败列表，定时重试
             failedUnregistered.add(url.toFullString());
             logger.error("Failed to uregister " + url + ", waiting for retry, cause: " + t.getMessage(), t);
         }
     }

     private void removeFailedRegistered(URL url) {
         String key = url.toFullString();
         failedRegistered.remove(key);
         failedUnregistered.remove(key);
     }

     private void removeFailedSubscribed(URL url, NotifyListener listener) {
         String key = url.toFullString();
         Set<NotifyListener> listeners = failedSubscribed.get(key);
         if (listeners != null) {
             listeners.remove(listener);
         }
         listeners = failedUnsubscribed.get(key);
         if (listeners != null) {
             listeners.remove(listener);
         }
         Map<NotifyListener, List<URL>> notified = failedNotified.get(key);
         if (notified != null) {
             notified.remove(listener);
         }
     }

     public void subscribe(URL url, NotifyListener listener) {
         super.subscribe(url, listener);
         try {
             // 向服务器端发送订阅请求
             doSubscribe(url, listener);
             removeFailedSubscribed(url, listener);
         } catch (Exception t) {
             if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
                 throw new IllegalStateException("Failed to subscribe " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
             }
             // 否则，将失败的订阅请求记录到失败列表，定时重试
             String key = url.toFullString();
             Set<NotifyListener> listeners = failedSubscribed.get(key);
             if (listeners == null) {
                 failedSubscribed.putIfAbsent(key, new ConcurrentHashSet<NotifyListener>());
                 listeners = failedSubscribed.get(key);
             }
             listeners.add(listener);
             logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
         }
     }

     public void unsubscribe(URL url, NotifyListener listener) {
         super.unsubscribe(url, listener);
         try {
             // 向服务器端发送取消订阅请求
             doUnsubscribe(url, listener);
             removeFailedSubscribed(url, listener);
         } catch (Exception t) {
             if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
                 throw new IllegalStateException("Failed to unsubscribe " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
             }
             // 否则，将失败的取消订阅请求记录到失败列表，定时重试
             String key = url.toFullString();
             Set<NotifyListener> listeners = failedUnsubscribed.get(key);
             if (listeners == null) {
                 failedUnsubscribed.putIfAbsent(key, new ConcurrentHashSet<NotifyListener>());
                 listeners = failedUnsubscribed.get(key);
             }
             listeners.add(listener);
             logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
         }
     }

     protected void notify(URL url, NotifyListener listener, List<URL> urls) {
         if (url == null) {
             throw new IllegalArgumentException("notify url == null");
         }
         if (listener == null) {
             throw new IllegalArgumentException("notify listener == null");
         }
         if ((urls == null || urls.size() == 0)
                 && ! url.getParameter(Constants.ADMIN_KEY, false)) {
             return;
         }
         try {
             listener.notify(urls);
         } catch (Exception t) {
             // 将失败的通知请求记录到失败列表，定时重试
             String key = url.toFullString();
             Map<NotifyListener, List<URL>> values = failedNotified.get(key);
             if (values == null) {
                 failedNotified.putIfAbsent(key, new ConcurrentHashMap<NotifyListener, List<URL>>());
                 values = failedNotified.get(key);
             }
             values.put(listener, urls);
             logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
         }
     }

     protected abstract void doRegister(URL url);

     protected abstract void doUnregister(URL url);

     protected abstract void doSubscribe(URL url, NotifyListener listener);

     protected abstract void doUnsubscribe(URL url, NotifyListener listener);

     protected void doRetry() {}

     public Set<String> getFailedRegistered() {
         return failedRegistered;
     }

     public Set<String> getFailedUnregistered() {
         return failedUnregistered;
     }

     public Map<String, Set<NotifyListener>> getFailedSubscribed() {
         return failedSubscribed;
     }

     public Map<String, Set<NotifyListener>> getFailedUnsubscribed() {
         return failedUnsubscribed;
     }

     public Map<String, Map<NotifyListener, List<URL>>> getFailedNotified() {
         return failedNotified;
     }

 }
	/*
	* Copyright 1999-2011 Alibaba Group.
	*
	* Licensed 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 com.alibaba.dubbo.registry.support;

	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;
	import java.util.concurrent.Executors;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.ScheduledFuture;
	import java.util.concurrent.TimeUnit;

	import com.alibaba.dubbo.common.Constants;
	import com.alibaba.dubbo.common.URL;
	import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
	import com.alibaba.dubbo.common.utils.NamedThreadFactory;
	import com.alibaba.dubbo.registry.NotifyListener;

	/**
	* FailbackRegistry
	*
	* @author william.liangf
	*/
	public abstract class FailbackRegistry extends AbstractRegistry {

	// 定时任务执行器
	private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryFailedRetryTimer", true));

	// 失败重试定时器，定时检查是否有请求失败，如有，无限次重试
	private final ScheduledFuture<?> retryFuture;

	private final Set<String> failedRegistered = new ConcurrentHashSet<String>();

	private final Set<String> failedUnregistered = new ConcurrentHashSet<String>();

	private final ConcurrentMap<String, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<String, Set<NotifyListener>>();

	private final ConcurrentMap<String, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<String, Set<NotifyListener>>();

	private final ConcurrentMap<String, Map<NotifyListener, List<URL>>> failedNotified = new ConcurrentHashMap<String, Map<NotifyListener, List<URL>>>();

	public FailbackRegistry(URL url) {
	super(url);
	int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
	this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
	public void run() {
	// 检测并连接注册中心
	try {
	retry();
	} catch (Throwable t) { // 防御性容错
	logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
	}
	}
	}, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
	}

	// 重试失败的动作
	private void retry() throws Exception {
	if (! failedRegistered.isEmpty()) {
	Set<String> failed = new HashSet<String>(failedRegistered);
	if (failed.size() > 0) {
	if (logger.isInfoEnabled()) {
	logger.info("Retry register " + failed);
	}
	try {
	for (String url : failed) {
	try {
	doRegister(URL.valueOf(url));
	failedRegistered.remove(url);
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	if(! failedUnregistered.isEmpty()) {
	Set<String> failed = new HashSet<String>(failedUnregistered);
	if (failed.size() > 0) {
	if (logger.isInfoEnabled()) {
	logger.info("Retry unregister " + failed);
	}
	try {
	for (String url : failed) {
	try {
	doUnregister(URL.valueOf(url));
	failedUnregistered.remove(url);
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry unregister " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry unregister " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	if (! failedSubscribed.isEmpty()) {
	Map<String, Set<NotifyListener>> failed = new HashMap<String, Set<NotifyListener>>(failedSubscribed);
	for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
	if (entry.getValue() == null \|\| entry.getValue().size() == 0) {
	failed.remove(entry.getKey());
	}
	}
	if (failed.size() > 0) {
	if (logger.isInfoEnabled()) {
	logger.info("Retry subscribe " + failed);
	}
	try {
	for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
	URL url = URL.valueOf(entry.getKey());
	Set<NotifyListener> listeners = entry.getValue();
	for (NotifyListener listener : listeners) {
	try {
	doSubscribe(url, listener);
	listeners.remove(listener);
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	if (! failedUnsubscribed.isEmpty()) {
	Map<String, Set<NotifyListener>> failed = new HashMap<String, Set<NotifyListener>>(failedUnsubscribed);
	for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
	if (entry.getValue() == null \|\| entry.getValue().size() == 0) {
	failed.remove(entry.getKey());
	}
	}
	if (failed.size() > 0) {
	if (logger.isInfoEnabled()) {
	logger.info("Retry unsubscribe " + failed);
	}
	try {
	for (Map.Entry<String, Set<NotifyListener>> entry : failed.entrySet()) {
	URL url = URL.valueOf(entry.getKey());
	Set<NotifyListener> listeners = entry.getValue();
	for (NotifyListener listener : listeners) {
	try {
	doUnsubscribe(url, listener);
	listeners.remove(listener);
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	if (! failedNotified.isEmpty()) {
	Map<String, Map<NotifyListener, List<URL>>> failed = new HashMap<String, Map<NotifyListener, List<URL>>>(failedNotified);
	for (Map.Entry<String, Map<NotifyListener, List<URL>>> entry : failed.entrySet()) {
	if (entry.getValue() == null \|\| entry.getValue().size() == 0) {
	failed.remove(entry.getKey());
	}
	}
	if (failed.size() > 0) {
	if (logger.isInfoEnabled()) {
	logger.info("Retry notify " + failed);
	}
	try {
	for (Map<NotifyListener, List<URL>> values : failed.values()) {
	for (Map.Entry<NotifyListener, List<URL>> entry : values.entrySet()) {
	try {
	NotifyListener listener = entry.getKey();
	List<URL> urls = entry.getValue();
	listener.notify(urls);
	values.remove(listener);
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	} catch (Throwable t) { // 忽略所有异常，等待下次重试
	logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
	}
	}
	}
	doRetry();
	}

	public void destroy() {
	super.destroy();
	try {
	retryFuture.cancel(true);
	} catch (Throwable t) {
	logger.warn(t.getMessage(), t);
	}
	}

	public void register(URL url, NotifyListener listener) {
	super.register(url, listener);
	try {
	// 向服务器端发送注册请求
	doRegister(url);
	removeFailedRegistered(url);
	} catch (Exception t) {
	if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
	throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
	}
	// 否则，将失败的注册请求记录到失败列表，定时重试
	failedRegistered.add(url.toFullString());
	logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
	}
	}

	public void unregister(URL url, NotifyListener listener) {
	super.unregister(url, listener);
	try {
	// 向服务器端发送取消注册请求
	doUnregister(url);
	removeFailedRegistered(url);
	} catch (Exception t) {
	if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
	throw new IllegalStateException("Failed to unregister " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
	}
	// 否则，将失败的取消注册请求记录到失败列表，定时重试
	failedUnregistered.add(url.toFullString());
	logger.error("Failed to uregister " + url + ", waiting for retry, cause: " + t.getMessage(), t);
	}
	}

	private void removeFailedRegistered(URL url) {
	String key = url.toFullString();
	failedRegistered.remove(key);
	failedUnregistered.remove(key);
	}

	private void removeFailedSubscribed(URL url, NotifyListener listener) {
	String key = url.toFullString();
	Set<NotifyListener> listeners = failedSubscribed.get(key);
	if (listeners != null) {
	listeners.remove(listener);
	}
	listeners = failedUnsubscribed.get(key);
	if (listeners != null) {
	listeners.remove(listener);
	}
	Map<NotifyListener, List<URL>> notified = failedNotified.get(key);
	if (notified != null) {
	notified.remove(listener);
	}
	}

	public void subscribe(URL url, NotifyListener listener) {
	super.subscribe(url, listener);
	try {
	// 向服务器端发送订阅请求
	doSubscribe(url, listener);
	removeFailedSubscribed(url, listener);
	} catch (Exception t) {
	if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
	throw new IllegalStateException("Failed to subscribe " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
	}
	// 否则，将失败的订阅请求记录到失败列表，定时重试
	String key = url.toFullString();
	Set<NotifyListener> listeners = failedSubscribed.get(key);
	if (listeners == null) {
	failedSubscribed.putIfAbsent(key, new ConcurrentHashSet<NotifyListener>());
	listeners = failedSubscribed.get(key);
	}
	listeners.add(listener);
	logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
	}
	}

	public void unsubscribe(URL url, NotifyListener listener) {
	super.unsubscribe(url, listener);
	try {
	// 向服务器端发送取消订阅请求
	doUnsubscribe(url, listener);
	removeFailedSubscribed(url, listener);
	} catch (Exception t) {
	if (getUrl().getParameter(Constants.CHECK_KEY, true)) { // 如果开启了启动时检测，则直接抛出异常
	throw new IllegalStateException("Failed to unsubscribe " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
	}
	// 否则，将失败的取消订阅请求记录到失败列表，定时重试
	String key = url.toFullString();
	Set<NotifyListener> listeners = failedUnsubscribed.get(key);
	if (listeners == null) {
	failedUnsubscribed.putIfAbsent(key, new ConcurrentHashSet<NotifyListener>());
	listeners = failedUnsubscribed.get(key);
	}
	listeners.add(listener);
	logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
	}
	}

	protected void notify(URL url, NotifyListener listener, List<URL> urls) {
	if (url == null) {
	throw new IllegalArgumentException("notify url == null");
	}
	if (listener == null) {
	throw new IllegalArgumentException("notify listener == null");
	}
	if ((urls == null \|\| urls.size() == 0)
	&& ! url.getParameter(Constants.ADMIN_KEY, false)) {
	return;
	}
	try {
	listener.notify(urls);
	} catch (Exception t) {
	// 将失败的通知请求记录到失败列表，定时重试
	String key = url.toFullString();
	Map<NotifyListener, List<URL>> values = failedNotified.get(key);
	if (values == null) {
	failedNotified.putIfAbsent(key, new ConcurrentHashMap<NotifyListener, List<URL>>());
	values = failedNotified.get(key);
	}
	values.put(listener, urls);
	logger.error("Failed to unsubscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
	}
	}

	protected abstract void doRegister(URL url);

	protected abstract void doUnregister(URL url);

	protected abstract void doSubscribe(URL url, NotifyListener listener);

	protected abstract void doUnsubscribe(URL url, NotifyListener listener);

	protected void doRetry() {}

	public Set<String> getFailedRegistered() {
	return failedRegistered;
	}

	public Set<String> getFailedUnregistered() {
	return failedUnregistered;
	}

	public Map<String, Set<NotifyListener>> getFailedSubscribed() {
	return failedSubscribed;
	}

	public Map<String, Set<NotifyListener>> getFailedUnsubscribed() {
	return failedUnsubscribed;
	}

	public Map<String, Map<NotifyListener, List<URL>>> getFailedNotified() {
	return failedNotified;
	}

	}