为了安全：服务启动的ip全部使用10.10.10.10
远程服务的暴露总体步骤：
- 将ref封装为invoker
- 将invoker转换为exporter
- 启动netty
- 注册服务到zookeeper
- 订阅
- 返回新的exporter实例
在7.4 服务远程暴露 - 创建Exporter与启动netty服务端中，实现了前三步，本节实现第四步：注册服务到zk。总体代码如下：RegistryProtocol.export(final Invoker<T> originInvoker)
final Registry registry = getRegistry(originInvoker);//创建ZookeeperRegistry实例：创建CuratorClient，并启动会话。
final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);//获取真正要注册在zk上的url
registry.register(registedProviderUrl);//创建节点（即注册服务到zk上）
说明：
- 第一句代码用来创建ZookeeperRegistry实例：创建CuratorClient，并启动会话。
- 第二句代码获取真正要注册在zk上的url
- 第三句代码实现创建节点（即注册服务到zk上）
一 创建ZookeeperRegistry实例
1 RegistryProtocol.getRegistry(final Invoker<?> originInvoker)
/**
* 根据invoker的地址获取registry实例
*/
private Registry getRegistry(final Invoker<?> originInvoker) {
URL registryUrl = originInvoker.getUrl();
if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);//zookeeper
registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
}
return registryFactory.getRegistry(registryUrl);
}
首先对originInvoker中的url进行处理：
- 将协议换成zookeeper
- 去掉registry=zookeeper的参数
来看一下originInvoker的url：（解码后的）
registry://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&export=dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=2791&side=provider×tamp=1507262031554&pid=2791®istry=zookeeper×tamp=1507262031521
说明：
- 第一个红色部分代表协议：zookeeper
- 第二个红色部分是export参数
- 第三个红色部分是registry=zookeeper
经过处理之后的registryUrl为：
zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&export=dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=2791&side=provider×tamp=1507262031554&pid=2791×tamp=1507262031521
之后使用注册工厂来创建注册中心。
2 RegistryFactory$Adaptive.getRegistry(com.alibaba.dubbo.common.URL registryUrl)
public class RegistryFactory$Adaptive implements com.alibaba.dubbo.registry.RegistryFactory {
public com.alibaba.dubbo.registry.Registry getRegistry(com.alibaba.dubbo.common.URL arg0) {
if (arg0 == null)
throw new IllegalArgumentException("url == null");
com.alibaba.dubbo.common.URL url = arg0;
String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );//zookeeper
if(extName == null)
throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.registry.RegistryFactory) name from url(" + url.toString() + ") use keys([protocol])");
com.alibaba.dubbo.registry.RegistryFactory extension = (com.alibaba.dubbo.registry.RegistryFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.registry.RegistryFactory.class).getExtension(extName);
return extension.getRegistry(arg0);
}
}
这里获取到的extension是ZookeeperRegistryFactory，之后，使用ZookeeperRegistryFactory进行Registry的创建。首先来看一下ZookeeperRegistryFactory的继承图：
getRegistry方法在ZookeeperRegistryFactory的父类AbstractRegistryFactory中。
3 AbstractRegistryFactory.getRegistry(URL registryUrl)
public Registry getRegistry(URL url) {
url = url.setPath(RegistryService.class.getName())
.addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
.removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
String key = url.toServiceString();
// 锁定注册中心获取过程，保证注册中心单一实例
LOCK.lock();
try {
Registry registry = REGISTRIES.get(key);
if (registry != null) {
return registry;
}
registry = createRegistry(url);
if (registry == null) {
throw new IllegalStateException("Can not create registry " + url);
}
REGISTRIES.put(key, registry);
return registry;
} finally {
// 释放锁
LOCK.unlock();
}
}
流程：
- 先处理url，之后获取Registry的key，然后根据该key从Map<String, Registry> REGISTRIES注册中心集合缓存中获取Registry，如果有，直接返回，如果没有，创建Registry，之后存入缓存，最后返回。
首先处理传入的registryUrl：
- 设置：path=com.alibaba.dubbo.registry.RegistryService
- 添加参数：interface=com.alibaba.dubbo.registry.RegistryService
- 去除export参数
最终得到的registryUrl如下：
zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&pid=2791×tamp=1507262031521
之后，很具上述的registryUrl创建Registry的key，该{ key : Registry }最终会被存储在Map<String, Registry> REGISTRIES注册中心集合（该属性是ZookeeperRegistryFactory父类AbstractRegistryFactory的一个属性）中。
根据registryUrl创建Registry的key：url.toServiceString()
public String toServiceString() {
return buildString(true, false, true, true);
}
private String buildString(boolean appendUser, boolean appendParameter, boolean useIP, boolean useService, String... parameters) {
StringBuilder buf = new StringBuilder();
if (protocol != null && protocol.length() > 0) { //protocol://
buf.append(protocol);
buf.append("://");
}
if (appendUser && username != null && username.length() > 0) { //protocol://username:password@host:port/group/interface{path}:version/parameters
buf.append(username);
if (password != null && password.length() > 0) {
buf.append(":");
buf.append(password);
}
buf.append("@");
}
String host;
if (useIP) {
host = getIp();
} else {
host = getHost();
}
if (host != null && host.length() > 0) {
buf.append(host);
if (port > 0) {
buf.append(":");
buf.append(port);
}
}
String path;
if (useService) {
path = getServiceKey();
} else {
path = getPath();
}
if (path != null && path.length() > 0) {
buf.append("/");
buf.append(path);
}
if (appendParameter) {
buildParameters(buf, true, parameters);
}
return buf.toString();
}
public String getServiceKey() {
String inf = getServiceInterface();//先获取interface参数，如果没有的话，取path的值，这里都是com.alibaba.dubbo.registry.RegistryService
if (inf == null) return null;
StringBuilder buf = new StringBuilder();
String group = getParameter(Constants.GROUP_KEY);
if (group != null && group.length() > 0) {
buf.append(group).append("/"); //interfacegroup
}
buf.append(inf);
String version = getParameter(Constants.VERSION_KEY);
if (version != null && version.length() > 0) {
buf.append(":").append(version);
}
return buf.toString();
}
最终得到的应该是这样的形式：protocol://username:password@host:port/group/interface{path}:version?key1=value1&key2=value2...。
这里key=zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService
之后来到了真正创建Registry的地方。
public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
private ZookeeperTransporter zookeeperTransporter;
public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
this.zookeeperTransporter = zookeeperTransporter;
}
public Registry createRegistry(URL url) {
return new ZookeeperRegistry(url, zookeeperTransporter);
}
}
这里的zookeeperTransporter对象是一个com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter$Adaptive对象。
在创建ZookeeperRegistry之前来看一下其继承图：
new ZookeeperRegistry(registryUrl, ZookeeperTransporter$Adaptive对象)
private final static int DEFAULT_ZOOKEEPER_PORT = 2181;
private final static String DEFAULT_ROOT = "dubbo";
private final String root;
private final Set<String> anyServices = new ConcurrentHashSet<String>();
private final ConcurrentMap<URL, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<URL, ConcurrentMap<NotifyListener, ChildListener>>();
private final ZookeeperClient zkClient; public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
if (url.isAnyHost()) {
throw new IllegalStateException("registry address == null");
}
String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);//dubbo
if (!group.startsWith(Constants.PATH_SEPARATOR)) {
group = Constants.PATH_SEPARATOR + group;
}
this.root = group;// /dubbo
zkClient = zookeeperTransporter.connect(url);//创建zk客户端，启动会话
zkClient.addStateListener(new StateListener() {//监听重新连接成功事件，重新连接成功后，之前已经完成注册和订阅的url要重新进行注册和订阅（因为临时节点可能已经跪了）
public void stateChanged(int state) {
if (state == RECONNECTED) {
try {
recover();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
});
}
new FailbackRegistry(registryUrl)
private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("DubboRegistryFailedRetryTimer", true));
// 失败重试定时器，定时检查是否有请求失败，如有，无限次重试
private final ScheduledFuture<?> retryFuture;
private final Set<URL> failedRegistered = new ConcurrentHashSet<URL>();
private final Set<URL> failedUnregistered = new ConcurrentHashSet<URL>();
private final ConcurrentMap<URL, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
private final ConcurrentMap<URL, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
private final ConcurrentMap<URL, Map<NotifyListener, List<URL>>> failedNotified = new ConcurrentHashMap<URL, Map<NotifyListener, List<URL>>>();
private AtomicBoolean destroyed = new AtomicBoolean(false);
public FailbackRegistry(URL url) {
super(url);
int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);//5*1000
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);
}
new AbstractRegistry(registryUrl)
// URL地址分隔符，用于文件缓存中，服务提供者URL分隔
private static final char URL_SEPARATOR = ' ';
// URL地址分隔正则表达式，用于解析文件缓存中服务提供者URL列表
private static final String URL_SPLIT = "\\s+";
// 本地磁盘缓存，其中特殊的key值.registies记录注册中心列表，其它均为notified服务提供者列表
private final Properties properties = new Properties();
// 文件缓存定时写入
private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));
//是否是同步保存文件
private final boolean syncSaveFile;
// 本地磁盘缓存文件
private File file;
private final AtomicLong lastCacheChanged = new AtomicLong();
private final Set<URL> registered = new ConcurrentHashSet<URL>();//已经注册的url集合
private final ConcurrentMap<URL, Set<NotifyListener>> subscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();//已经订阅的<URL, Set<NotifyListener>>
private final ConcurrentMap<URL, Map<String, List<URL>>> notified = new ConcurrentHashMap<URL, Map<String, List<URL>>>();//已经通知的<URL, Map<String, List<URL>>>
private URL registryUrl;//注册url
private AtomicBoolean destroyed = new AtomicBoolean(false); public AbstractRegistry(URL url) {
setUrl(url);
// 启动文件保存定时器
syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getHost() + ".cache");
File file = null;
if (ConfigUtils.isNotEmpty(filename)) {
file = new File(filename);
if (!file.exists() && file.getParentFile() != null && !file.getParentFile().exists()) {
if (!file.getParentFile().mkdirs()) {//创建文件所在的文件夹 /Users/jigangzhao/.dubbo/
throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
}
}
}
this.file = file;
loadProperties();
notify(url.getBackupUrls());
}
先简单的总结一下：父子三代分别做的事情：
- AbstractRegistry主要用来维护缓存文件。
- FailbackRegistry主要用来做失败重试操作（包括：注册失败／反注册失败／订阅失败／反订阅失败／通知失败的重试）；也提供了供ZookeeperRegistry使用的zk重连后的恢复工作的方法。
- ZookeeperRegistry创建zk客户端，启动会话；并且调用FailbackRegistry实现zk重连后的恢复工作。
先看AbstractRegistry
- 设置属性registryUrl=url：zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&pid=4685×tamp=1507286468150
- 创建文件/Users/jigangzhao/.dubbo/dubbo-registry-10.211.55.5.cache的文件夹/Users/jigangzhao/.dubbo
- 设置属性file：/Users/jigangzhao/.dubbo/dubbo-registry-10.211.55.5.cache文件，该文件存储信息将是这样的：
com.alibaba.dubbo.demo.DemoService=empty\://10.10.10.10\:20880/com.alibaba.dubbo.demo.DemoService?anyhost\=true&application\=demo-provider&category\=configurators&check\=false&dubbo\=2.0.0&generic\=false&interface\=com.alibaba.dubbo.demo.DemoService&methods\=sayHello&pid\=5259&side\=provider×tamp\=1507294508053
- 如果file存在，将file中的内容写入properties属性；既然有读file，那么是什么时候写入file的呢？AbstractRegistry创建了一个含有一个名字为DubboSaveRegistryCache的后台线程的FixedThreadPool，只在在notify(URL url, NotifyListener listener, List<URL> urls)方法中会被调用，我们此处由于ConcurrentMap<URL, Set<NotifyListener>> subscribed为空，所以AbstractRegistry(URL url)中的notify(url.getBackupUrls())不会执行，此处也不会创建文件。
- 最后是notify(url.getBackupUrls())（TODO 这里后续会写）
再来看FailbackRegistry：
只做了一件事，启动了一个含有一个名为DubboRegistryFailedRetryTimer的后台线程的ScheduledThreadPool，线程创建5s后开始第一次执行retry()，之后每隔5s执行一次。来看一下retry()
/**
* 将所有注册失败的url（failedRegistered中的url）进行注册，之后从failedRegistered进行移除；
* 将所有反注册失败的url（failedUnregistered中的url）进行反注册，之后从failedUnregistered进行移除；
* 将所有订阅失败的url（failedSubscribed中的url）进行重新订阅，之后从failedSubscribed进行移除；
* 将所有反订阅失败的url（failedUnsubscribed中的url）进行反订阅，之后从failedUnsubscribed进行移除；
* 将所有通知失败的url（failedNotified中的url）进行通知，之后从failedNotified进行移除；
*/
protected void retry() {
if (!failedRegistered.isEmpty()) {
Set<URL> failed = new HashSet<URL>(failedRegistered);
if (failed.size() > 0) {
if (logger.isInfoEnabled()) {
logger.info("Retry register " + failed);
}
try {
for (URL url : failed) {
try {
doRegister(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<URL> failed = new HashSet<URL>(failedUnregistered);
if (failed.size() > 0) {
if (logger.isInfoEnabled()) {
logger.info("Retry unregister " + failed);
}
try {
for (URL url : failed) {
try {
doUnregister(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<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedSubscribed);
for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(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<URL, Set<NotifyListener>> entry : failed.entrySet()) {
URL url = entry.getKey();
Set<NotifyListener> listeners = entry.getValue();
for (NotifyListener listener : listeners) {
try {
doSubscribe(url, listener);//listener需要一个一个订阅，每订阅一个，就将该listener从当前的url监听列表中移除
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<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedUnsubscribed);
for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(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<URL, Set<NotifyListener>> entry : failed.entrySet()) {
URL url = entry.getKey();
Set<NotifyListener> listeners = entry.getValue();
for (NotifyListener listener : listeners) {
try {
doUnsubscribe(url, listener);//listener需要一个一个反订阅，每反订阅一个，就将该listener从当前的url监听列表中移除
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<URL, Map<NotifyListener, List<URL>>> failed = new HashMap<URL, Map<NotifyListener, List<URL>>>(failedNotified);
for (Map.Entry<URL, Map<NotifyListener, List<URL>>> entry : new HashMap<URL, Map<NotifyListener, List<URL>>>(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);
}
}
}
}
最后回到我们的主角：ZookeeperRegistry
首先是为属性设置root=/dubbo，之后创建zk客户端，启动会话，最后创建了一个StateListener监听器，监听重新连接成功事件，重新连接成功后，之前已经完成注册和订阅的url要重新进行注册和订阅（因为临时节点可能已经跪了）。
来看创建zk客户端，启动会话的代码，这是此处最核心的部分：
ZookeeperTransporter$Adaptive.connect(com.alibaba.dubbo.common.URL registryUrl)
public com.alibaba.dubbo.remoting.zookeeper.ZookeeperClient connect(com.alibaba.dubbo.common.URL arg0) {
if (arg0 == null)
throw new IllegalArgumentException("url == null");
com.alibaba.dubbo.common.URL url = arg0;
String extName = url.getParameter("client", url.getParameter("transporter", "zkclient"));//curator
if(extName == null)
throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter) name from url(" + url.toString() + ") use keys([client, transporter])");
com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter extension = (com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter.class).getExtension(extName);
return extension.connect(arg0);
}
这里创建的extension是CuratorZookeeperTransporter实例。
public class CuratorZookeeperTransporter implements ZookeeperTransporter {
public ZookeeperClient connect(URL url) {
return new CuratorZookeeperClient(url);
}
}
new CuratorZookeeperClient(registryUrl)
private final CuratorFramework client;
public CuratorZookeeperClient(URL url) {
super(url);
try {
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
.connectString(url.getBackupAddress())
.retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
.connectionTimeoutMs(5000);
String authority = url.getAuthority();
if (authority != null && authority.length() > 0) {
builder = builder.authorization("digest", authority.getBytes());
}
client = builder.build();
client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
public void stateChanged(CuratorFramework client, ConnectionState state) {
if (state == ConnectionState.LOST) {
CuratorZookeeperClient.this.stateChanged(StateListener.DISCONNECTED);
} else if (state == ConnectionState.CONNECTED) {
CuratorZookeeperClient.this.stateChanged(StateListener.CONNECTED);
} else if (state == ConnectionState.RECONNECTED) {
CuratorZookeeperClient.this.stateChanged(StateListener.RECONNECTED);
}
}
});
client.start();
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
这里首先执行父类AbstractZookeeperClient的构造器来初始化一些参数，之后创建CuratorFramework客户端，然后添加了ConnectionStateListener监听器，监听连接断开／连接成功／重新连接成功事件，之后作出相应的操作（实际上这里只有重新连接成功事件会被处理，而处理器实际上就是ZookeeperRegistry构造器中的那个执行recover()的StateListener），
protected void stateChanged(int state) {
for (StateListener sessionListener : getSessionListeners()) {
sessionListener.stateChanged(state);//此处查找实现类，只有ZookeeperRegistry构造器中的那个StateListener
}
}
最后阻塞，直到创建会话完成。
来看一下父类AbstractZookeeperClient：
private final URL url;
private final Set<StateListener> stateListeners = new CopyOnWriteArraySet<StateListener>();
private final ConcurrentMap<String, ConcurrentMap<ChildListener, TargetChildListener>> childListeners = new ConcurrentHashMap<String, ConcurrentMap<ChildListener, TargetChildListener>>();
private volatile boolean closed = false; public AbstractZookeeperClient(URL url) {
this.url = url;
}
说明：
- 设置属性url=registryUrl：zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&pid=4685×tamp=1507286468150
- 创建了一个Set<StateListener> stateListeners，ZookeeperRegistry构造器中的那个执行recover()的StateListener就将会放在这里
至此，一个完整的ZookeeperRegistry实例就创建完成了，来看一下属性：
- ZookeeperClient zkClient = CuratorZookeeperClient实例
- CuratorFramework client：CuratorFrameworkImpl实例
- String url：zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&pid=4685×tamp=1507286468150
- Set<StateListener> stateListeners：{ 监听了重连成功事件的执行recover()的StateListener }
- String root="/dubbo"
- URL registryUrl = zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&client=curator&dubbo=2.0.0&interface=com.alibaba.dubbo.registry.RegistryService&pid=4685×tamp=1507286468150
- Set<URL> registered：0//已经注册的url集合，此处为空
- ConcurrentMap<URL, Set<NotifyListener>> subscribed：0//已经订阅的<URL, Set<NotifyListener>>
- ConcurrentMap<URL, Map<String, List<URL>>> notified：0//已经通知的<URL, Map<String, List<URL>>>
- Set<URL> failedRegistered：0//注册失败的url
- Set<URL> failedUnregistered：0//反注册失败的url
- ConcurrentMap<URL, Set<NotifyListener>> failedSubscribed：0//订阅失败的url
- ConcurrentMap<URL, Set<NotifyListener>> failedUnsubscribed：0//反订阅失败的url
- ConcurrentMap<URL, Map<NotifyListener, List<URL>>> failedNotified：0//通知失败的url
- ConcurrentMap<URL, ConcurrentMap<NotifyListener, ChildListener>> zkListeners：0
还有一个定时线程：DubboRegistryFailedRetryTimer每隔5s执行一次retry()，进行失败重试。
最后，该ZookeeperRegistry会存储在ZookeeperRegistry的父类的static属性Map<String, Registry> REGISTRIES中：
Map<String, Registry> REGISTRIES：{ "zookeeper://10.211.55.5:2181/com.alibaba.dubbo.registry.RegistryService" : ZookeeperRegistry实例 }
二 获取真正要注册到zk的节点url
final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
/**
* 1 获取originInvoker的export参数值：就是providerUrl
* 2 去除providerUrl中所有参数名是"."开头的，然后去除参数monitor
*/
private URL getRegistedProviderUrl(final Invoker<?> originInvoker) {
URL providerUrl = getProviderUrl(originInvoker);
//注册中心看到的地址
final URL registedProviderUrl = providerUrl.removeParameters(getFilteredKeys(providerUrl)).removeParameter(Constants.MONITOR_KEY);
return registedProviderUrl;
} /**
* 从invoker的URL中的Map<String, String> parameters中获取key为export的地址providerUrl：
*/
private URL getProviderUrl(final Invoker<?> origininvoker) {
String export = origininvoker.getUrl().getParameterAndDecoded(Constants.EXPORT_KEY);
if (export == null || export.length() == 0) {
throw new IllegalArgumentException("The registry export url is null! registry: " + origininvoker.getUrl());
}
URL providerUrl = URL.valueOf(export);
return providerUrl;
} //过滤URL中不需要输出的参数(以点号开头的)
private static String[] getFilteredKeys(URL url) {
Map<String, String> params = url.getParameters();
if (params != null && !params.isEmpty()) {
List<String> filteredKeys = new ArrayList<String>();
for (Map.Entry<String, String> entry : params.entrySet()) {
if (entry != null && entry.getKey() != null && entry.getKey().startsWith(Constants.HIDE_KEY_PREFIX)) {
filteredKeys.add(entry.getKey());
}
}
return filteredKeys.toArray(new String[filteredKeys.size()]);
} else {
return new String[]{};
}
}
最后得到的registedProviderUrl是：
- dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=4758&side=provider×tamp=1507289961588
三 注册服务到zk
registry.register(registedProviderUrl);//创建节点（即注册服务到zk上）
这里的registry是ZookeeperRegistry。register(registedProviderUrl)方法在ZookeeperRegistry的父类FailbackRegistry中实现。
1 FailbackRegistry.register(registedProviderUrl)
@Override
public void register(URL url) {
if (destroyed.get()){
return;
}
super.register(url);
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
// 向服务器端发送注册请求
doRegister(url);
} catch (Exception e) {
Throwable t = e;
// 如果开启了启动时检测check=true，则直接抛出异常，不会加入到failedRegistered中
boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
&& url.getParameter(Constants.CHECK_KEY, true)
&& !Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
boolean skipFailback = t instanceof SkipFailbackWrapperException;
if (check || skipFailback) {
if (skipFailback) {
t = t.getCause();
}
throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
} else {
logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
}
// 将失败的注册请求记录到失败列表，定时重试
failedRegistered.add(url);
}
}
首先调用父类AbstractRegistry的register(registedProviderUrl)将当前的registeredProviderUrl放到Set<URL> registered属性中，如下：
public void register(URL url) {
if (url == null) {
throw new IllegalArgumentException("register url == null");
}
if (logger.isInfoEnabled()) {
logger.info("Register: " + url);
}
registered.add(url);
}
之后，从failedRegistered和failedUnregistered两个url集合中删除该url。然后执行真正的服务注册（创建节点，doRegister(url)），如果在创建过程中抛出异常，如果url的协议不是consumer并且开启了check=true的属性并且当前存储的URL registryUrl也有check=true的话，那么直接抛出异常，不会将该url加入到failedRegistered集合；当然抛出的异常如果是SkipFailbackWrapperException，那么也会直接抛出异常，不会将该url加入到failedRegistered集合。否则，会将该url加入到failedRegistered集合，然后DubboRegistryFailedRetryTimer线程会每隔5s执行一次doRegister(url)。
我们来看真正doRegister(url)。
2 ZookeeperRegistry.doRegister(registedProviderUrl)
protected void doRegister(URL url) {
try {
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
首先是对入参registedProviderUrl进行一顿处理，
private String toUrlPath(URL url) {
return toCategoryPath(url) + Constants.PATH_SEPARATOR + URL.encode(url.toFullString());
}
private String toCategoryPath(URL url) {
return toServicePath(url) + Constants.PATH_SEPARATOR + url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
}
private String toServicePath(URL url) {
String name = url.getServiceInterface();
if (Constants.ANY_VALUE.equals(name)) {
return toRootPath();
}
return toRootDir() + URL.encode(name);// /dubbo/com.alibaba.dubbo.demo.DemoService
}
private String toRootDir() {
if (root.equals(Constants.PATH_SEPARATOR)) {
return root;
}
return root + Constants.PATH_SEPARATOR;// /dubbo/
}
private String toRootPath() {
return root;
}
这里就体现了上边的ZookeeperRegistry的root属性的作用。最终实际上得到的是：/dubbo/interface/category/encode过的export，该节点也将是创建在zk上的节点。
- /dubbo是根节点
- /interface是服务接口
- /category是providers／consumers／routers／configurators等
最终得到的url是：
- /dubbo/com.alibaba.dubbo.demo.DemoService/providers/dubbo%3A%2F%2F10.10.10.10%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D5148%26side%3Dprovider%26timestamp%3D1507291294629
- 解码后：/dubbo/com.alibaba.dubbo.demo.DemoService/providers/dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=5148&side=provider×tamp=1507291294629
最后执行zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true))来创建节点，该方法由CuratorZookeeperClient的父类AbstractZookeeperClient来执行：
public void create(String path, boolean ephemeral) {
int i = path.lastIndexOf('/');
if (i > 0) {
create(path.substring(0, i), false);
}
if (ephemeral) {
createEphemeral(path);
} else {
createPersistent(path);
}
}
这里实际上是通过递归分别创建持久化的/dubbo，/dubbo/com.alibaba.dubbo.demo.DemoService以及/dubbo/com.alibaba.dubbo.demo.DemoService/providers节点；最后创建临时节点/dubbo/com.alibaba.dubbo.demo.DemoService/providers/dubbo%3A%2F%2F10.10.10.10%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D5148%26side%3Dprovider%26timestamp%3D1507291294629，而实际上，如果使用了curator的话，可以直接使用递归创建节点即可（结合zk的特性，只有最后一个字节点可以是临时节点，父节点一定是持久化节点），这里这样的写法应该是兼容不能递归创建节点的Zkclient客户端。值得注意的是，url.getParameter(Constants.DYNAMIC_KEY, true)为true则最终创建的节点是临时节点，否则是持久化节点。
创建节点的操作是在CuratorZookeeperClient中进行的。
public void createPersistent(String path) {
try {
client.create().forPath(path);
} catch (NodeExistsException e) {
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
public void createEphemeral(String path) {
try {
client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
} catch (NodeExistsException e) {
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
到此为止，我们去zk上看一下节点的创建情况。
或者从zkui上看一下：
隐藏掉的是ip：10.10.10.10。
到目前为止，我们再来看看ZookeeperRegistry的属性变化。相较于注册前：
- Set<URL> registered：[ dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=5214&side=provider×tamp=1507293238549 ]