2. 程式人生 > >Dubbo Provider啟動流程原始碼分析

Dubbo Provider啟動流程原始碼分析

阿新 • • 發佈：2019-02-19

簡單的官方demo：

provider的java程式碼：

public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-provider.xml"});
        context.start();

        System.in.read(); // 按任意鍵退出
    }

provider的spring配置：

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
       xmlns="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
    http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd"
 
>

    <!-- 提供方應用資訊，用於計算依賴關係 -->
    <dubbo:application name="demo-provider"/>

    <!-- 使用multicast廣播註冊中心暴露服務地址 -->
    <dubbo:registry address="multicast://224.5.6.7:1234"/>

    <!-- 用dubbo協議在20880埠暴露服務 -->
    <dubbo:protocol name="dubbo" port="20880"/>

    <!-- 和本地bean一樣實現服務 -->
 

    <bean id="demoService" class="com.alibaba.dubbo.demo.provider.DemoServiceImpl"/>

    <!-- 宣告需要暴露的服務介面 -->
    <dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService"/>

</beans>

通過配置檔案，以及spring的mvc初始化流程，我們可以假設下服務啟動流程：
1. spring配置檔案的載入以及初始化
2. bean的單例生成
3. dubbo管理所有服務實現單例物件
4. 向配置中心註冊服務資訊

spring自定義標籤
dubbo標籤初始化實現了Spring提供的NamespaceHandler介面，所以下面先看看DubboNamespaceHandler類：

public class DubboNamespaceHandler extends NamespaceHandlerSupport {

    static {
        Version.checkDuplicate(DubboNamespaceHandler.class);
    }

    public void init() {
        // DubboBeanDefinitionParser定義瞭如何解析dubbo節點資訊
        // DubboBeanDefinitionParser的第一個引數是beanclass

        // 應用相關配置
        registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
        // 模組相關配置
        registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
        // 註冊中心相關配置
        registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
        // 
        registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
        // 服務提供者配置
        registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
        // 服務消費者配置
        registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
        // 網路協議相關配置
        registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
        // 服務bean配置
        registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
        // 
        registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
        //
        registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
    }

}

在DubboBeanDefinitionParser中的parse中，解析設定了大部分配置資訊以及服務資訊。
我們可以關注下其中beanclass的原始碼，因為這章主要分析的是provider，這裡從provider進行分析：
首先是beanName叫做ServiceBean的bean例項。

public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener, BeanNameAware {
    ...
    // 初始化bean的時候執行
    public void afterPropertiesSet() throws Exception {
        ...// 初始化各種配置
        // 釋出服務
        if (!isDelay()) {
            export();
        }
    }
}

釋出程式碼：

    public synchronized void export() {
        ...
        // 延遲匯出
        if (delay != null && delay > 0) {
            delayExportExecutor.schedule(new Runnable() {
                public void run() {
                    doExport();
                }
            }, delay, TimeUnit.MILLISECONDS);
        } else {
            doExport();
        }
    }

匯出程式碼：

   protected synchronized void doExport() {
        ...
        doExportUrls();
    }
    private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

    // method => invoker => exporter
    private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
        ...
        // 匯出服務
        String contextPath = protocolConfig.getContextpath();
        if ((contextPath == null || contextPath.length() == 0) && provider != null) {
            contextPath = provider.getContextpath();
        }

        // 註冊中心可以是zk，consul等
        // 註冊中心host
        String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
        // 註冊中心port
        Integer port = this.findConfigedPorts(protocolConfig, name, map);
        URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

        // 獲取暴露範圍配置
        String scope = url.getParameter(Constants.SCOPE_KEY);
        if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
            // 如果暴露本地服務
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
                exportLocal(url);
            }
            // 如果暴露遠端服務，走服務發現流程
            if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
                if (registryURLs != null && registryURLs.size() > 0 && url.getParameter("register", true)) {
                    for (URL registryURL : registryURLs) {
                        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
                        // 註冊地址
                        URL monitorUrl = loadMonitor(registryURL);
                        // 動態代理，將class+method包裝位invoker，ref是服務的具體例項物件obj，invoker是個可執行物件
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                        // invoker=>exporter，最終在服務端儲存下來的是exporter，對服務的暴露和引用都是通過這個物件實現的，而這個物件的實現由協議決定
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                } else {
                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
                    Exporter<?> exporter = protocol.export(invoker);
                    exporters.add(exporter);
                }
            }
        }
        this.urls.add(url);
    }

invoker 生成，動態代理的過程

定義程式碼：

ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

@SPI("javassist") //使用javassist位元組碼技術生成物件
public interface ProxyFactory {

    /**
     * create proxy.
     * 定義生成代理物件的方法
     *
     * @param invoker
     * @return proxy
     */
    @Adaptive({Constants.PROXY_KEY})
    <T> T getProxy(Invoker<T> invoker) throws RpcException;

    /**
     * create invoker.
     * getProxy呼叫的引數生成
     *
     * @param <T>
     * @param proxy
     * @param type
     * @param url
     * @return invoker
     */
    @Adaptive({Constants.PROXY_KEY})
    <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException;

}

