diff --git a/dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java b/dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java index 2440a4a777..f51bb76d1a 100644 --- a/dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java +++ b/dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java @@ -255,6 +255,7 @@ public class Constants { public static final String DEFAULT_CHANNEL_HANDLER = "default"; + public static final String ANY_VALUE = "*"; public static final String COMMA_SEPARATOR = ","; diff --git a/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/StringUtils.java b/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/StringUtils.java index 16ccc0d763..eca8cdd6b7 100644 --- a/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/StringUtils.java +++ b/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/StringUtils.java @@ -73,6 +73,20 @@ public final class StringUtils return str != null && str.length() > 0; } + /** + * + * @param s1 + * @param s2 + * @return + */ + public static boolean isEquals(String s1, String s2) { + if (s1 == null && s2 == null) + return true; + if (s1 == null || s2 == null) + return false; + return s1.equals(s2); + } + /** * is integer string. * diff --git a/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/UrlUtils.java b/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/UrlUtils.java index b42dce7e1f..86bfeacc2b 100644 --- a/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/UrlUtils.java +++ b/dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/UrlUtils.java @@ -324,5 +324,14 @@ public class UrlUtils { } return forbid; } - + + public static boolean isMatch(URL consumerUrl, URL providerUrl) { + String consumerGroup = consumerUrl.getParameter(Constants.GROUP_KEY); + String consumerVersion = consumerUrl.getParameter(Constants.VERSION_KEY); + String providerGroup = providerUrl.getParameter(Constants.GROUP_KEY); + String providerVersion = providerUrl.getParameter(Constants.VERSION_KEY); + return (Constants.ANY_VALUE.equals(consumerGroup) || StringUtils.isEquals(consumerGroup, providerGroup)) + && (Constants.ANY_VALUE.equals(consumerVersion) || StringUtils.isEquals(consumerVersion, providerVersion)); + } + } \ No newline at end of file diff --git a/dubbo-config/src/main/java/com/alibaba/dubbo/config/ServiceConfig.java b/dubbo-config/src/main/java/com/alibaba/dubbo/config/ServiceConfig.java index 42f666768e..528ac61c16 100644 --- a/dubbo-config/src/main/java/com/alibaba/dubbo/config/ServiceConfig.java +++ b/dubbo-config/src/main/java/com/alibaba/dubbo/config/ServiceConfig.java @@ -380,8 +380,8 @@ public class ServiceConfig extends AbstractServiceConfig { } } if (generic) { - map.put("generic", "true"); - map.put("methods", "*"); + map.put("generic", String.valueOf(true)); + map.put("methods", Constants.ANY_VALUE); } else { map.put("revision", Version.getVersion(interfaceClass, version)); map.put("methods", StringUtils.join(new HashSet(Arrays.asList(Wrapper.getWrapper(interfaceClass).getDeclaredMethodNames())), ",")); diff --git a/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java b/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java index 9cee9dc285..b0e56b1220 100644 --- a/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java +++ b/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java @@ -17,7 +17,6 @@ package com.alibaba.dubbo.registry.multicast; import java.io.IOException; import java.net.DatagramPacket; -import java.net.DatagramSocket; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.MulticastSocket; @@ -27,7 +26,6 @@ import java.util.List; import com.alibaba.dubbo.common.URL; import com.alibaba.dubbo.common.logger.Logger; import com.alibaba.dubbo.common.logger.LoggerFactory; -import com.alibaba.dubbo.common.utils.NetUtils; import com.alibaba.dubbo.common.utils.StringUtils; import com.alibaba.dubbo.registry.NotifyListener; import com.alibaba.dubbo.registry.support.AbstractRegistry; @@ -54,10 +52,6 @@ public class MulticastRegistry extends AbstractRegistry { private MulticastSocket mutilcastSocket; - private InetSocketAddress datagramAddress; - - private DatagramSocket datagramSocket; - public MulticastRegistry(URL url) { super(url); if (! isMulticastAddress(url.getHost())) { @@ -91,41 +85,6 @@ public class MulticastRegistry extends AbstractRegistry { } catch (IOException e) { throw new IllegalStateException(e.getMessage(), e); } - int port = 0; - String udp = url.getParameter("udp"); - if (udp == null || udp.length() == 0 || "true".equals(udp)) { - port = NetUtils.getAvailablePort(); - } else if (! "false".equals(udp)) { - port = Integer.parseInt(udp); - } - if (port > 0) { - try { - datagramAddress = new InetSocketAddress(NetUtils.getLocalHost(), url.getParameter("udp", NetUtils.getAvailablePort())); - datagramSocket = new DatagramSocket(datagramAddress); - Thread thread = new Thread(new Runnable() { - public void run() { - byte[] buf = new byte[1024]; - DatagramPacket recv = new DatagramPacket(buf, buf.length); - while (true) { - try { - datagramSocket.receive(recv); - String msg = new String(recv.getData()).trim(); - if (logger.isInfoEnabled()) { - logger.info("Receive udp message: " + msg + " from " + recv.getSocketAddress()); - } - MulticastRegistry.this.receive(msg, (InetSocketAddress) recv.getSocketAddress()); - } catch (IOException e) { - logger.error(e.getMessage(), e); - } - } - } - }, "MulticastRegistryUDP"); - thread.setDaemon(true); - thread.start(); - } catch (IOException e) { - throw new IllegalStateException(e.getMessage(), e); - } - } } private static boolean isMulticastAddress(String ip) { @@ -172,18 +131,14 @@ public class MulticastRegistry extends AbstractRegistry { String service = url.getServiceKey(); if (getRegistered().containsKey(service)) { for (URL u : getRegistered().get(service)) { - if (datagramSocket != null && "udp".equals(url.getProtocol())) { - sendTo(REGISTER + " " + u.toFullString(), url); - } else { - send(REGISTER + " " + u.toFullString()); - } + unicast(REGISTER + " " + u.toFullString(), url); } } } else if (msg.startsWith(UNSUBSCRIBE)) { } } - private void send(String msg) { + private void broadcast(String msg) { if (logger.isInfoEnabled()) { logger.info("Send multicast message: " + msg + " to " + mutilcastAddress + ":" + mutilcastSocket.getLocalPort()); } @@ -195,13 +150,13 @@ public class MulticastRegistry extends AbstractRegistry { } } - private void sendTo(String msg, URL url) { + private void unicast(String msg, URL url) { if (logger.isInfoEnabled()) { logger.info("Send udp message: " + msg + " to " + url.getAddress()); } try { - DatagramPacket hi = new DatagramPacket(msg.getBytes(), msg.length(), InetAddress.getByName(url.getHost()), url.getPort()); - datagramSocket.send(hi); + DatagramPacket hi = new DatagramPacket(msg.getBytes(), msg.length(), InetAddress.getByName(url.getHost()), mutilcastSocket.getLocalPort()); + mutilcastSocket.send(hi); } catch (Exception e) { throw new IllegalStateException(e.getMessage(), e); } @@ -209,21 +164,17 @@ public class MulticastRegistry extends AbstractRegistry { public void register(String service, URL url) { super.register(service, url); - send(REGISTER + " " + url.toFullString()); + broadcast(REGISTER + " " + url.toFullString()); } public void unregister(String service, URL url) { - send(UNREGISTER + " " + url.toFullString()); + broadcast(UNREGISTER + " " + url.toFullString()); } public void subscribe(String service, URL url, NotifyListener listener) { super.subscribe(service, url, listener); - if (datagramAddress != null) { - url = url.setProtocol("udp").setHost(datagramAddress.getAddress().getHostAddress()).setPort(datagramAddress.getPort()); - } else { - url = url.setProtocol("multicast").setHost(mutilcastAddress.getHostAddress()).setPort(mutilcastSocket.getLocalPort()); - } - send(SUBSCRIBE + " " + url.toFullString()); + url = url.setProtocol("multicast").setHost(mutilcastAddress.getHostAddress()).setPort(mutilcastSocket.getLocalPort()); + broadcast(SUBSCRIBE + " " + url.toFullString()); synchronized (this) { try { this.wait(5000); @@ -233,12 +184,8 @@ public class MulticastRegistry extends AbstractRegistry { } public void unsubscribe(String service, URL url, NotifyListener listener) { - if (datagramAddress != null) { - url = url.setProtocol("udp").setHost(datagramAddress.getAddress().getHostAddress()).setPort(datagramAddress.getPort()); - } else { - url = url.setProtocol("multicast").setHost(mutilcastAddress.getHostAddress()).setPort(mutilcastSocket.getLocalPort()); - } - send(UNSUBSCRIBE + " " + url.toFullString()); + url = url.setProtocol("multicast").setHost(mutilcastAddress.getHostAddress()).setPort(mutilcastSocket.getLocalPort()); + broadcast(UNSUBSCRIBE + " " + url.toFullString()); } public boolean isAvailable() { diff --git a/dubbo-registry-zookeeper/src/main/java/com/alibaba/dubbo/registry/zookeeper/ZookeeperRegistry.java b/dubbo-registry-zookeeper/src/main/java/com/alibaba/dubbo/registry/zookeeper/ZookeeperRegistry.java index 78c762437f..e91fa1a39d 100644 --- a/dubbo-registry-zookeeper/src/main/java/com/alibaba/dubbo/registry/zookeeper/ZookeeperRegistry.java +++ b/dubbo-registry-zookeeper/src/main/java/com/alibaba/dubbo/registry/zookeeper/ZookeeperRegistry.java @@ -15,7 +15,6 @@ */ package com.alibaba.dubbo.registry.zookeeper; -import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -35,6 +34,7 @@ import com.alibaba.dubbo.common.Constants; import com.alibaba.dubbo.common.URL; import com.alibaba.dubbo.common.logger.Logger; import com.alibaba.dubbo.common.logger.LoggerFactory; +import com.alibaba.dubbo.common.utils.UrlUtils; import com.alibaba.dubbo.registry.NotifyListener; import com.alibaba.dubbo.registry.Registry; import com.alibaba.dubbo.rpc.RpcException; @@ -52,6 +52,8 @@ public class ZookeeperRegistry implements Registry { private final URL url; + private final String root; + private final boolean auth; private final ZooKeeper zookeeper; @@ -79,7 +81,17 @@ public class ZookeeperRegistry implements Registry { if (auth) { zookeeper.addAuthInfo(url.getUsername(), url.getPassword().getBytes()); } - } catch (IOException e) { + String root = url.getParameter("root"); + if (root != null && root.length() > 0) { + root = SEPARATOR + root; + this.root = root; + if (zookeeper.exists(root, false) == null) { + zookeeper.create(root, new byte[0], auth ? Ids.CREATOR_ALL_ACL : Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + } + } else { + this.root = ""; + } + } catch (Exception e) { throw new IllegalStateException(e.getMessage(), e); } } @@ -148,7 +160,7 @@ public class ZookeeperRegistry implements Registry { } serviceWathers.put(listener, wather); List providers = zookeeper.getChildren(service, wather); - List urls = toUrls(service, providers); + List urls = toUrls(url, providers); if (urls != null && urls.size() > 0) { listener.notify(urls); } @@ -173,7 +185,7 @@ public class ZookeeperRegistry implements Registry { } } } - + public List lookup(URL url) { if (url == null) { throw new IllegalArgumentException("lookup url == null"); @@ -181,21 +193,22 @@ public class ZookeeperRegistry implements Registry { try { String service = toServicePath(url); List providers = zookeeper.getChildren(service, false); - return toUrls(service, providers); + return toUrls(url, providers); } catch (Throwable e) { throw new RpcException("Failed to lookup " + url + ", cause: " + e.getMessage(), e); } } private String toServicePath(URL url) { - return SEPARATOR + URL.encode(url.getServiceKey()); + return root + SEPARATOR + URL.encode(url.getParameter(Constants.INTERFACE_KEY, url.getPath())); } private String toProviderPath(URL url) { return SEPARATOR + URL.encode(url.toIdentityString()); } - private List toUrls(String service, List providers) throws KeeperException, InterruptedException { + private List toUrls(URL consumer, List providers) throws KeeperException, InterruptedException { + String service = toServicePath(consumer); List urls = new ArrayList(); for (String provider : providers) { String path = service + provider; @@ -207,7 +220,10 @@ public class ZookeeperRegistry implements Registry { query = "?" + new String(data); } } - urls.add(URL.valueOf(URL.decode(provider + query))); + URL url = URL.valueOf(URL.decode(provider + query)); + if (UrlUtils.isMatch(consumer, url)) { + urls.add(url); + } } return urls; } diff --git a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/RegistryDirectory.java b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/RegistryDirectory.java index d755d1a91a..d125d24b52 100644 --- a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/RegistryDirectory.java +++ b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/RegistryDirectory.java @@ -57,8 +57,6 @@ public class RegistryDirectory extends AbstractDirectory implements Notify private static final Logger logger = LoggerFactory.getLogger(RegistryDirectory.class); - private static final String ALL_METHOD_NAME = "*"; - private volatile boolean forbidden = false; private final String serviceKey; @@ -278,7 +276,7 @@ public class RegistryDirectory extends AbstractDirectory implements Notify if (methods != null && methods.length > 0) { for (String method : methods) { if (method != null && method.length() > 0 - && ! ALL_METHOD_NAME.equals(method)) { + && ! Constants.ANY_VALUE.equals(method)) { List> methodInvokers = methodInvokerMap.get(method); if (methodInvokers == null) { methodInvokers = new ArrayList>(); @@ -291,7 +289,7 @@ public class RegistryDirectory extends AbstractDirectory implements Notify } invokersList.add(invoker); } - methodInvokerMap.put(ALL_METHOD_NAME, invokersList); + methodInvokerMap.put(Constants.ANY_VALUE, invokersList); } // sort and unmodifiable for (String method : new HashSet(methodInvokerMap.keySet())) { @@ -403,7 +401,7 @@ public class RegistryDirectory extends AbstractDirectory implements Notify invokers = methodInvokerMap.get(methodName); } if(invokers == null) { - invokers = methodInvokerMap.get(ALL_METHOD_NAME); + invokers = methodInvokerMap.get(Constants.ANY_VALUE); } if(invokers == null) { Iterator>> iterator = methodInvokerMap.values().iterator();