dubbo源码分析-服务端注册流程-笔记

原创
2018/12/03 14:52
阅读数 3.4K

前面,我们已经知道,基于spring这个解析入口,到发布服务的过程,接着基于DubboProtocol去发布,最终调用Netty的api创建了一个NettyServer。

那么继续沿着RegistryProtocol.export这个方法,来看看注册服务的代码:

RegistryProtocol.export

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
    //export invoker
    final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); //发布本地服务
    //registry provider
    final Registry registry = getRegistry(originInvoker);
    final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
    registry.register(registedProviderUrl);
    // 订阅override数据
    // FIXME 提供者订阅时,会影响同一JVM即暴露服务,又引用同一服务的的场景,因为subscribed以服务名为缓存的key,导致订阅信息覆盖。
    final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
    final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
    overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
    registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    //保证每次export都返回一个新的exporter实例
    return new Exporter<T>() {
        public Invoker<T> getInvoker() {
            return exporter.getInvoker();
        }
        public void unexport() {
           try {
              exporter.unexport();
           } catch (Throwable t) {
               logger.warn(t.getMessage(), t);
            }
            try {
               registry.unregister(registedProviderUrl);
            } catch (Throwable t) {
               logger.warn(t.getMessage(), t);
            }
            try {
               overrideListeners.remove(overrideSubscribeUrl);
               registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
            } catch (Throwable t) {
               logger.warn(t.getMessage(), t);
            }
        }
    };
}

getRegistry

  • 这个方法是invoker的地址获取registry实例
/**
 * 根据invoker的地址获取registry实例
 * @param originInvoker
 * @return
 */
private Registry getRegistry(final Invoker<?> originInvoker){
    URL registryUrl = originInvoker.getUrl(); //获得registry://192.168.11.156:2181的协议地址
    if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) {
//得到zookeeper的协议地址
        String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
        //registryUrl就会变成了zookeeper://192.168.11.156
registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
    }
//registryFactory是什么?
    return registryFactory.getRegistry(registryUrl);
}

registryFactory.getRegistry

  • 这段代码很明显了,通过前面这段代码的分析,其实就是把registry的协议头改成服务提供者配置的协议地址,也就是我们配置的
  • <dubbo:registry address=”zookeeper://192.168.11.156:2181”/>
  • 然后registryFactory.getRegistry的目的,就是通过协议地址匹配到对应的注册中心。
  • 那registryFactory是一个什么样的对象呢?我们找一下这个代码的定义
private RegistryFactory registryFactory;

public void setRegistryFactory(RegistryFactory registryFactory) {
    this.registryFactory = registryFactory;
}

  • 这个代码有点眼熟,再来看看RegistryFactory这个类的定义,我猜想一定是一个扩展点,不信,咱们看
  • 并且,大家还要注意这里面的一个方法上,有一个@Adaptive的注解,说明什么? 这个是一个自适应扩展点。
  • 按照我们之前看过代码,自适应扩展点加在方法层面上,表示会动态生成一个自适应的适配器。
  • 所以这个自适应适配器应该是RegistryFactory$Adaptive
@SPI("dubbo")
public interface RegistryFactory {

    /**
     * 连接注册中心.
     * 
     * 连接注册中心需处理契约:<br>
     * 1. 当设置check=false时表示不检查连接,否则在连接不上时抛出异常。<br>
     * 2. 支持URL上的username:password权限认证。<br>
     * 3. 支持backup=10.20.153.10备选注册中心集群地址。<br>
     * 4. 支持file=registry.cache本地磁盘文件缓存。<br>
     * 5. 支持timeout=1000请求超时设置。<br>
     * 6. 支持session=60000会话超时或过期设置。<br>
     * 
     * @param url 注册中心地址,不允许为空
     * @return 注册中心引用,总不返回空
     */
    @Adaptive({"protocol"})
    Registry getRegistry(URL url);

}

RegistryFactory$Adaptive

我们拿到这个动态生成的自适应扩展点,看看这段代码里面的实现

  1. 从url中拿到协议头信息,这个时候的协议头是zookeeper://
  2. 通过ExtensionLoader.getExtensionLoader(RegistryFactory.class).getExtension(“zookeeper”)去获得一个指定的扩展点,而这个扩展点的配置在
    • dubbo-registry-zookeeper/resources/META-INF/dubbo/internal/com.alibaba.dubbo.registry.RegistryFactory
    • 得到一个ZookeeperRegistryFactory
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());
        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);
    }
}

ZookeeperRegistryFactory

这个方法中并没有getRegistry方法,而是在父类AbstractRegistryFactory

  1. 从缓存REGISTRIES中,根据key获得对应的Registry
  2. 如果不存在,则创建Registry
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();
    }
}