// jdk 動態代理
public class JdkProxyFactory extends AbstractProxyFactory {

    @SuppressWarnings("unchecked")
    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), interfaces, new InvokerInvocationHandler(invoker));
    }

    public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                Method method = proxy.getClass().getMethod(methodName, parameterTypes);
                return method.invoke(proxy, arguments);
            }
        };
    }

}

// javassist動態代理
public class JavassistProxyFactory extends AbstractProxyFactory {

    @SuppressWarnings("unchecked")
    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }

    public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper類不能正確處理帶$的類名
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }

}

public abstract class AbstractProxyInvoker<T> implements Invoker<T> {

    private final T proxy; //在proxyFactory.getInvoker的時候被設定，即

    private final Class<T> type;

    private final URL url;

    public Result invoke(Invocation invocation) throws RpcException {
        try {
            return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
        } catch (InvocationTargetException e) {
            return new RpcResult(e.getTargetException());
        } catch (Throwable e) {
            throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

    // 呼叫的是jdkproxyfactory和javassistproxyfactory定義的方法
    protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable;

}

實際函式呼叫：

// 泛型呼叫
public class ExtensionLoader<T> {
    // 快取
    private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<Class<?>, ExtensionLoader<?>>();
    private static final ConcurrentMap<Class<?>, Object> EXTENSION_INSTANCES = new ConcurrentHashMap<Class<?>, Object>();
    // class->value
    private final Holder<Object> cachedAdaptiveInstance = new Holder<Object>();

    // 獲取loader，優先快取
    public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
        ExtensionLoader<T> loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
        if (loader == null) {
            EXTENSION_LOADERS.putIfAbsent(type, new ExtensionLoader<T>(type));
            loader = (ExtensionLoader<T>) EXTENSION_LOADERS.get(type);
        }
        return loader;
    }
    // 每個class綁定了一個value，獲取這個value，優先快取
    public T getAdaptiveExtension() {
        Object instance = cachedAdaptiveInstance.get();
        if (instance == null) {
            if (createAdaptiveInstanceError == null) {
                synchronized (cachedAdaptiveInstance) {
                    instance = cachedAdaptiveInstance.get();
                    if (instance == null) {
                            instance = createAdaptiveExtension();
                            cachedAdaptiveInstance.set(instance);
                    }
                }
            }
        }
        return (T) instance;
    }
}

exporter 的生成

抽象類

public abstract class AbstractProtocol implements Protocol {

    // 一個協議對應著多個exporter
    protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>();

    // 對應著一堆invoker
    protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();
}

對應的實現類
以thrift作為交換資料協議為例

public class ThriftProtocol extends AbstractProtocol {

    // thrift port
    public static final int DEFAULT_PORT = 40880;

    // 對應的資料交換方
    // ip:port -> ExchangeServer
    private final ConcurrentMap<String, ExchangeServer> serverMap =
            new ConcurrentHashMap<String, ExchangeServer>();

    // 服務釋出的函式呼叫
    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        // 只能使用 thrift codec
        URL url = invoker.getUrl().addParameter(Constants.CODEC_KEY, ThriftCodec.NAME);
        // find server.
        String key = url.getAddress();
        //client 也可以暴露一個只有server可以呼叫的服務。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer && !serverMap.containsKey(key)) {
            serverMap.put(key, getServer(url));
        }
        // export service.
        key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        // 快取起來
        exporterMap.put(key, exporter);

        return exporter;
    }

    public void destroy() {
        // 銷燬invoker
        super.destroy();
        // 移除消費ip，關閉server
        for (String key : new ArrayList<String>(serverMap.keySet())) {
            ExchangeServer server = serverMap.remove(key);
            if (server != null) {
                server.close(getServerShutdownTimeout());
            } // ~ end of if ( server != null )
        } // ~ end of loop serverMap
    } // ~ end of method destroy

    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        ThriftInvoker<T> invoker = new ThriftInvoker<T>(type, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }
}

相關推薦

Dubbo Provider啟動流程原始碼分析

簡單的官方demo： provider的java程式碼： public static void main(String[] args) throws Exception { ClassPathXmlApplicationContext c

dubbo原始碼學習（二）dubbo容器啟動流程簡略分析

dubbo版本2.6.3 繼續之前的dubbo原始碼閱讀，從com.alibaba.dubbo.container.Main.main(String[] args)作為入口 簡單的資料一下啟動的流程 1、com.alibaba.dubbo.container.Main.main(String[] arg

Flink on Yarn模式啟動流程原始碼分析

此文已由作者嶽猛授權網易雲社群釋出。 歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。 Flink on yarn的啟動流程可以參見前面的文章 Flink on Yarn啟動流程,下面主要是從原始碼角度看下這個實現，可能有的地方理解有誤，請給予指正，多謝。 --> 1.命令列啟動yarn sessi

Activity啟動流程原始碼分析(應用中)

在移動應用開發中，Android四大元件之一Activity是最常用的。很多介面，如：閃屏、主介面、次功能介面等都需要Activity來作為主要的載體；介面與介面之間，即不同的Activity之間也都存在跳轉切換，弄懂這其中跳轉切換原理，將有助於我們更好的理解A

Java NIO框架Netty教程（四） – ServerBootStrap啟動流程原始碼分析

該圖是OneCoder通過閱讀Netty原始碼，逐漸記錄下來的。基本可以說明Netty服務的啟動流程。這裡在具體講解一下。 首先說明，我們這次順利的流程是基於NioSocketServer的。也就是基於Java NIO Selector的實現方式。在第六講《Java NIO框架Netty教程(六)-Jav

（Android 9.0）Activity啟動流程原始碼分析

前言 熟悉Activity的啟動流程和執行原理是一個合格的應用開發人員所應該具備的基本素質，其重要程度就

SpringBoot 原始碼解析 （二）----- Spring Boot精髓：啟動流程原始碼分析

本文從原始碼的角度來看看Spring Boot的啟動過程到底是怎麼樣的，為何以往紛繁複雜的配置到如今可以這麼簡便。 入口類 @SpringBootApplication public class HelloWorldMainApplication { public static void main

一、【徹底精通Spring Boot2原始碼】 SpringApplication.run() 啟動流程原始碼徹底剖析

       本文是對spring boot2 啟動整個流程進行原始碼解析，也是我發表springboot系列的第一篇文章，解析的過程我會一行一行解讀，網上很多的解析案例都是貼一段原始碼，然後說幾句概況性的語句，看了之後也是丈二和尚 摸不到頭腦，既然解析原

Netty NioEventLoop 啟動過程原始碼分析

原文連結：https://wangwei.one/posts/netty-nioeventloop-analyse-for-startup.html 前面 ，我們分析了NioEventLoop的建立過程，接下來我們開始分析NioEventLoop的啟動和執行邏輯。

Netty（五）服務端啟動過程原始碼分析——好文摘抄

下面先來一段 Netty 服務端的程式碼： public class NettyServer { public void bind(int port){ // 建立EventLoopGroup EventLoopGroup bossGroup = new

大資料之Spark（三）--- Spark核心API，Spark術語，Spark三級排程流程原始碼分析

一、Spark核心API ----------------------------------------------- [SparkContext] 連線到spark叢集,入口點. [HadoopRDD] extends RDD 讀取hadoop

Django rest framework 的認證流程(原始碼分析一)

一、基本流程舉例: urlpatterns = [ url(r'^admin/', admin.site.urls), url(r'^users/', views.HostView.as_view()), ] urls

Android中system server程序啟動流程原始碼解析 系統服務啟動

system server 前言 System Server fork SystemServer SystemServer.main() SystemServer.createSystemContext SystemSe

Uboot啟動過程原始碼分析之第二階段

UBoot的最終目標是啟動核心 1.從Flash中讀出核心 2.啟動核心 通過呼叫lib_arm/board.c中的start_armboot函式進入uboot第二階段 第二階段總結圖 typedef struct global_data { bd_t *bd; unsigned

Uboot啟動過程原始碼分析之第一階段（硬體相關）

從上一個部落格知道uboot的入口點在 cpu/arm920t/start.s 開啟cpu/arm920t/start.s 跳轉到reset reset: /* * set the cpu to SVC32 mode// CUP設定為管理模式 */ mrs r0,cps

Spring boot啟動流程原始碼解析

閱讀須知 版本：2.0.4 文章中使用/* */註釋的方法會做深入分析 正文 @SpringBootApplication public class BootApplication { public static void main(String[

spring-security-oauth2(三) 認證流程原始碼分析

認證流程原始碼分析 之前的程式碼中，我們自定義了登陸路徑，自定義成功和失敗處理器以及自定義的使用者登陸資訊校驗，下面我們通過簡單的原始碼分析，來把這些串聯起來 認證流程處理說明 認證結果如何在多個請求之間共享 獲取認證使用者資訊 認證處理流程說明 spring-

Spring Boot 2.x 啟動全過程原始碼分析（全）

上篇《Spring Boot 2.x 啟動全過程原始碼分析（一）入口類剖析》我們分析了 Spring Boot 入口類 SpringApplication 的原始碼，並知道了其構造原理，這篇我們繼續往下面分析其核心 run 方法。 SpringApplication 例

Spring IOC容器啟動流程原始碼解析(一)——容器概念詳解及原始碼初探

目錄 1. 前言 1.1 IOC容器到底是什麼 IOC和AOP是Spring框架的核心功能,而IOC又是AOP實現的基礎,因而可以說IOC是整個Spring框架的基石。那麼什麼是IOC？IOC即控制反轉,通俗的說就是讓Spring框架來幫助我們完成物件的依賴管理和生命週期控制等等工作。從面向物件的角度來說,

Tigase 傳送訊息的流程原始碼分析

XMPP 的<message/>節是使用基本的”push”方法來從一個地方到另一個地方得到訊息。因為訊息通常是不告知的，它們是一種”fire-and-forget”(發射後自尋目的)的機制來從一個地方到另一個地方快速獲取資訊訊息節有五種不同的型別，通過 type 屬性來進行區分：例如 chat 型