3.dubbo源碼-ExtensionLoader擴展機制

ExtensionLoader的使用

Dubbo中隨處可見這樣的代碼：

ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);

這段代碼非常重要，了解它的實現是掌握dubbo源碼的重中之重络断，否則很多地方的源碼會不知其所以然，接下來以獲取ThreadPool為例答捕，講解dubbo如何通過ExtensionLoader根據url獲取具體的ThreadPool祷肯；

Provider&ThreadPool

Dubbo服務端接收到所有ChannelState類型的請求，都在com.alibaba.dubbo.remoting.transport.dispatcher.all.AllChannelHandler中處理国拇，以Provider端處理接收到的請求為例昙篙，處理的代碼如下：

public void received(Channel channel, Object message) throws RemotingException {
    ExecutorService cexecutor = getExecutorService();
    try {
        cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
    } catch (Throwable t) {
        throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
    }
}

getExecutorService()就是獲取連接池ExecutorService的秘豹；對應的就是父類WrappedChannelHandler中的申明protected final ExecutorService executor携御；在new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)時，調用父類WrappedChannelHandler的構造方法時初始化既绕，源碼如下：

public WrappedChannelHandler(ChannelHandler handler, URL url) {
    this.handler = handler;
    this.url = url;
    executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);

    String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
    if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
        componentKey = Constants.CONSUMER_SIDE;
    }
    DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
    dataStore.put(componentKey, Integer.toString(url.getPort()), executor);
}

從這段代碼可知啄刹，獲取ExecutorService的代碼就是ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);，接下來我們看看dubbo如何通過這一行代碼獲取具體的ThreadPool;

ExtensionLoader

這個類非常重要凄贩，幾乎貫穿在dubbo中每一個模塊誓军，例如Protocol，LoadBalance等疲扎；因為dubbo采用SPI微內核實現方式昵时，所以外部可以從容擴展自定義實現，例如自定義實現一種線程池AfeiThreadPool椒丧，而dubbo決定具體采用哪個實現類壹甥，都在這個類中決定，ExtensionLoader主要有兩大功能：1壶熏、獲取ExtensionLoader實例句柠，2、獲取Adaptive實例棒假。調用方得到Adaptive實例后就能調用具體業(yè)務實現溯职，例如FixedThreadPool，下面的分析以獲取ThreadPool的實現為例帽哑；

1. 獲取ExtensionLoader實例

首先根據ThreadPool得到ExtensionLoader谜酒，實現源碼如下：

public static <T> ExtensionLoader<T> getExtensionLoader(Class<T> type) {
    if (type == null)
        throw new IllegalArgumentException("Extension type == null");
    // 獲取ExtensionLoader時參數即ThreadPool必須是接口類型
    if(!type.isInterface()) {
        throw new IllegalArgumentException("Extension type(" + type + ") is not interface!");
    }
    // 獲取ExtensionLoader時參數即ThreadPool必須有SPI注解，否則不可以擴展
    if(!withExtensionAnnotation(type)) {
        throw new IllegalArgumentException("Extension type(" + type + 
                ") is not extension, because WITHOUT @" + SPI.class.getSimpleName() + " Annotation!");
    }
    
    // 將接口類型與其對應的ExtensionLoader本地緩存到`ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS`中
    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;
}

2. 獲取Adaptive實例

得到ExtensionLoader后祝拯，再得到具體的Adaptive實例甚带；其中cachedAdaptiveInstance定義為：Holder<Object> cachedAdaptiveInstance她肯，如果本地沒有緩存實例，那么調用createAdaptiveExtension()獲取鹰贵，源碼如下：

public T getAdaptiveExtension() {
    // 先從本地緩存中查詢能否得到Adaptive實例
    Object instance = cachedAdaptiveInstance.get();
    if (instance == null) {
        // double-check方式保證線程安全（單實例的一種實現方式）
        if(createAdaptiveInstanceError == null) {
            synchronized (cachedAdaptiveInstance) {
                instance = cachedAdaptiveInstance.get();
                if (instance == null) {
                    try {
                        // 獲取實例
                        instance = createAdaptiveExtension();
                        // 將實例本地緩存
                        cachedAdaptiveInstance.set(instance);
                    } catch (Throwable t) {
                        createAdaptiveInstanceError = t;
                        throw new IllegalStateException("fail to create adaptive instance: " + t.toString(), t);
                    }
                }
            }
        }
        else {
            throw new IllegalStateException("fail to create adaptive instance: " + createAdaptiveInstanceError.toString(), createAdaptiveInstanceError);
        }
    }

    return (T) instance;
}

Hold申明如下晴氨，用volatile來保證數據的一致性：

/**
 * Helper Class for hold a value.
 *
 * @author william.liangf
 */
public class Holder<T> {
    
