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

DUBBO-432 修复SimpleRegistry不可用问题

This commit is contained in:
liangfei0201 2012-06-29 18:41:16 +08:00
parent d62835c64d
commit 4fd13c321a
8 changed files with 114 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
dubbo-container/dubbo-container-spring/src/main/java/com/alibaba/dubbo/container/spring/SpringContainer.java
View File

@ -17,6 +17,7 @@ package com.alibaba.dubbo.container.spring;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.ConfigUtils;
@ -35,9 +36,13 @@ public class SpringContainer implements Container {
public static final String DEFAULT_SPRING_CONFIG = "classpath*:META-INF/spring/*.xml";
ClassPathXmlApplicationContext context;
static ClassPathXmlApplicationContext context;
public void start() {
public static ClassPathXmlApplicationContext getContext() {
return context;
}
public void start() {
String configPath = ConfigUtils.getProperty(SPRING_CONFIG);
if (configPath == null || configPath.length() == 0) {
configPath = DEFAULT_SPRING_CONFIG;

2
dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryDirectory.java
View File

@ -114,7 +114,7 @@ public class RegistryDirectory<T> extends AbstractDirectory<T> implements Notify
this.serviceType = serviceType;
this.serviceKey = url.getServiceKey();
this.queryMap = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
this.overrideDirectoryUrl = this.directoryUrl = url.clearParameters().addParameters(queryMap).removeParameter(Constants.MONITOR_KEY);
this.overrideDirectoryUrl = this.directoryUrl = url.setPath(url.getServiceInterface()).clearParameters().addParameters(queryMap).removeParameter(Constants.MONITOR_KEY);
String group = directoryUrl.getParameter( Constants.GROUP_KEY, "" );
this.multiGroup = group != null && ("*".equals(group) || group.contains( "," ));
String methods = queryMap.get(Constants.METHODS_KEY);

14
dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java
View File

@ -30,9 +30,11 @@ import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.registry.NotifyListener;
import com.alibaba.dubbo.registry.Registry;
import com.alibaba.dubbo.registry.RegistryFactory;
import com.alibaba.dubbo.registry.RegistryService;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Protocol;
import com.alibaba.dubbo.rpc.ProxyFactory;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.cluster.Cluster;
import com.alibaba.dubbo.rpc.protocol.InvokerWrapper;
@ -63,6 +65,12 @@ public class RegistryProtocol implements Protocol {
this.registryFactory = registryFactory;
}
private ProxyFactory proxyFactory;
public void setProxyFactory(ProxyFactory proxyFactory) {
this.proxyFactory = proxyFactory;
}
public int getDefaultPort() {
return 9090;
}
@ -210,9 +218,13 @@ public class RegistryProtocol implements Protocol {
return key;
}
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
@SuppressWarnings("unchecked")
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));

2
dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java
View File

@ -261,7 +261,7 @@ public abstract class AbstractRegistry implements Registry {
}
return null;
}
public List<URL> lookup(URL url) {
List<URL> result = new ArrayList<URL>();
Map<String, List<URL>> notifiedUrls = getNotified().get(url);

6
dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java
View File

@ -220,7 +220,7 @@ public abstract class FailbackRegistry extends AbstractRegistry {
throw new IllegalArgumentException("notify listener == null");
}
try {
super.notify(url, listener, urls);
doNotify(url, listener, urls);
} catch (Exception t) {
// 将失败的通知请求记录到失败列表，定时重试
Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
@ -233,6 +233,10 @@ public abstract class FailbackRegistry extends AbstractRegistry {
}
}
protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
super.notify(url, listener, urls);
}
@Override
protected void recover() throws Exception {
// register

22
dubbo-simple/dubbo-monitor-simple/src/main/java/com/alibaba/dubbo/monitor/simple/RegistryContainer.java
View File

@ -27,15 +27,13 @@ import java.util.concurrent.ConcurrentHashMap;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.ConcurrentHashSet;
import com.alibaba.dubbo.common.utils.ConfigUtils;
import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.dubbo.container.Container;
import com.alibaba.dubbo.container.spring.SpringContainer;
import com.alibaba.dubbo.registry.NotifyListener;
import com.alibaba.dubbo.registry.Registry;
import com.alibaba.dubbo.registry.RegistryFactory;
import com.alibaba.dubbo.registry.RegistryService;
/**
* RegistryContainer
@ -44,12 +42,8 @@ import com.alibaba.dubbo.registry.RegistryFactory;
*/
public class RegistryContainer implements Container {
private static final Logger logger = LoggerFactory.getLogger(RegistryContainer.class);
public static final String REGISTRY_ADDRESS = "dubbo.registry.address";
private final RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension();
private final Set<String> applications = new ConcurrentHashSet<String>();
private final Map<String, Set<String>> providerServiceApplications = new ConcurrentHashMap<String, Set<String>>();
@ -66,7 +60,7 @@ public class RegistryContainer implements Container {
private final Map<String, List<URL>> serviceConsumers = new ConcurrentHashMap<String, List<URL>>();
private Registry registry;
private RegistryService registry;
private static RegistryContainer INSTANCE = null;
@ -81,7 +75,7 @@ public class RegistryContainer implements Container {
return INSTANCE;
}
public Registry getRegistry() {
public RegistryService getRegistry() {
return registry;
}
@ -215,8 +209,7 @@ public class RegistryContainer implements Container {
if (url == null || url.length() == 0) {
throw new IllegalArgumentException("Please set java start argument: -D" + REGISTRY_ADDRESS + "=zookeeper://127.0.0.1:2181");
}
URL registryUrl = URL.valueOf(url);
registry = registryFactory.getRegistry(registryUrl);
registry = (RegistryService) SpringContainer.getContext().getBean("registryService");
URL subscribeUrl = new URL(Constants.ADMIN_PROTOCOL, NetUtils.getLocalHost(), 0, "",
Constants.INTERFACE_KEY, Constants.ANY_VALUE,
Constants.GROUP_KEY, Constants.ANY_VALUE,
@ -306,11 +299,6 @@ public class RegistryContainer implements Container {
}
public void stop() {
try {
registry.destroy();
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
}

10
dubbo-simple/dubbo-monitor-simple/src/main/resources/META-INF/spring/dubbo-monitor-simple.xml
View File

@ -24,6 +24,11 @@
<property name="location" value="classpath:dubbo.properties" />
</bean>
<bean id="monitorService" class="com.alibaba.dubbo.monitor.simple.SimpleMonitorService">
<property name="statisticsDirectory" value="${dubbo.statistics.directory}" />
<property name="chartsDirectory" value="${dubbo.charts.directory}" />
</bean>
<dubbo:application name="${dubbo.application.name}" owner="${dubbo.application.owner}" />
<dubbo:registry address="${dubbo.registry.address}" />
@ -32,9 +37,6 @@
<dubbo:service interface="com.alibaba.dubbo.monitor.MonitorService" ref="monitorService" delay="-1" />
<bean id="monitorService" class="com.alibaba.dubbo.monitor.simple.SimpleMonitorService">
<property name="statisticsDirectory" value="${dubbo.statistics.directory}" />
<property name="chartsDirectory" value="${dubbo.charts.directory}" />
</bean>
<dubbo:reference id="registryService" interface="com.alibaba.dubbo.registry.RegistryService" />
</beans>

124
dubbo-simple/dubbo-registry-simple/src/main/java/com/alibaba/dubbo/registry/simple/SimpleRegistryService.java
View File

@ -15,6 +15,8 @@
*/
package com.alibaba.dubbo.registry.simple;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -30,47 +32,57 @@ import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.dubbo.common.utils.UrlUtils;
import com.alibaba.dubbo.registry.NotifyListener;
import com.alibaba.dubbo.registry.RegistryService;
import com.alibaba.dubbo.registry.support.FailbackRegistry;
import com.alibaba.dubbo.registry.support.AbstractRegistry;
import com.alibaba.dubbo.rpc.RpcContext;
/**
* DubboRegistryService
* SimpleRegistryService
*
* @author william.liangf
*/
public class SimpleRegistryService extends FailbackRegistry {
public class SimpleRegistryService extends AbstractRegistry {
private final ConcurrentMap<String, Set<String>> remoteRegistered = new ConcurrentHashMap<String, Set<String>>();
private final ConcurrentMap<String, Set<URL>> remoteRegistered = new ConcurrentHashMap<String, Set<URL>>();
private final ConcurrentMap<String, ConcurrentMap<String, Set<NotifyListener>>> remoteSubscribed = new ConcurrentHashMap<String, ConcurrentMap<String, Set<NotifyListener>>>();
private final ConcurrentMap<String, ConcurrentMap<URL, Set<NotifyListener>>> remoteSubscribed = new ConcurrentHashMap<String, ConcurrentMap<URL, Set<NotifyListener>>>();
private final static Logger logger = LoggerFactory.getLogger(SimpleRegistryService.class);
public SimpleRegistryService() {
super(new URL("dubbo", NetUtils.getLocalHost(), 0, RegistryService.class.getName()));
super(new URL("dubbo", NetUtils.getLocalHost(), 0, RegistryService.class.getName(), "file", "N/A"));
}
public boolean isAvailable() {
return true;
}
public List<URL> lookup(URL url) {
List<URL> urls = new ArrayList<URL>();
for (URL u: getRegistered()) {
if (UrlUtils.isMatch(url, u)) {
urls.add(u);
}
}
return urls;
}
public void register(URL url) {
String client = RpcContext.getContext().getRemoteAddressString();
Set<String> urls = remoteRegistered.get(client);
Set<URL> urls = remoteRegistered.get(client);
if (urls == null) {
remoteRegistered.putIfAbsent(client, new ConcurrentHashSet<String>());
remoteRegistered.putIfAbsent(client, new ConcurrentHashSet<URL>());
urls = remoteRegistered.get(client);
}
urls.add(url.toFullString());
urls.add(url);
super.register(url);
registered(url);
}
public void unregister(URL url) {
String client = RpcContext.getContext().getRemoteAddressString();
Set<String> urls = remoteRegistered.get(client);
Set<URL> urls = remoteRegistered.get(client);
if (urls != null && urls.size() > 0) {
urls.remove(url.toFullString());
urls.remove(url);
}
super.unregister(url);
unregistered(url);
@ -78,23 +90,23 @@ public class SimpleRegistryService extends FailbackRegistry {
public void subscribe(URL url, NotifyListener listener) {
if (getUrl().getPort() == 0) {
URL registryUrl = RpcContext.getContext().getInvoker().getUrl();
if (registryUrl != null && registryUrl.getPort() > 0) {
URL registryUrl = RpcContext.getContext().getUrl();
if (registryUrl != null && registryUrl.getPort() > 0
&& RegistryService.class.getName().equals(registryUrl.getPath())) {
super.setUrl(registryUrl);
super.register(registryUrl);
}
}
String client = RpcContext.getContext().getRemoteAddressString();
ConcurrentMap<String, Set<NotifyListener>> clientListeners = remoteSubscribed.get(client);
ConcurrentMap<URL, Set<NotifyListener>> clientListeners = remoteSubscribed.get(client);
if (clientListeners == null) {
remoteSubscribed.putIfAbsent(client, new ConcurrentHashMap<String, Set<NotifyListener>>());
remoteSubscribed.putIfAbsent(client, new ConcurrentHashMap<URL, Set<NotifyListener>>());
clientListeners = remoteSubscribed.get(client);
}
String key = url.toFullString();
Set<NotifyListener> listeners = clientListeners.get(key);
Set<NotifyListener> listeners = clientListeners.get(url);
if (listeners == null) {
clientListeners.putIfAbsent(key, new ConcurrentHashSet<NotifyListener>());
listeners = clientListeners.get(key);
clientListeners.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
listeners = clientListeners.get(url);
}
listeners.add(listener);
super.subscribe(url, listener);
@ -107,15 +119,13 @@ public class SimpleRegistryService extends FailbackRegistry {
unregister(url);
}
String client = RpcContext.getContext().getRemoteAddressString();
Map<String, Set<NotifyListener>> clientListeners = remoteSubscribed.get(client);
Map<URL, Set<NotifyListener>> clientListeners = remoteSubscribed.get(client);
if (clientListeners != null && clientListeners.size() > 0) {
String key = url.toFullString();
Set<NotifyListener> listeners = clientListeners.get(key);
Set<NotifyListener> listeners = clientListeners.get(url);
if (listeners != null && listeners.size() > 0) {
listeners.remove(listener);
}
}
super.unregister(url);
}
protected void registered(URL url) {
@ -124,7 +134,7 @@ public class SimpleRegistryService extends FailbackRegistry {
if (UrlUtils.isMatch(key, url)) {
List<URL> list = lookup(key);
for (NotifyListener listener : entry.getValue()) {
notify(key, listener, list);
listener.notify(list);
}
}
}
@ -136,46 +146,64 @@ public class SimpleRegistryService extends FailbackRegistry {
if (UrlUtils.isMatch(key, url)) {
List<URL> list = lookup(key);
for (NotifyListener listener : entry.getValue()) {
notify(key, listener, list);
listener.notify(list);
}
}
}
}
protected void subscribed(URL url, NotifyListener listener) {
List<URL> urls = lookup(url);
notify(url, listener, urls);
protected void subscribed(final URL url, final NotifyListener listener) {
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
new Thread(new Runnable() {
public void run() {
Map<String, List<URL>> map = new HashMap<String, List<URL>>();
for (URL u: getRegistered()) {
if (UrlUtils.isMatch(url, u)) {
String service = u.getServiceInterface();
List<URL> list = map.get(service);
if (list == null) {
list = new ArrayList<URL>();
map.put(service, list);
}
list.add(u);
}
}
for (List<URL> list : map.values()) {
try {
listener.notify(list);
} catch (Throwable e) {
logger.warn("Discard to notify " + url.getServiceKey() + " to listener " + listener);
}
}
}
}, "DubboMonitorNotifier").start();
} else {
List<URL> list = lookup(url);
try {
listener.notify(list);
} catch (Throwable e) {
logger.warn("Discard to notify " + url.getServiceKey() + " to listener " + listener);
}
}
}
public void doRegister(URL url) {
}
public void doUnregister(URL url) {
}
public void doSubscribe(URL url, NotifyListener listener) {
}
public void doUnsubscribe(URL url, NotifyListener listener) {
}
public void disconnect() {
String client = RpcContext.getContext().getRemoteAddressString();
if (logger.isInfoEnabled()) {
logger.info("Disconnected " + client);
}
Set<String> urls = remoteRegistered.get(client);
Set<URL> urls = remoteRegistered.get(client);
if (urls != null && urls.size() > 0) {
for (String url : urls) {
unregister(URL.valueOf(url));
for (URL url : urls) {
unregister(url);
}
}
Map<String, Set<NotifyListener>> listeners = remoteSubscribed.get(client);
Map<URL, Set<NotifyListener>> listeners = remoteSubscribed.get(client);
if (listeners != null && listeners.size() > 0) {
for (Map.Entry<String, Set<NotifyListener>> entry : listeners.entrySet()) {
String url = entry.getKey();
for (Map.Entry<URL, Set<NotifyListener>> entry : listeners.entrySet()) {
URL url = entry.getKey();
for (NotifyListener listener : entry.getValue()) {
unsubscribe(URL.valueOf(url), listener);
unsubscribe(url, listener);
}
}
}