dubbo原始碼解析之服務export 遠端服務②

阿新 • • 發佈：2019-10-07

前期回顧

sharding-jdbc原始碼解析更新完畢

spring原始碼解析更新完畢

spring-mvc原始碼解析更新完畢

spring-tx原始碼解析更新完畢

spring-boot原始碼解析更新完畢

rocketmq原始碼解析更新完畢

dubbo原始碼解析更新中

sharding-sphere原始碼解析計劃中

netty原始碼解析計劃中

spring-cloud-alibaba-dubbo原始碼解析計劃中

github https://github.com/tianheframe

sharding-jdbc原始碼解析更新完畢

rocketmq 更新完畢

dubbo原始碼解析更新中

seata原始碼解析更新中

spring-cloud-tianhe 更新中

mq-tianhe 計劃中

rpc-tianhe 計劃中

原始碼解析

https://github.com/tianheframe/dubbo.git dubbo原始碼解析後的原始碼和公號文章同步更新。

返回到這個方法，服務註冊資訊恢復com.alibaba.dubbo.registry.support.FailbackRegistry#recover

@Override    protected void recover() throws Exception { 
        // 獲取服務註冊的url集合        Set<URL> recoverRegistered = new HashSet<URL>(getRegistered());        if (!recoverRegistered.isEmpty()) {            if (logger.isInfoEnabled()) {                logger.info("Recover register url " + recoverRegistered);            }            for (URL url : recoverRegistered) { 
//                儲存註冊失敗的服務註冊url                failedRegistered.add(url);            }        }        // 獲取訂閱的服務url集合        Map<URL, Set<NotifyListener>> recoverSubscribed = new HashMap<URL, Set<NotifyListener>>(getSubscribed());        if (!recoverSubscribed.isEmpty()) {            if (logger.isInfoEnabled()) {                logger.info("Recover subscribe url " + recoverSubscribed.keySet());            }            for (Map.Entry<URL, Set<NotifyListener>> entry : recoverSubscribed.entrySet()) {                URL url = entry.getKey();                for (NotifyListener listener : entry.getValue()) {//                    新增訂閱失敗的服務url                    addFailedSubscribed(url, listener);                }            }        }    }

進入這個方法com.alibaba.dubbo.registry.support.FailbackRegistry#register服務註冊

@Override    public void register(URL url) {//        添加註冊服務url=》        super.register(url);        failedRegistered.remove(url);        failedUnregistered.remove(url);        try {            // Sending a registration request to the server side 向伺服器端傳送註冊請求=》ZookeeperRegistry            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)                    && !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);        }    }

進入這個方法com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry#doRegister進行zk服務註冊，這裡的邏輯就是建立zk臨時節點

@Override    protected void doRegister(URL url) {        try {//            服務註冊，建立zk節點，如果dynamic配置的是true，建立的就是zk臨時節點=》            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);        }    }

這裡預設建立的是臨時節點，這也就是zk註冊的服務所在節點掛了之後其他客戶端節點本地的服務列表會更新的原因，不會呼叫到不存在的服務，當然也存在zk臨時節點刪除，通知其他訂閱這個節點的客戶端時候出現網路抖動，zk會做處理確保一定能通知到，這種中間處理也能要業務邏輯要做處理了

/dubbo/com.alibaba.dubbo.demo.DemoService/providers/dubbo://172.28.84.147:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bean.name=com.alibaba.dubbo.demo.DemoService&dubbo=2.0.2&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=76579&side=provider&timestamp=1569898563184 服務註冊的zk path是這樣的

如果註冊失敗的話怎麼辦呢，在建立com.alibaba.dubbo.registry.support.FailbackRegistry#FailbackRegistry物件的時候構造方法邏輯中，重試引數retry.period 預設值是每5秒鐘會做重試處理，這裡也可以自定義修改

public FailbackRegistry(URL url) {        super(url);//        每5秒中會重試註冊失敗的服務資訊，可以修改這個引數retry.period        this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);        this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {            @Override            public void run() {                // Check and connect to the registry                try {                    retry();                } catch (Throwable t) { // Defensive fault tolerance                    logger.error("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);                }            }        }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);    }

進入這個方法服務註冊失敗重試邏輯，com.alibaba.dubbo.registry.support.FailbackRegistry#retry

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 {//                            zk服務註冊                            doRegister(url);                            failedRegistered.remove(url);                        } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry 這裡異常不做處理等待下次重試                            logger.warn("Failed to retry register " + failed + ", waiting for again, cause: " + t.getMessage(), t);                        }                    }                } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                    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.isEmpty()) {                if (logger.isInfoEnabled()) {                    logger.info("Retry unregister " + failed);                }                try {                    for (URL url : failed) {                        try {//                            取消服務註冊失敗重試                            doUnregister(url);                            failedUnregistered.remove(url);                        } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                            logger.warn("Failed to retry unregister  " + failed + ", waiting for again, cause: " + t.getMessage(), t);                        }                    }                } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                    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);                                listeners.remove(listener);                            } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                                logger.warn("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);                            }                        }                    }                } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                    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().isEmpty()) {                    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);                                listeners.remove(listener);                            } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                                logger.warn("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);                            }                        }                    }                } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                    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) { // Ignore all the exceptions and wait for the next retry                                logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);                            }                        }                    }                } catch (Throwable t) { // Ignore all the exceptions and wait for the next retry                    logger.warn("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);                }            }        }    }

com.alibaba.dubbo.registry.support.FailbackRegistry#failedRegistered對服務註冊失敗的重新註冊

對服務取消註冊失敗的進行重新取消服務註冊com.alibaba.dubbo.registry.support.FailbackRegistry#failedUnregistered，進入這個方法com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistry#doUnregister，這裡的操作就是刪除刪除zk臨時節點，刪除zk臨時節點後其他訂閱服務的服務節點會收到zk的監聽器重新重新整理已經生成的代理invoker物件，客戶端在進行負載均衡的時候是直接路由到具體的invoker

 @Override    protected void doUnregister(URL url) {        try {            zkClient.delete(toUrlPath(url));        } catch (Throwable e) {            throw new RpcException("Failed to unregister " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);        }    }

訂閱失敗的重新訂閱com.alibaba.dubbo.registry.support.FailbackRegistry#failedSubscribed

下篇接著介紹服務訂閱失敗的怎麼進行重新服務訂閱

說在最後

本次解析僅代表個人觀點，僅供

dubbo原始碼解析之服務export 遠端服務②

dubbo原始碼解析之服務export 遠端服務②

Dubbo原始碼解析之服務端接收訊息

Dubbo原始碼解析之服務釋出與註冊

Dubbo原始碼解析之客戶端初始化及服務呼叫

dubbo原始碼淺析（五）-遠端服務呼叫流程

Dubbo原始碼解析之LoadBalance負載均衡

Dubbo原始碼解析之consumer關聯provider

Dubbo原始碼解析之consumer呼叫篇

Dubbo原始碼解析之provider呼叫篇

Dubbo原始碼解析之consumer初始化

Dubbo原始碼解析之SPI（一）：擴充套件類的載入過程

dubbo原始碼解析之負載均衡

Dubbo原理和原始碼解析之服務暴露

Dubbo原理和原始碼解析之服務引用

9、dubbo原始碼分析之服務遠端暴露(中)

Dubbo原始碼分析之四：服務的呼叫

分散式任務排程平臺XXL-JOB--原始碼解析六:executor執行器原始碼解析之啟動jetty服務接受請求並執行排程

Istio技術與實踐01：原始碼解析之Pilot多雲平臺服務發現機制

Istio技術與實踐02：原始碼解析之Istio on Kubernetes 統一服務發現

7、dubbo原始碼分析之服務本地暴露

dubbo原始碼解析之服務export 遠端服務②

相關推薦