在AbstractConfigurationProvider類中loadSources方法會將所有的source進行封裝成SourceRunner放到了Map<String, SourceRunner> sourceRunnerMap之中。相關代碼如下:
1 Map<String, String> selectorConfig = context.getSubProperties( 2 BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX); 3 4 ChannelSelector selector = ChannelSelectorFactory.create( 5 sourceChannels, selectorConfig); 6 7 ChannelProcessor channelProcessor = new ChannelProcessor(selector); 8 Configurables.configure(channelProcessor, context); 9 source.setChannelProcessor(channelProcessor); 10 sourceRunnerMap.put(sourceName, 11 SourceRunner.forSource(source));
每個source都有selector。上述代碼會獲取配置文件中關於source的selector配置信息;然后構造ChannelSelector對象selector;並封裝selector對象成ChannelProcessor對象channelProcessor;執行channelProcessor.configure方法進行配置;設置soure的channelprocessor,最后封裝為sourceRunner和source名稱一起放入sourceRunnerMap中。
一、ChannelSelector selector = ChannelSelectorFactory.create(sourceChannels, selectorConfig)會根據配置文件中指定的類型實例化一個ChannelSelector(共兩種ReplicatingChannelSelector復制和MultiplexingChannelSelector復用)如果沒有指定類型默認是ReplicatingChannelSelector,也就是配置文件中不用配置selector會將每個event復制發送到多個channel;selector.setChannels(channels);對此slector進行配置configure(context)。這兩中selector都實現了三個方法getRequiredChannels(Event event)、getOptionalChannels(Event event) 以及configure(Context context)。其實Event要發送到的channel有兩種組成:RequiredChannels和OptionalChannels,對應兩個方法。
(1)ReplicatingChannelSelector的configure(context)方法會獲得通過"optional"在配置文件中指定的可選發送的channels(可以多個,通過空格分割);獲取requiredChannels是此source對應的channel中可以活動的channel列表;然后獲取所有channel的名字及其與channel的映射channelNameMap;然后將可選的channel加入optionalChannels並從requiredChannels去掉有對應的channel,在這里並沒有檢查可選channel的合法性以及可以配置此source指定的channel之外的channel,requiredChannels和optionalChannels不能有交集,有交集的話會從requiredChannels中刪除相交的channel,所以如果配置文件中optional指定的channel列表和source指定的列表相同getOptionalChannels方法有可能會返回全部可活動channel列表使得數據重復,所以建議optional指定的channel最好是source指定之外的其他channel(比如是其他source的channel)。getOptionalChannels方法就是直接返回optionalChannels列表,getRequiredChannels方法返回requiredChannels列表,如果requiredChannels為null,則返回全部的可以活動的channel列表。
(2)MultiplexingChannelSelector的configure(context)先獲取要匹配的event的header的headerName,只能選擇一個headerName;獲得默認發送到的channel列表defaultChannels,可以指定多個默認channel;獲得mapping的各個子值,及對應的channel名稱mapConfig;用來存儲header不同的值及其對應的要發送到的channel列表(每個map可以發送到多個channel中,每個channel也可以同時對應多個mapping),存入channelMapping(這個數據結構是用來存儲mapping值及對應的channel列表的);optionalChannels是配置的可選值及其要發送到的channel列表的映射關系,channelMapping中已經出現的channel不允許再次在optionalChannels出現(防止數據重復),如果channelMapping沒有這個值對應的channel列表(表示可能會使用默認的channel列表)則使過濾與默認channel列表的交集,optionalChannels存儲的是對應header的各個值及其等於該值的event要發送到的可選擇的channel列表。getOptionalChannels(Event event)方法返回的是optionalChannels中該event的指定header對應的可選擇的channel列表。getRequiredChannels(Event event)方法返回的是channelMapping中該event的指定header對應的channel列表,如果為null(表示由於該event的headers沒有匹配的channel就發送到默認的channel中)就返回默認發送列表defaultChannels。需要說明的是選擇器配置文件中的"default"、"mapping."、"optional."這三個是同等級的,沒有匹配后兩者的值時才會選擇發送到default對應的channel列表,后兩者的值都是event的header中對應配置文件中指定的"header"的各種值。當調用getRequiredChannels(Event event)和getOptionalChannels(Event event)方法時都會對這個event的相應header查找對應要發送到的channel列表。
二、 ChannelProcessor channelProcessor = new ChannelProcessor(selector)這個是封裝選擇器構造channelprocessor。其構造方法會賦值selector並構造一個InterceptorChain對象interceptorChain。ChannelProcessor類負責管理選擇器selector和攔截器interceptor。
三、執行channelProcessor.configure(Context)進行必要的配置,該方法會調用channelProcessor.configureInterceptors(context)對攔截器們進行獲取和配置,configureInterceptors方法會先從配置文件中獲取interceptor的組件名字interceptorNames[](可以多個),然后獲取所有的“interceptors.”的配置信息interceptorContexts,然后遍歷所有interceptorNames從配置文件中獲取屬於這個interceptor的配置信息及類型(type),根據類型構建相應的interceptor並進行配置configure,加入interceptors列表(用來存放實例化的interceptor);最后將列表傳遞給interceptorChain。關於更多interceptor的信息可以看這篇Flume-NG源碼閱讀之Interceptor(原創) 。
四、source.setChannelProcessor(channelProcessor)賦值。各個source通過getChannelProcessor()方法獲取processor調用其processEventBatch(events)或者processEvent(event)來將event送到channel中。
五、sourceRunnerMap.put(sourceName,SourceRunner.forSource(source))將source封裝成SourceRunner放入sourceRunnerMap。SourceRunner.forSource會根據這個source所實現的接口封裝成不同的Runner,有兩種接口PollableSource和EventDrivenSource,前者是有自己線程來驅動的需要實現process方法,后者是沒有單獨的線程來驅動的沒有process方法。
1 public static SourceRunner forSource(Source source) { 2 SourceRunner runner = null; 3 4 if (source instanceof PollableSource) { 5 runner = new PollableSourceRunner(); 6 ((PollableSourceRunner) runner).setSource((PollableSource) source); 7 } else if (source instanceof EventDrivenSource) { 8 runner = new EventDrivenSourceRunner(); 9 ((EventDrivenSourceRunner) runner).setSource((EventDrivenSource) source); 10 } else { 11 throw new IllegalArgumentException("No known runner type for source " 12 + source); 13 } 14 15 return runner; 16 }
(1)PollableSourceRunner的start()方法會獲取source的ChannelProcessor,然后執行其initialize()方法,該方法會調用interceptorChain.initialize()方法對攔截器們進行初始化(遍歷所有攔截器然后執行攔截器的initialize()方法);然后執行source.start()啟動source;再啟動一個線程PollingRunner,它的run方法會始終執行source.process()並根據返回的狀態值做一些統計工作。
(2)EventDrivenSourceRunner的start()方法會獲取source的ChannelProcessor,然后執行其initialize()方法,該方法會調用interceptorChain.initialize()方法對攔截器們進行初始化(遍歷所有攔截器然后執行攔截器的initialize()方法);然后執行source.start()啟動source。
這樣就完成了sourceRunnerMap的組裝。當在Application中的startAllComponents方法中通過materializedConfiguration.getSourceRunners()獲取所有的SourceRunner並放入supervisor.supervise中去執行,會調用到SourceRunner.start()方法,即上面剛講到的內容。這樣source就啟動了。然后當將封裝的Events或者Event發送到channel時,需要使用對應的方法ChannelProcessor.processEventBatch(List<Event> events)或者ChannelProcessor.processEvent(Event event)就可以將數據從source傳輸到channel中,這兩個方法都會在開始調用interceptorChain.intercept(events)或者interceptorChain.intercept(event)對event增加headers(如果有多個interceptor會遍歷interceptors處理每個event)。ChannelProcessor都是通過在source中直接調用getChannelProcessor()(在所有的source的父類AbstractSource中實現的)獲得。看一看processEventBatch(List<Event> events)代碼:
1 public void processEventBatch(List<Event> events) { 2 Preconditions.checkNotNull(events, "Event list must not be null"); 3 4 events = interceptorChain.intercept(events); 5 6 Map<Channel, List<Event>> reqChannelQueue = //需要發送到的每個channel及其要發送到這個channel的event列表 7 new LinkedHashMap<Channel, List<Event>>(); 8 9 Map<Channel, List<Event>> optChannelQueue = //可選的每個channel及其要發送到這個channel的event列表 10 new LinkedHashMap<Channel, List<Event>>(); 11 12 for (Event event : events) { 13 List<Channel> reqChannels = selector.getRequiredChannels(event); //獲取需要發送到的所有channel 14 15 for (Channel ch : reqChannels) { 16 List<Event> eventQueue = reqChannelQueue.get(ch); 17 if (eventQueue == null) { 18 eventQueue = new ArrayList<Event>(); 19 reqChannelQueue.put(ch, eventQueue); 20 } 21 eventQueue.add(event); //將event放入對應channel的event列表 22 } 23 24 List<Channel> optChannels = selector.getOptionalChannels(event); //獲取可選的要發送到的所有channel 25 26 for (Channel ch: optChannels) { 27 List<Event> eventQueue = optChannelQueue.get(ch); 28 if (eventQueue == null) { 29 eventQueue = new ArrayList<Event>(); 30 optChannelQueue.put(ch, eventQueue); 31 } 32 33 eventQueue.add(event); //將event放入對應channel的event列表 34 } 35 } 36 37 // Process required channels 38 for (Channel reqChannel : reqChannelQueue.keySet()) { 39 Transaction tx = reqChannel.getTransaction(); //創建事務 40 Preconditions.checkNotNull(tx, "Transaction object must not be null"); 41 try { 42 tx.begin(); 43 44 List<Event> batch = reqChannelQueue.get(reqChannel); 45 46 for (Event event : batch) { //發送到需要發送到的channel 47 reqChannel.put(event); 48 } 49 50 tx.commit(); 51 } catch (Throwable t) { 52 tx.rollback(); //事務回滾 53 if (t instanceof Error) { 54 LOG.error("Error while writing to required channel: " + 55 reqChannel, t); 56 throw (Error) t; 57 } else { 58 throw new ChannelException("Unable to put batch on required " + 59 "channel: " + reqChannel, t); 60 } 61 } finally { 62 if (tx != null) { 63 tx.close(); 64 } 65 } 66 }
上述代碼不復雜,會獲得所有需要發送到的channel和所有可選的channel,然后針對每個channel,將所有event放入一個列表與該channel組成映射;然后會遍歷兩種channel列表中的每個channel將它對應的所有event發送到對應的channel中。這個方法寫的不夠友好,還可以再優化,因為方法的參數本身就是一個列表可以省去一層for循環,直接將reqChannelQueue.put(ch, eventQueue)和optChannelQueue.put(ch, eventQueue)中的eventQueue改為傳遞過來的參數List<Event> events就可以達到優化的目的。
processEvent(Event event)方法就更簡單了,將這個event發送到這兩種channel列表中每個channel就可以。
在發送到channel的過程中我們也發現都會有事務的創建(getTransaction())、開始(tx.begin())、提交(tx.commit())、回滾(tx.rollback())、關閉(tx.close())等操作,這是必須的。在sink中這些操作需要顯示的去調用,而在source端則封裝在processEvent和processEventBatch方法中,不需要顯示的調用了,但不是不調用。
至此,sourceRunner的配置、初始化、執行就講解完畢了。在配置文件中看到的interceptor和selector都是在這里進行配置及執行的。通過了解上述,我們自定義source組件是不是更容易了。呵呵
后續還有精彩內容!敬請期待哈!