createRegistry

创建一个注册中心,这个是一个抽象方法,具体的实现在对应的子类实例中实现的,在ZookeeperRegistryFactory中

public Registry createRegistry(URL url) {
       return new ZookeeperRegistry(url, zookeeperTransporter);
   }
通过zkClient,获得一个zookeeper的连接实例
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);
    if (! group.startsWith(Constants.PATH_SEPARATOR)) {
        group = Constants.PATH_SEPARATOR + group;
    }
    this.root = group; //设置根节点
    zkClient = zookeeperTransporter.connect(url);//建立连接
      zkClient.addStateListener(new StateListener() {
        public void stateChanged(int state) {
           if (state == RECONNECTED) {
           try {
      recover();
   } catch (Exception e) {
      logger.error(e.getMessage(), e);
   }
           }
        }
    });
}
  • 代码分析到这里,我们对于getRegistry得出了一个结论,根据当前注册中心的配置信息,获得一个匹配的注册中心,也就是ZookeeperRegistry

registry.register(registedProviderUrl);

  • 继续往下分析,会调用registry.register去将dubbo://的协议地址注册到zookeeper上
  • 这个方法会调用FailbackRegistry类中的register. 为什么呢?
    • 因为ZookeeperRegistry这个类中并没有register这个方法,但是他的父类FailbackRegistry中存在register方法,而这个类又重写了AbstractRegistry类中的register方法。
    • 所以我们可以直接定位大FailbackRegistry这个类中的register方法中

FailbackRegistry.register

  1. FailbackRegistry,从名字上来看,是一个失败重试机制
  2. 调用父类的register方法,讲当前url添加到缓存集合中
  3. 调用doRegister方法,这个方法很明显,是一个抽象方法,会由ZookeeperRegistry子类实现。
@Override
public void register(URL url) {
    super.register(url);
    failedRegistered.remove(url);
    failedUnregistered.remove(url);
    try {
        // 向服务器端发送注册请求
        doRegister(url);
    } catch (Exception e) {
        Throwable t = e;

        // 如果开启了启动时检测,则直接抛出异常
        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);
    }
}

ZookeeperRegistry.doRegister

终于找到你了,调用zkclient.create在zookeeper中创建一个节点。


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

  • RegistryProtocol.export 这个方法中后续的代码就不用再分析了。
  • 就是去对服务提供端去注册一个zookeeper监听,当监听发生变化的时候,服务端做相应的处理。

在register 方法里面,调用subscribe 方法,订阅注册中心变化

    /**
     * 订阅符合条件的已注册数据,当有注册数据变更时自动推送.
     * 
     * 订阅需处理契约:<br>
     * 1. 当URL设置了check=false时,订阅失败后不报错,在后台定时重试。<br>
     * 2. 当URL设置了category=routers,只通知指定分类的数据,多个分类用逗号分隔,并允许星号通配,表示订阅所有分类数据。<br>
     * 3. 允许以interface,group,version,classifier作为条件查询,如:interface=com.alibaba.foo.BarService&version=1.0.0<br>
     * 4. 并且查询条件允许星号通配,订阅所有接口的所有分组的所有版本,或:interface=*&group=*&version=*&classifier=*<br>
     * 5. 当注册中心重启,网络抖动,需自动恢复订阅请求。<br>
     * 6. 允许URI相同但参数不同的URL并存,不能覆盖。<br>
     * 7. 必须阻塞订阅过程,等第一次通知完后再返回。<br>
     * 
     * @param url 订阅条件,不允许为空,如:consumer://10.20.153.10/com.alibaba.foo.BarService?version=1.0.0&application=kylin
     * @param listener 变更事件监听器,不允许为空
     */
    void subscribe(URL url, NotifyListener listener);
  • subscribe ->doSubscribe ->notify ->  
    protected void notify(URL url, NotifyListener listener, List<URL> urls) {
        if (url == null) {
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("notify listener == null");
        }
        if ((urls == null || urls.size() == 0) 
                && ! Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            logger.warn("Ignore empty notify urls for subscribe url " + url);
            return;
        }
        if (logger.isInfoEnabled()) {
            logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
        }
        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) {
            if (UrlUtils.isMatch(url, u)) {
            	String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            	List<URL> categoryList = result.get(category);
            	if (categoryList == null) {
            		categoryList = new ArrayList<URL>();
            		result.put(category, categoryList);
            	}
            	categoryList.add(u);
            }
        }
        if (result.size() == 0) {
            return;
        }
        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) {
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        }
        // 第一次主动调用 notify
        // 对  /router    /providers    /configerations 路径下的变更  进行notify
        //后续(zookeeper watcher 机制)
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            saveProperties(url);
            listener.notify(categoryList);
        }
    }
展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部