    private volatile T value;
    
    public void set(T value) {
        this.value = value;
    }
    
    public T get() {
        return value;
    }

}

如果本地還沒有緩存實例，那么調用createAdaptiveExtension()獲取Adaptive實例碉输，源碼如下:

private T createAdaptiveExtension() {
    try {
        return injectExtension((T) getAdaptiveExtensionClass().newInstance());
    } catch (Exception e) {
        throw new IllegalStateException("Can not create adaptive extenstion " + type + ", cause: " + e.getMessage(), e);
    }
}

getAdaptiveExtensionClass()源碼如下籽前，先調用getExtensionClasses();，如果不能得到Adaptive類敷钾，那么調用createAdaptiveExtensionClass()獲取Adaptive類枝哄；

private Class<?> getAdaptiveExtensionClass() {
    getExtensionClasses();
    if (cachedAdaptiveClass != null) {
        return cachedAdaptiveClass;
    }
    return cachedAdaptiveClass = createAdaptiveExtensionClass();
}

接下來我們看getExtensionClasses()方法中如何得到Adaptive擴展類，其主要分為如下幾個步驟：

  1. 得到ThreadPool接口上申明的SPI注解的值（"fixed"）阻荒，就是默認名稱cachedDefaultName 挠锥；
  2. 遍歷類路徑下三個目錄（"META-INF/dubbo/internal"，"META-INF/dubbo/"侨赡，"META-INF/services/"）蓖租，查找所有命名為type.getName()即com.alibaba.dubbo.common.threadpool.ThreadPool的文件，得到文件中所有申明的實現集合羊壹；
  3. 查找有Adaptive注解的類蓖宦，就是要找的Adaptive類；
  4. 如果沒有Adaptive類油猫，那么嘗試能否通過Wrapper方式獲取類稠茂，如果可以獲取，那么緩存到Set<Class<?>> cachedWrapperClasses;中情妖；
  5. 把com.alibaba.dubbo.common.threadpool.ThreadPool文件中得到的所有類緩存到Holder<Map<String, Class<?>>> cachedClasses中睬关；
private Map<String, Class<?>> loadExtensionClasses() {
    // 如果入參類型申明了SPI注解，那么SPI注解中的value就是默認實現名稱毡证，即cachedDefaultName （SPI注解中value的值只能有1個共螺，超過1個會拋出異常）；
    final SPI defaultAnnotation = type.getAnnotation(SPI.class);
    if(defaultAnnotation != null) {
        String value = defaultAnnotation.value();
        if(value != null && (value = value.trim()).length() > 0) {
            String[] names = NAME_SEPARATOR.split(value);
            if(names.length > 1) {
                throw new IllegalStateException("more than 1 default extension name on extension " + type.getName()
                        + ": " + Arrays.toString(names));
            }
            if(names.length == 1) cachedDefaultName = names[0];
        }
    }
    
    // 遍歷類路徑下的（DUBBO_INTERNAL_DIRECTORY情竹，DUBBO_DIRECTORY藐不，SERVICES_DIRECTORY）三個目錄，將符合type類型的所有實現存到Map中秦效；
    Map<String, Class<?>> extensionClasses = new HashMap<String, Class<?>>();
    loadFile(extensionClasses, DUBBO_INTERNAL_DIRECTORY);
    loadFile(extensionClasses, DUBBO_DIRECTORY);
    loadFile(extensionClasses, SERVICES_DIRECTORY);
    return extensionClasses;
}

2.2 創(chuàng)建Adaptive擴展類

由于getExtensionClasses()中不能得到ThreadPool類型的Adaptive類（帶有Adaptive注解）雏蛮，所以需要通過createAdaptiveExtensionClass()創(chuàng)建Adaptive擴展類，由源碼可知阱州，其創(chuàng)建Adaptive擴展類分為如下幾個步驟：

  1. 通過createAdaptiveExtensionClassCode()得到用于生成Adaptive擴展類的源代碼code
  2. 得到類加載器ClassLoader
  3. 通過擴展機制得到編譯器（默認用JavassistCompiler）將擴展類源代碼code編譯成字節(jié)碼從而得到Adaptive類
private Class<?> createAdaptiveExtensionClass() {
    String code = createAdaptiveExtensionClassCode();
    ClassLoader classLoader = findClassLoader();
    com.alibaba.dubbo.common.compiler.Compiler compiler = ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.compiler.Compiler.class).getAdaptiveExtension();
    return compiler.compile(code, classLoader);
}

