技術標籤：dubbo

目錄

getRegistry

registryFactory

createRegistry

ZookeeperRegistry

ZookeeperRegistry.doRegister

在之前的釋出流程的RegistryProtocol.export中看到了服務註冊流程。

從export方法中抽離出來的部分程式碼。

@Override
    public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
    ...
    // 把dubbo:// url註冊到zk上
        final Registry registry = getRegistry(originInvoker);
        final URL registeredProviderUrl = getRegisteredProviderUrl(providerUrl, registryUrl);
        ProviderInvokerWrapper<T> providerInvokerWrapper = ProviderConsumerRegTable.registerProvider(originInvoker,
                registryUrl, registeredProviderUrl);
        //to judge if we need to delay publish
        boolean register = registeredProviderUrl.getParameter("register", true);
        if (register) {
            register(registryUrl, registeredProviderUrl);
            providerInvokerWrapper.setReg(true);
        }
    ...

}
 

getRegistry

private Registry getRegistry(final Invoker<?> originInvoker) {
        URL registryUrl = getRegistryUrl(originInvoker);
        return registryFactory.getRegistry(registryUrl);
    }

首先把url轉化為配置的具體協議，例如zookeeper://ip:port. 這樣後續獲得的註冊中心就會是基於zk的實現

然後根據具體協議，從registryFactory中獲得指定的註冊中心實現

registryFactory

根據RegistryProtocol中的程式碼可以看出，是通過依賴注入來實現擴充套件點

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

在/META-INF/dubbo/internal路徑下找到RegistryFactory的配置檔案中找到擴充套件點實現，org.apache.dubbo.registry.zookeeper.ZookeeperRegistryFactory。

getRegistry(URL url)方法其實是在父類AbstractRegistryFactory中

@Override
    public Registry getRegistry(URL url) {
        url = URLBuilder.from(url)
                .setPath(RegistryService.class.getName())
                .addParameter(INTERFACE_KEY, RegistryService.class.getName())
                .removeParameters(EXPORT_KEY, REFER_KEY)
                .build();
        String key = url.toServiceStringWithoutResolving();
        // 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;
            }
            //create registry by spi/ioc
            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

建立一個zookeeperRegistry，把url和zookeepertransporter作為引數傳入。

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

ZookeeperRegistry

ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter)

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("registry address == null");
        }
        String group = url.getParameter(GROUP_KEY, DEFAULT_ROOT);
        if (!group.startsWith(PATH_SEPARATOR)) {
            group = PATH_SEPARATOR + group;
        }
        this.root = group;
        //產生一個zookeeper連線
        zkClient = zookeeperTransporter.connect(url);
        //新增zookeeper狀態變化事件
        zkClient.addStateListener(state -> {
            if (state == StateListener.RECONNECTED) {
                try {
                    recover();
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        });
    }

再回到RegistryProtocol.export，進入register(registryUrl, registeredProviderUrl)，會發現registry.register(registeredProviderUrl)

根據上面的分析可知registryFactory.getRegistry(registryUrl)返回的是ZookeeperRegistry，那麼就去ZookeeperRegistry中找register方法，發現找不到，那就去ZookeeperRegistry的父類FailbackRegistry中尋找。

@Override
    public void register(URL url) {
        super.register(url);
        removeFailedRegistered(url);
        removeFailedUnregistered(url);
        try {
            // Sending a registration request to the server side
            // 呼叫子類實現真正的服務註冊，把url註冊到zk上
            doRegister(url);
        } catch (Exception e) {
            Throwable t = e;

            // If the startup detection is opened, the Exception is thrown directly.
            boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                    && url.getParameter(Constants.CHECK_KEY, true)
                    && !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
            // 將失敗的註冊請求記錄到失敗列表，定時重試
            addFailedRegistered(url);
        }
    }

ZookeeperRegistry.doRegister

把服務地址註冊到zk

@Override
    public void doRegister(URL url) {
        try {
            zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
        } catch (Throwable e) {
            throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }