ApacheNifi概念介紹、源碼解析、開發(fā)指南（中文）

NiFi文檔

1.初識NiFi

1.1.概述

NiFi最早是美國國家安全局內(nèi)部使用的工具,用來投遞海量的傳感器數(shù)據(jù),后來由apache基金會開

源。NiFi基本設(shè)計理念：FlowBasedProgramming,

1.2.核心概念

?FlowFile

FlowFile表示在系統(tǒng)中移動的每個對象，F(xiàn)lowFile由兩部分組成：

ocontent內(nèi)容，即數(shù)據(jù)本身

oattributes屬性，每條數(shù)據(jù)帶上的屬性信息.以鍵值對的形式.

?FlowFileProcessor

oFlowFile處理器，由它完成對數(shù)據(jù)的實(shí)際處理工作,包括但不限于對數(shù)據(jù)內(nèi)容和屬性的加載，路

由，轉(zhuǎn)換，輸出等.

o處理器最靈活之處在于處理器可以讀寫FlowFile的屬性信息,并且用自帶的領(lǐng)域特定語言(DSL)

對屬性進(jìn)行編程.

?Connection

。由Connections把各個處理器鏈接起來，從而形成數(shù)據(jù)處理流程的有向無環(huán)圖(DAG圖).也稱

數(shù)據(jù)流，NiFi中的Flow.

。Connection同時充當(dāng)處理器間的隊(duì)列，并且隊(duì)列的屬性高度可配置.

o這些隊(duì)列可以配置優(yōu)先級，可以在設(shè)置閾值，可以實(shí)現(xiàn)反壓。

?FlowController

o流控制器對用戶不可見的,它充當(dāng)維護(hù)處理器如何連接和管理所有處理器所使用的線程及其分

配的重要角色。

FlowController充當(dāng)促進(jìn)處理器之間FlowFiles交換的代理。

ProcessGroup

o為了方便管理，把一組特定的處理器及其連接組成的Flow放到一個處理組中去，可以通過輸

入端口接收數(shù)據(jù)并通過輸出端口發(fā)送數(shù)據(jù)。

o以這種方式，處理組可以通過組合其他組來創(chuàng)建全新組，形成更加復(fù)雜的DAG圖（Flow流）。

1.3.關(guān)鍵特性

?Flow流高度可管理

?保證交付

NiFi的一個核心理念是即使在非常高的規(guī)模下，保證交付也是必須的。這是通過有效使用專門的

持久化的預(yù)寫日志（WAL）和內(nèi)容存儲庫來實(shí)現(xiàn)的。它們的設(shè)計可以實(shí)現(xiàn)非常高的事務(wù)處理速率，

有效的負(fù)載分散，寫入時復(fù)制以及發(fā)揮傳統(tǒng)磁盤讀/寫的優(yōu)勢。

?背壓和數(shù)據(jù)緩沖機(jī)制

NiFi支持緩沖所有隊(duì)列數(shù)據(jù)，以及在這些隊(duì)列達(dá)到指定限制時提供背壓的能力，或者在數(shù)據(jù)達(dá)到

指定時間時使數(shù)據(jù)過期失效。

?可配置優(yōu)先級的隊(duì)列

NiFi允許設(shè)置一個或多個優(yōu)先級策略，用于如何從隊(duì)列中檢索數(shù)據(jù)。默認(rèn)是先進(jìn)先出，但有時候

應(yīng)該先拉取最新的數(shù)據(jù)，最大的數(shù)據(jù)或其他一些自定義方案。

?Flow流可配置特定的QoS（延遲v吞吐量，容量損失等）

在Flow流中有一些點(diǎn)是很關(guān)鍵的，且不能容忍丟失.或者有時候必須在幾秒鐘內(nèi)處理和交付它。

NiFi可以對這些問題進(jìn)行細(xì)粒度的特定配置。

?易于使用

?可視化的控制和命令

得益于強(qiáng)大的web操作界面.無論多么復(fù)雜的數(shù)據(jù)流都能在web界面上直觀的呈現(xiàn),整個數(shù)據(jù)

處理流程，包括設(shè)計，控制，反饋和監(jiān)控都可在web界面完成，一步到位.任何更改都能在界面上立馬

生效,完全不要部署的過程,對于整個數(shù)據(jù)流，更可以對中間某個處理器進(jìn)行單獨(dú)變更，實(shí)時生效.

?數(shù)據(jù)流模板

對于設(shè)計好的數(shù)據(jù)流處理流，可以保存為模板來進(jìn)行復(fù)用.模板可以導(dǎo)出成xml文件，導(dǎo)入到其他

NiFi中進(jìn)行多處使用.

?數(shù)據(jù)溯源

flowfile流過Flow流時，NiFi會自動記錄，索引并提供可用的起源數(shù)據(jù)，包括導(dǎo)入，導(dǎo)出，轉(zhuǎn)

換等。這些信息對于故障排除，優(yōu)化等很有用處.

?對歷史數(shù)據(jù)進(jìn)行細(xì)粒度的恢復(fù)

NiFi的內(nèi)容存儲庫旨在充當(dāng)歷史記錄的滾動緩沖區(qū)。數(shù)據(jù)僅在內(nèi)容存儲庫過期時或存儲空間不足

時才會被刪除。這與數(shù)據(jù)起源能力相結(jié)合，提供了非常精細(xì)的操作功能.包括對數(shù)據(jù)歷史中的某一

個點(diǎn)的點(diǎn)擊查看內(nèi)容，下載內(nèi)容，處理回放等功能.所有數(shù)據(jù)都可以回溯到它生命周期中很早的某一

點(diǎn).

安全機(jī)制

?系統(tǒng)內(nèi)部安全

Flow流中的流動的數(shù)據(jù)都可以進(jìn)行加密傳輸

?用戶使用安全

支持用戶認(rèn)證和不同級別的用戶授權(quán)（可讀，管理數(shù)據(jù)流，系統(tǒng)管理）

?多租戶授權(quán)

?可擴(kuò)展的架構(gòu)設(shè)計

?可擴(kuò)展組件

NiFi的核心設(shè)計就是擴(kuò)展.它的processors,ControllerServices,ReportingTasks,Pri

oritizers,andCustomerUserInterfaces都是可擴(kuò)展的.

?隔離的類加載器

自定義的類加載器保證了擴(kuò)展的組件簡單的依賴關(guān)系.

?點(diǎn)到點(diǎn)的通信協(xié)議

NiFi實(shí)例之間的通信協(xié)議是NiFi點(diǎn)到點(diǎn)（S2S）協(xié)議。S2s可以輕松，高效，安全地將數(shù)據(jù)從

一個NiFi實(shí)例傳輸?shù)搅硪粋€實(shí)例。NiFi客戶端的庫也可以輕松在其他應(yīng)用程序使用，以通過S

2s來與NiFi實(shí)例進(jìn)行通信。S2s中支持基于套接字的協(xié)議和HTTP（S）協(xié)議作為底層傳輸協(xié)

議，使得可以將代理服務(wù)器嵌入到S2s通信中。

?靈活的擴(kuò)容模型

?更多的NiFi實(shí)例

可以搭建NiFi集群，也可以不組成集群，多臺機(jī)器使用點(diǎn)到點(diǎn)協(xié)議來協(xié)作.

?更大的并發(fā)數(shù)量

直接修改處理器的并發(fā)數(shù)

1.4.架構(gòu)

匚OS/Host

白JVMQWebServer

芯FlowController

Processor1ExtensionN

mFlowFile塞Content塞Provenance

RepositoryRepositoryRepository

LocalStorage

?WebServer

web服務(wù)器的提供基于http的命令和控制API。

?FlowController

流量控制器是操作的大腦。它為擴(kuò)展程序提供運(yùn)行所需的線程，并管理擴(kuò)展程序何時接收執(zhí)行資源的

時間表。

?Processor

處理組件

?Extensions

擴(kuò)展組件

?FlowFileRepository

通過FlowFileRespository可跟蹤Flow中處于活動狀態(tài)的FlowFile的狀態(tài)。存儲庫的實(shí)現(xiàn)是插件

