【四】dubbo源码分析之服务提供方registry(服务注册)创建注册中心实例向注册中心注册服务

65 阅读 0 评论 43 点赞

我是靠谱客的博主平常外套，最近开发中收集的这篇文章主要介绍【四】dubbo源码分析之服务提供方registry(服务注册)创建注册中心实例向注册中心注册服务，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

服务注册操作对于 Dubbo 来说不是必需的，通过服务直连的方式就可以绕过注册中心。

直连方式不利于服务治理通常只在测试中使用。

前面两章已经讲了spring+dubbo注解方式启动、dubbo服务方暴露，这里直接接着讲服务注册。

示例是注册到zookeeper。

入口在RegistryProtocol.export

@Override
    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        //export invoker
        // 导出服务
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

         // 获取注册中心 URL，以 zookeeper 注册中心为例
        URL registryUrl = getRegistryUrl(originInvoker);

        //registry provider
        //根据 URL 加载 Registry 实现类，比如 ZookeeperRegistry
        //获取 Registry注册中心实例
        final Registry registry = getRegistry(originInvoker);

        // 获取已注册的服务提供者 URL
        final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

        //to judge to delay publish whether or not
        // 获取 register 参数
        boolean register = registeredProviderUrl.getParameter("register", true);

        // 向服务提供者与消费者注册表中注册服务提供者
        ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

        // 根据 register 的值决定是否注册服务
        if (register) {

             // 向注册中心注册服务
            register(registryUrl, registeredProviderUrl);
            ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
        }

        // Subscribe the override data
        // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
        // 获取订阅 URL
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);

       // 创建监听器
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);

        // 向注册中心进行订阅 override 数据
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

        //Ensure that a new exporter instance is returned every time export
         // 创建并返回 DestroyableExporter
        return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
    }

做了几件事

1.暴露invoker，就是服务暴露，doLocalExport方法里面调用了DubboProtocol.export,生成并返回exporter

2.获取注册中心 URL，以 zookeeper 注册中心为例，得到的示例 URL 如下：

zookeeper://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=dubbo-annotation-provider&dubbo=2.0.2&export=dubbo%3A%2F%2FXXX.XXX.XXX.XXX%3A20880%2Forg.apache.dubbo.samples.api.client.HelloService%3Fanyhost%3Dtrue%26application

%3Ddubbo-annotation-provider%26bean.name%3DServiceBean%3Aorg.apache.dubbo.samples.api.client.HelloService%26bind.ip%3DXXX.XXX.XXX.XXX%26bind.port

%3D20880%26default.timeout%3D1000%26dubbo%3D2.0.2%26generic%3Dfalse%26interface%3Dorg.apache.dubbo.samples.api.client.HelloService%26methods

%3DsayHello%26pid%3D392%26side%3Dprovider%26timestamp%3D1603694032387&pid=392&timestamp=1603694031591

3.final Registry registry = getRegistry(originInvoker);

根据 URL 加载 Registry 实现类，比如 ZookeeperRegistry

4.获取已注册的服务提供者 URL，

比如： dubbo://XXX.XXX.XXX.XXX:20880/org.apache.dubbo.samples.api.client.HelloService?anyhost=true&application=dubbo-annotation-provider&bean.name=ServiceBean:org.apache.dubbo.samples.api.client.HelloService&default.timeout=1000

&dubbo=2.0.2&generic=false&interface=org.apache.dubbo.samples.api.client.HelloService&methods=sayHello&pid=392&side=provider&timestamp=1603694032387

5.获取 register 参数

6.向服务提供者与消费者注册表中注册服务提供者

7.根据 register 的值决定是否注册服务 register(registryUrl, registeredProviderUrl);

8.获取订阅 URL，

比如： provider://XXX.XXX.XXX.XXX:20880/org.apache.dubbo.samples.api.client.HelloService?anyhost=true

&application=dubbo-annotation-provider&bean.name=ServiceBean:org.apache.dubbo.samples.api.client.HelloService

&category=configurators&check=false&default.timeout=1000&dubbo=2.0.2&generic=false

&interface=org.apache.dubbo.samples.api.client.HelloService&methods=sayHello&pid=392&side=provider&timestamp=1603694032387

