From 4fd13c321afd5003e33138d5bf285a8199c6a783 Mon Sep 17 00:00:00 2001 From: liangfei0201 Date: Fri, 29 Jun 2012 18:41:16 +0800 Subject: [PATCH] =?UTF-8?q?DUBBO-432=20=E4=BF=AE=E5=A4=8DSimpleRegistry?= =?UTF-8?q?=E4=B8=8D=E5=8F=AF=E7=94=A8=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../container/spring/SpringContainer.java | 9 +- .../integration/RegistryDirectory.java | 2 +- .../integration/RegistryProtocol.java | 14 +- .../registry/support/AbstractRegistry.java | 2 +- .../registry/support/FailbackRegistry.java | 6 +- .../monitor/simple/RegistryContainer.java | 22 +--- .../META-INF/spring/dubbo-monitor-simple.xml | 10 +- .../simple/SimpleRegistryService.java | 124 +++++++++++------- 8 files changed, 114 insertions(+), 75 deletions(-) diff --git a/dubbo-container/dubbo-container-spring/src/main/java/com/alibaba/dubbo/container/spring/SpringContainer.java b/dubbo-container/dubbo-container-spring/src/main/java/com/alibaba/dubbo/container/spring/SpringContainer.java index c955259e56..09dd95e837 100644 --- a/dubbo-container/dubbo-container-spring/src/main/java/com/alibaba/dubbo/container/spring/SpringContainer.java +++ b/dubbo-container/dubbo-container-spring/src/main/java/com/alibaba/dubbo/container/spring/SpringContainer.java @@ -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; diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryDirectory.java index 3a8cb7ecfb..d7d5beee4a 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryDirectory.java @@ -114,7 +114,7 @@ public class RegistryDirectory extends AbstractDirectory 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); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java index 0d4dfb985c..81bd2035c3 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/integration/RegistryProtocol.java @@ -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 Invoker refer(Class type, URL url) throws RpcException { + @SuppressWarnings("unchecked") + public Invoker refer(Class 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 qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY)); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java index 5fc6f95dc1..f4bc977815 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistry.java @@ -261,7 +261,7 @@ public abstract class AbstractRegistry implements Registry { } return null; } - + public List lookup(URL url) { List result = new ArrayList(); Map> notifiedUrls = getNotified().get(url); diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java index 8e3b682292..ce06ba3849 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/com/alibaba/dubbo/registry/support/FailbackRegistry.java @@ -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> listeners = failedNotified.get(url); @@ -233,6 +233,10 @@ public abstract class FailbackRegistry extends AbstractRegistry { } } + protected void doNotify(URL url, NotifyListener listener, List urls) { + super.notify(url, listener, urls); + } + @Override protected void recover() throws Exception { // register diff --git a/dubbo-simple/dubbo-monitor-simple/src/main/java/com/alibaba/dubbo/monitor/simple/RegistryContainer.java b/dubbo-simple/dubbo-monitor-simple/src/main/java/com/alibaba/dubbo/monitor/simple/RegistryContainer.java index 9db3a049f5..61bfcffe37 100644 --- a/dubbo-simple/dubbo-monitor-simple/src/main/java/com/alibaba/dubbo/monitor/simple/RegistryContainer.java +++ b/dubbo-simple/dubbo-monitor-simple/src/main/java/com/alibaba/dubbo/monitor/simple/RegistryContainer.java @@ -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 applications = new ConcurrentHashSet(); private final Map> providerServiceApplications = new ConcurrentHashMap>(); @@ -66,7 +60,7 @@ public class RegistryContainer implements Container { private final Map> serviceConsumers = new ConcurrentHashMap>(); - 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); - } } } \ No newline at end of file diff --git a/dubbo-simple/dubbo-monitor-simple/src/main/resources/META-INF/spring/dubbo-monitor-simple.xml b/dubbo-simple/dubbo-monitor-simple/src/main/resources/META-INF/spring/dubbo-monitor-simple.xml index f6bfef224f..d8a9105eeb 100644 --- a/dubbo-simple/dubbo-monitor-simple/src/main/resources/META-INF/spring/dubbo-monitor-simple.xml +++ b/dubbo-simple/dubbo-monitor-simple/src/main/resources/META-INF/spring/dubbo-monitor-simple.xml @@ -24,6 +24,11 @@ + + + + + @@ -32,9 +37,6 @@ - - - - + \ No newline at end of file diff --git a/dubbo-simple/dubbo-registry-simple/src/main/java/com/alibaba/dubbo/registry/simple/SimpleRegistryService.java b/dubbo-simple/dubbo-registry-simple/src/main/java/com/alibaba/dubbo/registry/simple/SimpleRegistryService.java index 5fbfc34c00..01269f4a0b 100644 --- a/dubbo-simple/dubbo-registry-simple/src/main/java/com/alibaba/dubbo/registry/simple/SimpleRegistryService.java +++ b/dubbo-simple/dubbo-registry-simple/src/main/java/com/alibaba/dubbo/registry/simple/SimpleRegistryService.java @@ -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> remoteRegistered = new ConcurrentHashMap>(); + private final ConcurrentMap> remoteRegistered = new ConcurrentHashMap>(); - private final ConcurrentMap>> remoteSubscribed = new ConcurrentHashMap>>(); + private final ConcurrentMap>> remoteSubscribed = new ConcurrentHashMap>>(); 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 lookup(URL url) { + List urls = new ArrayList(); + 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 urls = remoteRegistered.get(client); + Set urls = remoteRegistered.get(client); if (urls == null) { - remoteRegistered.putIfAbsent(client, new ConcurrentHashSet()); + remoteRegistered.putIfAbsent(client, new ConcurrentHashSet()); 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 urls = remoteRegistered.get(client); + Set 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> clientListeners = remoteSubscribed.get(client); + ConcurrentMap> clientListeners = remoteSubscribed.get(client); if (clientListeners == null) { - remoteSubscribed.putIfAbsent(client, new ConcurrentHashMap>()); + remoteSubscribed.putIfAbsent(client, new ConcurrentHashMap>()); clientListeners = remoteSubscribed.get(client); } - String key = url.toFullString(); - Set listeners = clientListeners.get(key); + Set listeners = clientListeners.get(url); if (listeners == null) { - clientListeners.putIfAbsent(key, new ConcurrentHashSet()); - listeners = clientListeners.get(key); + clientListeners.putIfAbsent(url, new ConcurrentHashSet()); + 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> clientListeners = remoteSubscribed.get(client); + Map> clientListeners = remoteSubscribed.get(client); if (clientListeners != null && clientListeners.size() > 0) { - String key = url.toFullString(); - Set listeners = clientListeners.get(key); + Set 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 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 list = lookup(key); for (NotifyListener listener : entry.getValue()) { - notify(key, listener, list); + listener.notify(list); } } } } - protected void subscribed(URL url, NotifyListener listener) { - List 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> map = new HashMap>(); + for (URL u: getRegistered()) { + if (UrlUtils.isMatch(url, u)) { + String service = u.getServiceInterface(); + List list = map.get(service); + if (list == null) { + list = new ArrayList(); + map.put(service, list); + } + list.add(u); + } + } + for (List list : map.values()) { + try { + listener.notify(list); + } catch (Throwable e) { + logger.warn("Discard to notify " + url.getServiceKey() + " to listener " + listener); + } + } + } + }, "DubboMonitorNotifier").start(); + } else { + List 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 urls = remoteRegistered.get(client); + Set 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> listeners = remoteSubscribed.get(client); + Map> listeners = remoteSubscribed.get(client); if (listeners != null && listeners.size() > 0) { - for (Map.Entry> entry : listeners.entrySet()) { - String url = entry.getKey(); + for (Map.Entry> entry : listeners.entrySet()) { + URL url = entry.getKey(); for (NotifyListener listener : entry.getValue()) { - unsubscribe(URL.valueOf(url), listener); + unsubscribe(url, listener); } } }