式的，默認(rèn)是位于指定磁盤分區(qū)上的持久性預(yù)寫日志。

?ContentRepository

ContentRepository作為FlowFile的存儲庫，實(shí)現(xiàn)是插件式的，默認(rèn)是一種相當(dāng)簡單的機(jī)制，該

機(jī)制將數(shù)據(jù)塊存儲在文件系統(tǒng)中?？梢灾付ǘ鄠€文件系統(tǒng)存儲位置，以便使用不同的物理分區(qū)以減少任何

單個卷上的爭用。

?ProvenanceRepository

ProvenanceRepository是存儲所有來源事件數(shù)據(jù)的地方。存儲庫實(shí)現(xiàn)是插件式的，默認(rèn)實(shí)現(xiàn)是使

用一個或多個物理磁盤卷。在每個位置內(nèi)，事件數(shù)據(jù)都被索引并可以搜索。

集群

QOS/Hosto

aQOS/HosI

不8、QOS/Hosto

戈S、OS/Hostoo

ZooKeeperServer

?ClusterCoordinator

?PrimaryNode

?ZooKeeperClient

從NiFi1.0版本開始，采用了零主群集的范例。NiFi群集中的每個節(jié)點(diǎn)都對數(shù)據(jù)執(zhí)行相同的任務(wù)，

但是每個節(jié)點(diǎn)都對不同的數(shù)據(jù)集進(jìn)行操作。通過ZooKeeper選擇一個節(jié)點(diǎn)作為集群協(xié)調(diào)器，并且故障轉(zhuǎn)

移由ZooKeeper自動處理。所有群集節(jié)點(diǎn)均向群集協(xié)調(diào)器報告心跳和狀態(tài)信息。群集協(xié)調(diào)器負(fù)責(zé)斷開和

連接節(jié)點(diǎn)。此外，每個群集都有一個主節(jié)點(diǎn)，該節(jié)點(diǎn)也由ZooKeeper選擇。作為DataFlow管理者，您

可以通過任何節(jié)點(diǎn)的用戶界面（UI）與NiFi群集進(jìn)行交互。您所做的任何更改都將復(fù)制到群集中的所有節(jié)

點(diǎn)，從而允許多個入口點(diǎn)。

2.源代碼淺析

2.1.1.總體結(jié)構(gòu)

▼?Snifi-1.10.0[nifi]-/SRC/nifi-1.10.0<modules>

?■.github<module>nifi-commons</module>

?■.idea<module>nifi-api</module>

?nifi-api<module>nifi-framework-api</module>

Anifi-assembly<module>nifi-bootstrap</module>

?nifi-bootstrap

<module>nifi-mock</module>

?*2nifi-commons

<module>nifi-nar-bundles</module>

A■"nifi-docker

<module>nifi-assembly</module>

?nifi-docs

AB"nifi-external<module>nifi-docs</module>

?nifi-framework-api<module>nifi-maven-archetypes</module>

?nifi-maven>archetypes<module>nifi-external</module>

?nifi-mock<module>nifi-toolkit</module>

a*2nifi-nar-bundles<module>nifi-docker</module>

AB"nifi-toolkit</modules>

?nifi-api

就是nifi的應(yīng)用程序接口，里面就是定義了整個工程用到的接口，注解，抽象類和枚舉等基本的接口

和信息.

?nifi-assembl

負(fù)責(zé)nifi的成品裝配，工程打包最后形成的可供部署的壓縮包就在這個工程里的target目錄內(nèi).

?nifi-bootstarp

負(fù)責(zé)nifi這個jvm應(yīng)用程序的啟動相關(guān)事宜

?nifi-commons

nifi諸多特性，比如datd-provenance,expression-language.s2s傳輸?shù)膶?shí)現(xiàn)就在這里,同時也

是nifi的工具類集合

?nifi-docker

nifi的docker應(yīng)用相關(guān)

?nifi-docs

nifi的文檔實(shí)現(xiàn)相關(guān)

?nifi-external

nifi內(nèi)部元信息和外部交換,主要用于集群模式下

?nifi-framework-api

