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

修改注册中心 

git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@155 1a56cb94-b969-4eaa-88fa-be21384802f2
This commit is contained in:
william.liangf 2011-11-01 05:03:56 +00:00
parent 96aee0c65c
commit 099d639bc9
7 changed files with 65 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
dubbo-common/src/main/java/com/alibaba/dubbo/common/Constants.java
View File

@ -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 = ",";

14
dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/StringUtils.java
View File

@ -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.
*

11
dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/UrlUtils.java
View File

@ -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));
}
}

4
dubbo-config/src/main/java/com/alibaba/dubbo/config/ServiceConfig.java
View File

@ -380,8 +380,8 @@ public class ServiceConfig<T> 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<String>(Arrays.asList(Wrapper.getWrapper(interfaceClass).getDeclaredMethodNames())), ","));

75
dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java
View File

@ -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() {

32
dubbo-registry-zookeeper/src/main/java/com/alibaba/dubbo/registry/zookeeper/ZookeeperRegistry.java
View File

@ -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<String> providers = zookeeper.getChildren(service, wather);
List<URL> urls = toUrls(service, providers);
List<URL> urls = toUrls(url, providers);
if (urls != null && urls.size() > 0) {
listener.notify(urls);
}
@ -173,7 +185,7 @@ public class ZookeeperRegistry implements Registry {
}
}
}
public List<URL> 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<String> 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<URL> toUrls(String service, List<String> providers) throws KeeperException, InterruptedException {
private List<URL> toUrls(URL consumer, List<String> providers) throws KeeperException, InterruptedException {
String service = toServicePath(consumer);
List<URL> urls = new ArrayList<URL>();
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;
}

8
dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/RegistryDirectory.java
View File

@ -57,8 +57,6 @@ public class RegistryDirectory<T> extends AbstractDirectory<T> 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<T> extends AbstractDirectory<T> 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<Invoker<T>> methodInvokers = methodInvokerMap.get(method);
if (methodInvokers == null) {
methodInvokers = new ArrayList<Invoker<T>>();
@ -291,7 +289,7 @@ public class RegistryDirectory<T> extends AbstractDirectory<T> 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<String>(methodInvokerMap.keySet())) {
@ -403,7 +401,7 @@ public class RegistryDirectory<T> extends AbstractDirectory<T> 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<List<Invoker<T>>> iterator = methodInvokerMap.values().iterator();