通過DEBUG方式得到ThreadPool的Adaptive擴展類的源代碼如下挑秉，由下面的源碼可知獲取ThreadPool類型的Adaptive類的步驟：

  1. 判斷請求URL中是否申明threadpool，如果沒有申明threadpool苔货，那么默認為fixed犀概，即擴展名稱立哑；
  2. 根據擴展名稱調用ExtensionLoader.getExtension(extName)得到具體的實現類
    從用于生成Adaptive擴展類的源碼可以得知，URL中對threadpool申明的優(yōu)先級要高于SPI注解申明姻灶；所以如果我們想修改默認的ThreadPool實現方式铛绰，可以通過URL中指定threadpool；
package com.alibaba.dubbo.common.threadpool;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class ThreadPool$Adpative implements com.alibaba.dubbo.common.threadpool.ThreadPool {
    public java.util.concurrent.Executor getExecutor(com.alibaba.dubbo.common.URL arg0) {
        if (arg0 == null) throw new IllegalArgumentException("url == null");
        com.alibaba.dubbo.common.URL url = arg0;
        String extName = url.getParameter("threadpool", "fixed");
        if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.common.threadpool.ThreadPool) name from url(" + url.toString() + ") use keys([threadpool])");
        com.alibaba.dubbo.common.threadpool.ThreadPool extension = (com.alibaba.dubbo.common.threadpool.ThreadPool)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.common.threadpool.ThreadPool.class).getExtension(extName);
        return extension.getExecutor(arg0);
    }
}

創(chuàng)建Adaptive擴展類后产喉，看.getExtension(String)方法如何得到具體的實現類捂掰，實現源碼如下，即首先嘗試從本地緩存cachedInstances中獲取曾沈，如果獲取不到这嚣，那么調用createExtension(name)創(chuàng)建擴展類實例并本地緩存；這里dubbo的作者又用到了double-check方式保證線程安全實現單實例塞俱；

public T getExtension(String name) {
    if (name == null || name.length() == 0)
        throw new IllegalArgumentException("Extension name == null");
    if ("true".equals(name)) {
        return getDefaultExtension();
    }
    Holder<Object> holder = cachedInstances.get(name);
    if (holder == null) {
        cachedInstances.putIfAbsent(name, new Holder<Object>());
        holder = cachedInstances.get(name);
    }
    Object instance = holder.get();
    if (instance == null) {
        synchronized (holder) {
            instance = holder.get();
            if (instance == null) {
                instance = createExtension(name);
                holder.set(instance);
            }
        }
    }
    return (T) instance;
}

createExtension(String name)部分實現源碼如下姐帚，由于前面已經調用了getExtensionClasses()得到所有ThreadPool類型并本地緩存在Holder<Map<String, Class<?>>> cachedClasses中，cachedClasses.get()得到保存的數據如下：

fixed=com.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool
cached=com.alibaba.dubbo.common.threadpool.support.cached.CachedThreadPool
limited=com.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool

根據name的值就能夠取得ThreadPool的具體實現類障涯，例如默認fixed對應的具體實現類就是FixedThreadPool卧土；如果設置threadpool="limited"，那么對應的具體實現類就是LimitedThreadPool像樊；如果擴展了自定義AfeiThreadPool并在META-INF/dubbo/com.alibaba.dubbo.common.threadpool.ThreadPool文件中申明afei=com.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool，那么設置threadpool="afei"旅敷，就能得到自定義的AfeiThreadPool：

private T createExtension(String name) {
    Class<?> clazz = getExtensionClasses().get(name);
    if (clazz == null) {
        throw findException(name);
    }
    ... ...
}

配置ThreadPool

得到具體的ThreadPool實現后生棍，dubbo根據url調整ThreadPoolExecutor，以FixedThreadPool為例（dubbo provider端默認采用的連接池處理方式）媳谁，源碼如下：