9.创建监听器

10.向注册中心进行订阅 override 数据

11.创建并返回 DestroyableExporter

服务注册的入口既是上面的register方法

RegistryProtocol.register

    public void register(URL registryUrl, URL registedProviderUrl) {

        // 获取 Registry。这里的registryFactory是RegistryFactory$Adaptive
        Registry registry = registryFactory.getRegistry(registryUrl);

       // 注册服务
        registry.register(registedProviderUrl);
    }

这里的registryFactory是RegistryFactory$Adaptive

做了两件事

1.获取 Registry注册中心实例

2.注册服务

创建注册中心实例

先看registryFactory.getRegistry

1.调用接口RegistryFactory

package com.alibaba.dubbo.registry;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.extension.Adaptive;
import com.alibaba.dubbo.common.extension.SPI;

/**
 * RegistryFactory. (SPI, Singleton, ThreadSafe)
 *
 * @see com.alibaba.dubbo.registry.support.AbstractRegistryFactory
 */
@SPI("dubbo")
public interface RegistryFactory {

    /**
     * Connect to the registry
     * <p>
     * Connecting the registry needs to support the contract: <br>
     * 1. When the check=false is set, the connection is not checked, otherwise the exception is thrown when disconnection <br>
     * 2. Support username:password authority authentication on URL.<br>
     * 3. Support the backup=10.20.153.10 candidate registry cluster address.<br>
     * 4. Support file=registry.cache local disk file cache.<br>
     * 5. Support the timeout=1000 request timeout setting.<br>
     * 6. Support session=60000 session timeout or expiration settings.<br>
     *
     * @param url Registry address, is not allowed to be empty
     * @return Registry reference, never return empty value
     */
    @Adaptive({"protocol"})
    Registry getRegistry(URL url);

}

2.ExtensionLoader.getExtensionLoader()

入参type=interface com.alibaba.dubbo.registry.RegistryFactory

返回loader=com.alibaba.dubbo.common.extension.ExtensionLoader[com.alibaba.dubbo.registry.RegistryFactory]

3.ExtensionLoader.getExtension()

入参name=zookeeper

返回instance=ZookeeperRegistryFactory

4.AbstractRegistryFactory.getRegistry

 @Override
    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 the registry access process to ensure a single instance of the registry
        LOCK.lock();
        try {

            // 访问缓存
            Registry registry = REGISTRIES.get(key);
            if (registry != null) {
                return registry;
            }

            // 缓存未命中，创建 Registry 实例
            registry = createRegistry(url);
            if (registry == null) {
                throw new IllegalStateException("Can not create registry " + url);
            }

             // 写入缓存
            REGISTRIES.put(key, registry);
            return registry;
        } finally {
            // Release the lock
            LOCK.unlock();
        }
    }

如果缓存中没有则调用子类的createRegistry方法创建Registry

但是我debug的时候缓存中有，就直接返回了。后来再debug发现创建Registry实例是在RegistryProtocol.export方法的final Registry registry = getRegistry(originInvoker);中做的

ZookeeperRegistryFactory.createRegistry

public class ZookeeperRegistryFactory extends AbstractRegistryFactory {

    // zookeeperTransporter 由 SPI 在运行时注入，类型为 ZookeeperTransporter$Adaptive
    private ZookeeperTransporter zookeeperTransporter;

    public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
        this.zookeeperTransporter = zookeeperTransporter;
    }

    @Override
    public Registry createRegistry(URL url) {
        return new ZookeeperRegistry(url, zookeeperTransporter);
    }

}

看ZookeeperRegistry的构造函数

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("registry address == null");
        }

        // 获取组名，默认为 dubbo
        String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
        if (!group.startsWith(Constants.PATH_SEPARATOR)) {

            // group = "/" + group
            group = Constants.PATH_SEPARATOR + group;
        }
        this.root = group;

       // 创建 Zookeeper 客户端，默认为 CuratorZookeeperTransporter
        zkClient = zookeeperTransporter.connect(url);

       // 添加状态监听器
        zkClient.addStateListener(new StateListener() {
            @Override
            public void stateChanged(int state) {
                if (state == RECONNECTED) {
                    try {
                        recover();
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        });
    }

做了几件事

1.获取组名，默认为 dubbo

2.zookeeperTransporter.connect(url)创建 Zookeeper 客户端，默认为 CuratorZookeeperTransporter

3.添加状态监听器

ZookeeperTransporter接口SPI扩展

@SPI("curator")
public interface ZookeeperTransporter {

    @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY})
    ZookeeperClient connect(URL url);

}