這就是nifi核心框架層的api,也就是架構(gòu)圖中的FlowController那一層，注意這里只是各種接口

信息定義，不是實(shí)現(xiàn).

?nifi-maven-archetypes

這里只是為了生成兩個mavenarchetype,一個是nifi自定義處理器的腳手架,一個是nifi自定

義服務(wù)的腳手架,這些腳手架在maven的中央倉庫都有提供.

?nifi-mock

用于nifi的mock測試

?nifi-nar-bundles

nifijava工具箱就是這里了.整個nifi里面大部分的maven工程都是這個工程的子工程,在這個

工程里面，一個bundle就是一個工具，也對應(yīng)著上面架構(gòu)圖里的Extension

?nifi-toolkit

這里面是nifi的命令行工具的實(shí)現(xiàn).nifi也提供了比較豐富的命令行指令.

2.1.2.Nifi程序入口

在nifi-bpootstrap模塊內(nèi)有一個org.apache.nifi.bootstrap.RunNifi的類，該類的main。方法

即為Nifi的啟動入口方法。

▼Bjnifi-bootstrapSystem.ouf.println(,

System.out.println("Restarx:StopApacheN1FJ,:fitisrunning,andthensti

▼?src

Systea.out.println("5tatus：DetermineifthereisarunningInstanceofApact

▼feimain

Systm.our.printIn("Dump:WriteaThreadDuaptothefilespecifiedby[optic

▼BijavaSystea.out.println(Diagnostics:Writediagnosticinformationtothefilesp<

▼Biorg.apache.nifi.bootstrap"thefilename,whichaoyresultinadditionaldiagnosticinfor?ationbeinc

?BiexceptionSyste?.our.println(°Run:StartanewinstanceofApacheNiFiandMonitorthe

?BinotificationSystefl.out.prlntlnO;

?ESutil>

?GBootstrapCodec

?QNIFILIstener@privatestaticString!)■>■(finalString!)orig)(returnArrays.copyOfRange(or:

1▼0RunNiFipublicstaticvoidmain(String[]args)throwslOExceptlon,InterruptedException(1

.if〔aros.lengthv111lengtha3)41

SfirtusprintUsagel];

ORunNiFi(File)return;

QcreateSensitiveKeyFile(File):Pat}

Gdiagnostics(Flle,boolean):void

Odump(File)：voidFiledunpEile-null;

Oenv():voidbooleanverbose■false：

QgetBootstrapFlle(Logger,String

finalStringand=args(0];

0getChildProcesses(String):List<

If(cnd.equal.5lgnoreCase(?nothe?Stnng："dump")){

QgetCurrentPort(Logger)：lnteger

if(argselength>1){

0getDefaultBootstrapConfFile():FduapFlle-newFHe(arg$(l]);

QgetFHe(String,File):File}else(

QgetHostname():String(MpfXte■null：

GqetLockFile(Loqqer):Fil

接著看start方法，里面做了很多前期的準(zhǔn)備性工作，主要是加載bootstrap.conf里配置的屬性，以及在

里面構(gòu)建另外一個javacmd命令：

finalProcessBuilderbuilder=newProcessBuilder():

mand(cmd)

Processprocess=builder.startf);

所以這個start方法是啟動了另外一個java進(jìn)程,這個進(jìn)程才是真正的NiFiruntime。

通過代碼跟蹤或查看日志，可見，這個cmd命令類似如下：

/opt/jdkl.8.0_131/bin/java

-classpath

/opt/nifi-1.7.1/./conf:/opt/nifi-1.7.1/./lib/logback-core-1.2.3.jar:/opt/nifi-1.7.1/./lib/jetty-schemas-3.1.jar:/o

pt/nifi-1.7.1/./lib/logback-classic-1.2.3.jar:/opt/nifi-1.7.1/./lib/jul-to-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/jcl-o

ver-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/nifi-properties-1.7.1.jar:/opt/nifi-1.7.1/./lib/nifi-runtime-1.7.1.jar:/o

pt/nifi-1.7.1/./lib/nifi-framework-api-1.7.1.jar:/opt/nifi-1.7.1/./lib/nifi-nar-utils-1.7.1.jar:/opt/nifi-1.7.1/./lib/

javax.servlet-api-3.1.0.jar:/opt/nifi-1.7.1/./lib/log4j-over-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/slf4j-api-1.7.25

.jar:/opt/nifi-1.7.1/./lib/nifi-api-1.7.1.jar

-Dpiler.disablejsrl99=true

-Xmx3g-Xms3g

-Djavax.security.auth.useSubjectCredsOnly=true

-Djava.security.egd=file:/dev/urandom

-D.http.allowRestrictedHeaders=true

-D.preferIPv4Stack=true

-Djava.awt.headless=true-XX:+UseGlGC

-Dtocol,handler.pkgs=.tocol

-Duser.timezone=Asia/Shanghai

-Dperties.file.path=/opt/nifi-l.7.l/./conf/perties

-Dnifi.bootstrap.listen.port=56653

-Dapp=NiFi

以盯礪日o「g.apache.nifi.NiFi，

可以清晰的看到，命令中實(shí)際執(zhí)行的是java類org.apache.nifi.NiFi的main方法。

2.1.3.Nifi啟動初始化

這個org.apache.nifi.NiFi類在以下模塊中：

nifi-nar-bundles

nifi-framework-bundle

+---nifi-framework

+---nifi-runtime

Nifi-framework模塊就是nifi框架的核心代碼

Org.apache.nifi.NiFi.main。方法如下:

Mainentrypointoftheapplication.

*@paramargsthingswhichareignored

7

publicstaticvoidmain(String[]args){

LOGGERinfo("LaunchingNiFi...");

"y{

NiFiPropertiesproperties=convertArgumentsToValidatedNiFiProperties(args);

newNiFi(properties);

}catch(finalThrowablet){

LOGGERerror("FailuretolaunchNiFidueto"+1.1);

)

)

Main。方法調(diào)用了NiFi的構(gòu)造方法:

publicNiFi(finalNiFiPropertiesproperties)

throwsClassNotFoundException,lOException,NoSuchMethodException,InstantiationException,

HlegalAccessException,IHegalArgumentException.InvocationTargetException{

this(properties.ClassLoader.getSystemClassLoader());

}

publicNiFi(finalNiFiPropertiesproperties.ClassLoaderrootClassLoader)

throwsClassNotFoundException.lOException,NoSuchMethodException.InstantiationException,

HlegalAccessException,IHegalArgumentException,InvocationTargetException{

第二個構(gòu)造方法是實(shí)際上的構(gòu)造方法，里面進(jìn)行了大量初始化操作，以下是非常關(guān)鍵的部分:

//expandthenars

finalExtensionMappingextensionMapping=NarUnpacker.unpack/Vars(properties,systemBundle);

//loadtheextensionsclassloaders

NarClassLoadersnarClassLoaders=NarClassLoadersHolder.gef/nstence():

narClassLoaders.init(rootClassLoader,

properties.getFrameworkWorkingDirectoryO,properties.getExtensionsWorkingDirectoryO);

這部分初始化了NiFi的Java獷展工具箱，這些工具從Nifi的用戶來說就是在NiFi安裝目錄的lib目錄下

的各個*.nar包，這個些nar包實(shí)際就是NiFi增加了特定額外信息的jar包集合的壓縮，本質(zhì)上還是jar

包。以下是我們解壓開的一個nar包，結(jié)構(gòu)如下：

▼toMETA-INF

▼■maven

▼■org.apache.nifi

▼■nifi-pol-nar

Bperties

過pom.xml

▼■bundled-dependencies

Abcpkix-jdk15on-1.60.jar

Rbcprov-jdk15on-1.60.jar

0commons-codec-1.10.jar

Rcommons-collections4-4.2.jar

?commons-compress-1.18.jar

Acommons-csv-1.5.jar

Rconrwnons-io-2.6.jar

Acommons-lang3-3.8.1.jar

啟commons-text-1.4.jar

篇curvesapi-1.04.jar

Rjack$on-annotations-2.9.0.jar

口jackson-core-2.9.7.jar

Rjackson-databind-2.9.7.jar

啟jBCrypt-0.4.1.jar

Rnffi-poi-processors-1.9.2.jar

Anifi-processor-utils-1.9.2.jar

國nifi-security-utils-1.9.2.jar

Anifi-standard-record-utils-1.9.2.jar

Anifi-utils-1.9.2.jar

0poi-4.0.0.jar

Apoi-ooxml-4.0.0.jar

Rpoi-ooxml-schemas-4.0.0.jar

0xerceslmpl-2.11.0.jar

啟xml-apis-1.4.01jar

Rxmlbeans-3.0.1.jar

■DEPENDENCIES

■LICENSE

■NOTICE

■MANIFEST.MF

那么回到NiFi的構(gòu)造方法內(nèi)，首先是解壓這些nar包,并在代碼內(nèi)用ExtensionMapping對象描述,

代碼如下：

//expandthenars

finalExtensionMappingextensionMapping=NarUnpacker.unpackA/ars(properties.systemBundle);

然后加載并初始化這些類加載器:

//loadtheextensionsclassloaders

NarClassLoadersnarClassLoaders=NarClassLoadersHolder.^ef/nstance(),

narClassLoaders.init(rootClassLoader,

properties.getFrameworkWorkingDirectory(),properties.getExtensionsWorkingDirectoryO);

//loadtheframeworkclassloader

finalClassLoaderframeworkClassLoader=narClassLoaders.getFrameworkBundle().getClassLoader();

if(frameworkClassLoader==null){

thrownewHlegalStateException("UnabletofindtheframeworkNARClassLoader.")：

)

finalSet<Bundle>narBundles=narClassLoaders.getBundles():

在NiFi的官方介紹中，有兩處它的特性介紹是擴(kuò)展和類加載隔離,這里我們可以對它這兩個特性的實(shí)

現(xiàn)?探究竟了.它為每一個nar包構(gòu)造了一個獨(dú)立的自定義的類加載器：NarClassLoader

publicclassNarClassLoaderextendsURLCIassLoader{

privatestaticfinalLoggerLOGGER=LoggerFactory.getLogger(NarClassLoader.class);

privatestaticfinalFileFilterJAR_FILTER-newFileFilter(){

@Override

publicbooleanaccept(Filepathname){

finalStringnameToTest=pathname.getName().toLowerCase();

returnnameToTest.endsWith(".jar")&&pathname.isFile();

}

}；

目前基本清晰，NiFi的擴(kuò)展性是由自定義的壓縮文件nar包和自定義的類加載器來提供的.接

著往下看：

//loadtheserverfromtheframeworkclassloader

Thread.currenf777read().setContextClassLoader(frameworkClassLoader);

Class<?>jettyServer=Class.for/VameCorg.apache.ni??.web.server.JettyServer.true,frameworkClassLoader);

Constructor<?>jettyConstructor=jettyServer.getConstructor(NiFiProperties.class.Set.class);

finallongstartTime=System.nanoT/mef);

nifiServer=(NiFiServer)jettyConstructor.newlnstance(properties.narBundles):

nifiServer.setExtensionMapping(extensionMapping);

nifiServer.setBundles(systemBundle.narBundles);

回想架構(gòu)圖VM的最上層是webserver,這個webserver就是在這里被加載了，這是?個jetty

server，繼續(xù)往下看：

if(shutdown){

LOGG￡Rinfo("NiFihasbeenshutdownviaNiFiBootstrap.WillnotstartController");

}else{

nifiServer.start();

if(bootstrapListener!=null){

bootstrapListener.sendStartedStatus(true);

)

finallongduration=System.nanoT/meO?startTime:

LOGGE/7.info("Controllerinitializationtook"+duration+"nanoseconds"

+"("+(int)TimeUnit.SECO/VDS.convert(duration.T\rne\Jn\{.NANOSECONDS)+"seconds)."):

)

start這個nifiServer,這個NiFi對象的構(gòu)造方法這里就全部走完了.

2.1.4.NiFi-Web

NiFiServer.start。的方法內(nèi)，代碼跳轉(zhuǎn)到nifi-framework下的一個了?模塊

nifi-web內(nèi)了。

〈parent〉

?*2nifi-user-actions

<groupld>org.apache.nifi</groupld>

▼*2nifi-web<artifactld>nifi-framework</artifactld>

?*2nifi-custom-ui-utilities<version>l.10.0</version>

?nifi-jetty</parent>

?nifi-ui-extension<artifactld>nifi-web</artifactld>

?*￡nifi-web-api<packaging>pom</packaging>

?*2nifi-web-content-access<modules>

?nifi-web-content-viewer<module>nifi-web-optimistic-l.ocking</inodule>

<module>nifi-custom-ui-utilities</modul.e>

?*￡nifi?web-docs

<module>nifi-web-security</modul.e>

?*2nifi<web-error

<module>nifi-web-api</module>

?*2nifi-web-optimistic-locking<module>nifi-web-error</module>

?nifi-web-security<module>nifi-web-docs</modul.e>

?*￡nifi-web-ui<module>nifi-web-content-viewer</module>

??target<module>nifi-web-ui</modul.e>

他nifi-web.iml<module>nifi-jetty</module>

mpom.xml<n>odule>nifi-web-content-access</module>

<module>nifi-ui-extension</module>

?由target

〈/modules〉

_____rUCraEawcrUIEI

2.1.5.nifi-jetty

與Web相關(guān)的代碼都在這個模塊了，包括Server和界面相關(guān)的代碼，上面提到的NiFiServer的實(shí)現(xiàn)

類JettyServer就在子模塊nifi-jetty內(nèi)了。

▼!(nifi-web>1

?*2nifl*custom*ui*utilities

937eOverride

▼hsrc9385publicvoid|rtrt(|>{

try《

▼?main

〃Createastandardextensionmanageranddiscoverextensions

?feidssembly

finalExtensionDiscoveringMandgerextensioritanager=newStandardExtensionOisc

extensiorManager.dIscoverExtension*(systertundle,bundles)：

▼Biorg.apache.nifi.web.serverextenslorHaMQer.logClas$Loaderttapplng();

?GHostHeaderHandler

▼OJettyServerJava//Settheextensionmanagerintotheholderwhichmakesitavailabletothe5

??JettyServerExtensiorHanagerttol.der.jnit(extensiorManager);

??ServerConnectorCreator

?GServerConfigurationExceptionHGeneratedocsforextensions

DoeGenerator.gener?te(props,extensionManager,extensionKapping);

?feSresources

?totest

接著看JettyServer這個類，上面的NiFi構(gòu)造方法里面最后是先實(shí)例化了這個JettyServer,然

后調(diào)用了start方法.先看它的構(gòu)造方法，只看注釋，找到了核心方法：

//loadwarsfromthebundle

finalHandlerwarHandlers=loadlnitialWars(bundles),

可以看到，其實(shí)就是把war包加載進(jìn)來了，這些war包就是nifi-web下面的子工程，有幾個子工

程的pom文件中配置的就是<packaging>war〈/packaging>

接著看這個Start方法：

第一句就是ExtensionManager.discoverExtensions(systemBundle,bundles);就是這里把所有的

擴(kuò)展類加載進(jìn)JVM了，看到看到ExtensionManager的注釋，這個注釋就說明了一切

ScansthroughtheclasspathtoloadallFlowFHeProcessors,FlowFileComparators,andReportingTasksusingthe

serviceproviderAPIandrunningthroughallclassloaders(root,NARs).

這個ExtensionManager在加載類的時候,用到了java的一種比較高級的機(jī)制，java

SPI(serviceproviderinterface),這種機(jī)制在很多框架中比如spring中大量使用

finalServiceLoader<?>serviceLoader=ServiceLoader./oad(entry.getKey().bundle.getClassLoader()):

這個機(jī)制解釋了為什么寫自定義的處理器的時候要在/resources/META-INF/services目錄下面

寫上配置文件.在自定義處理開發(fā)的時候，一定要注意寫這個配置文件，否則類是加載不進(jìn)來的

接著start這個jettyserver,接著往下看，只看注釋，可以看到,大致就是做了