public class FixedThreadPool implements ThreadPool {

    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
        return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, 
                queues == 0 ? new SynchronousQueue<Runnable>() : 
                    (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                            : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}
  1. 如果url中配置了線程名稱涂滴，那么進行設置，否則基于Constants.DEFAULT_THREAD_NAME得到默認線程名稱;
  2. 如果url中配置了線程數量晴音，那么進行設置柔纵，否則用默認線程數量：Constants.DEFAULT_THREADS，即默認200個線程锤躁；
  3. 如果url中配置了線程隊列搁料，那么進行設置，否則用默認線程隊列：Constants.DEFAULT_QUEUES系羞，即默認為0郭计；
    例如我們可以在provider.xml中對線程池做如下自定義：
    <dubbo:protocol threadpool="fixed" threads="288" queues="10"/>
最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯系作者
  • 人面猴
    序言：七十年代末，一起剝皮案震驚了整個濱河市椒振，隨后出現的幾起案子昭伸，更是在濱河造成了極大的恐慌，老刑警劉巖澎迎，帶你破解...
    沈念sama閱讀 210,914評論 6贊 490
  • 死咒
    序言：濱河連續(xù)發(fā)生了三起死亡事件庐杨，死亡現場離奇詭異选调，居然都是意外死亡，警方通過查閱死者的電腦和手機灵份，發(fā)現死者居然都...
    沈念sama閱讀 89,935評論 2贊 383
  • 救了他兩次的神仙讓他今天三更去死
    文/潘曉璐 我一進店門仁堪，熙熙樓的掌柜王于貴愁眉苦臉地迎上來，“玉大人各吨，你說我怎么就攤上這事枝笨。” “怎么了揭蜒？”我有些...
    開封第一講書人閱讀 156,531評論 0贊 345
  • 道士緝兇錄：失蹤的賣姜人
    文/不壞的土叔 我叫張陵横浑，是天一觀的道長。 經常有香客問我屉更，道長徙融，這世上最難降的妖魔是什么？ 我笑而不...
    開封第一講書人閱讀 56,309評論 1贊 282
  • ?港島之戀（遺憾婚禮）
    正文 為了忘掉前任瑰谜，我火速辦了婚禮欺冀，結果婚禮上，老公的妹妹穿的比我還像新娘萨脑。我一直安慰自己隐轩，他們只是感情好，可當我...
    茶點故事閱讀 65,381評論 5贊 384
  • 惡毒庶女頂嫁案：這布局不是一般人想出來的
    文/花漫 我一把揭開白布渤早。 她就那樣靜靜地躺著职车，像睡著了一般。 火紅的嫁衣襯著肌膚如雪鹊杖。 梳的紋絲不亂的頭發(fā)上悴灵，一...
    開封第一講書人閱讀 49,730評論 1贊 289
  • 城市分裂傳說
    那天，我揣著相機與錄音骂蓖，去河邊找鬼积瞒。 笑死，一個胖子當著我的面吹牛登下，可吹牛的內容都是我干的茫孔。 我是一名探鬼主播，決...
    沈念sama閱讀 38,882評論 3贊 404
  • 雙鴛鴦連環(huán)套：你想象不到人心有多黑
    文/蒼蘭香墨 我猛地睜開眼被芳，長吁一口氣：“原來是場噩夢啊……” “哼银酬！你這毒婦竟也來了？” 一聲冷哼從身側響起筐钟，我...
    開封第一講書人閱讀 37,643評論 0贊 266
  • 萬榮殺人案實錄
    序言：老撾萬榮一對情侶失蹤揩瞪，失蹤者是張志新（化名）和其女友劉穎，沒想到半個月后篓冲，有當地人在樹林里發(fā)現了一具尸體李破，經...
    沈念sama閱讀 44,095評論 1贊 303
  • ?護林員之死
    正文 獨居荒郊野嶺守林人離奇死亡宠哄，尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內容為張勛視角 年9月15日...
    茶點故事閱讀 36,448評論 2贊 325
  • ?白月光啟示錄
    正文 我和宋清朗相戀三年，在試婚紗的時候發(fā)現自己被綠了嗤攻。 大學時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片毛嫉。...
    茶點故事閱讀 38,566評論 1贊 339
  • 活死人
    序言：一個原本活蹦亂跳的男人離奇死亡，死狀恐怖妇菱，靈堂內的尸體忽然破棺而出承粤，到底是詐尸還是另有隱情，我是刑警寧澤闯团，帶...
    沈念sama閱讀 34,253評論 4贊 328
  • ?日本核電站爆炸內幕
    正文 年R本政府宣布辛臊，位于F島的核電站，受9級特大地震影響房交，放射性物質發(fā)生泄漏彻舰。R本人自食惡果不足惜，卻給世界環(huán)境...
    茶點故事閱讀 39,829評論 3贊 312
  • 男人毒藥：我在死后第九天來索命
    文/蒙蒙 一候味、第九天 我趴在偏房一處隱蔽的房頂上張望刃唤。 院中可真熱鬧，春花似錦白群、人聲如沸尚胞。這莊子的主人今日做“春日...
    開封第一講書人閱讀 30,715評論 0贊 21
  • 一樁弒父案帜慢，背后竟有這般陰謀
    文/蒼蘭香墨 我抬頭看了看天上的太陽笼裳。三九已至，卻和暖如春崖堤，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背耐床。 一陣腳步聲響...
    開封第一講書人閱讀 31,945評論 1贊 264
  • 情欲美人皮
    我被黑心中介騙來泰國打工密幔， 沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留，地道東北人撩轰。 一個月前我還...
    沈念sama閱讀 46,248評論 2贊 360
  • 代替公主和親
    正文 我出身青樓胯甩，卻偏偏與公主長得像，于是被迫代替她去往敵國和親堪嫂。 傳聞我的和親對象是個殘疾皇子偎箫，可洞房花燭夜當晚...
    茶點故事閱讀 43,440評論 2贊 348

推薦閱讀更多精彩內容