ExtensionLoader.getExtensionLoader（）

入参type=interface com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter

出参loader=com.alibaba.dubbo.common.extension.ExtensionLoader[com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter]

ExtensionLoader.getExtension()

入参name=curator

出参instance=CuratorZookeeperTransporter

zookeeperTransporter.connect()

package com.alibaba.dubbo.remoting.zookeeper.curator;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.remoting.zookeeper.ZookeeperClient;
import com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter;

public class CuratorZookeeperTransporter implements ZookeeperTransporter {

    @Override
    public ZookeeperClient connect(URL url) {
        return new CuratorZookeeperClient(url);
    }

}

CuratorZookeeperClient的构造函数

public CuratorZookeeperClient(URL url) {
        super(url);
        try {

            // 创建 CuratorFramework 构造器
            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                    .connectString(url.getBackupAddress())
                    .retryPolicy(new RetryNTimes(1, 1000))
                    .connectionTimeoutMs(5000);
            String authority = url.getAuthority();
            if (authority != null && authority.length() > 0) {
                builder = builder.authorization("digest", authority.getBytes());
            }

            // 构建 CuratorFramework 实例
            client = builder.build();

             // 添加监听器
            client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                @Override
                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);
        }
    }

做了几件事：

1.创建 CuratorFramework 构造器

2.构建 CuratorFramework 实例

3.添加监听器

4.启动客户端

现在注册中心实例创建好了，接下来要做的事情是向注册中心注册服务。

向注册中心注册服务

RegistryProtocol.register方法中的registry.register(registedProviderUrl);

这里的registry是FailbackRegistry

FailbackRegistry.register()

@Override
    public void register(URL url) {
        super.register(url);
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try {
            // Sending a registration request to the server side
           // 模板方法，由子类实现
            doRegister(url);
        } catch (Exception e) {
            Throwable t = e;

            // If the startup detection is opened, the Exception is thrown directly.
             // 获取 check 参数，若 check = true 将会直接抛出异常
            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);
            }

            // Record a failed registration request to a failed list, retry regularly
            // 记录注册失败的链接
            failedRegistered.add(url);
        }
    }

ZookeeperRegistry.doRegister()

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

做了一件事

通过 Zookeeper 客户端创建节点，节点路径由 toUrlPath 方法生成。

路径格式如下:

/dubbo/org.apache.dubbo.samples.api.client.HelloService/providers/dubbo%3A%2F%2FXXX.XXX.XXX.XXX%3A20880%2Forg.apache.dubbo.samples.api.client.HelloService

%3Fanyhost%3Dtrue%26application%3Ddubbo-annotation-provider%26bean.name%3DServiceBean%3Aorg.apache.dubbo.samples.api.client.HelloService%26default.timeout%3D1000%26dubbo

%3D2.0.2%26generic%3Dfalse%26interface%3Dorg.apache.dubbo.samples.api.client.HelloService%26methods%3DsayHello%26pid%3D11324%26side%3Dprovider%26timestamp%3D1603701475951

AbstractZookeeperClient.create()方法

 @Override
    public void create(String path, boolean ephemeral) {
        if (!ephemeral) {

            // 如果要创建的节点类型非临时节点，那么这里要检测节点是否存在
            if (checkExists(path)) {
                return;
            }
        }
        int i = path.lastIndexOf('/');
        if (i > 0) {

            // 递归创建上一级路径
            create(path.substring(0, i), false);
        }

        // 根据 ephemeral 的值创建临时或持久节点
        if (ephemeral) {
            createEphemeral(path);
        } else {
            createPersistent(path);
        }
    }

做了几件事：