由于nifi cluster的數(shù)據(jù)分發(fā)只能使用Remote ProcessGroup沛豌，然后通過(guò)site-to-site協(xié)議來(lái)分發(fā)數(shù)據(jù)到各個(gè)nifi instance。所以看了Remote ProcessGroup的源碼着绊，現(xiàn)將理解記錄下來(lái)。
RemoteProcessGroup
org.apache.nifi.groups.RemoteProcessGroup是Remote ProcessGroup的接口熟尉，主要函數(shù)為：
/**
* Initiates communications between this instance and the remote instance.
*/
void startTransmitting();
/**
* Immediately terminates communications between this instance and the
* remote instance.
*/
void stopTransmitting();
org.apache.nifi.remote.StandardRemoteProcessGroup實(shí)現(xiàn)了RemoteProcessGroup接口归露，在函數(shù)startTransmitting()中啟動(dòng)了綁定給它的StandardRemoteGroupPort。
StandardRemoteGroupPort
org.apache.nifi.remote.StandardRemoteGroupPort由RemoteProcessGroup啟動(dòng)斤儿，是實(shí)現(xiàn)數(shù)據(jù)傳輸?shù)闹饕惥绨饕獙?duì)象是SiteToSiteClient，主要函數(shù)由onSchedulingStart()和onTrigger()往果。
public void onSchedulingStart() {
super.onSchedulingStart();
final long penalizationMillis = FormatUtils.getTimeDuration(remoteGroup.getYieldDuration(), TimeUnit.MILLISECONDS);
final SiteToSiteClient client = new SiteToSiteClient.Builder()
.url(remoteGroup.getTargetUri().toString())
.portIdentifier(getIdentifier())
.sslContext(sslContext)
.useCompression(isUseCompression())
.eventReporter(remoteGroup.getEventReporter())
.peerPersistenceFile(getPeerPersistenceFile(getIdentifier(), nifiProperties))
.nodePenalizationPeriod(penalizationMillis, TimeUnit.MILLISECONDS)
.timeout(remoteGroup.getCommunicationsTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
.transportProtocol(remoteGroup.getTransportProtocol())
.httpProxy(new HttpProxy(remoteGroup.getProxyHost(), remoteGroup.getProxyPort(), remoteGroup.getProxyUser(), remoteGroup.getProxyPassword()))
.build();
clientRef.set(client);
}
onSchedulingStart()函數(shù)創(chuàng)建SiteToSiteClient疆液，這是site-to-site協(xié)議的主要client接口。
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) {
...
final SiteToSiteClient client = getSiteToSiteClient();
final Transaction transaction;
transaction = client.createTransaction(transferDirection);
...
transferFlowFiles(transaction, context, session, firstFlowFile);
....
}
}
onTrigger()函數(shù)用來(lái)傳輸數(shù)據(jù)到另外的nifi instance陕贮，此處用到了Transaction來(lái)傳輸數(shù)據(jù)堕油。
SiteToSiteClient
===========
SiteToSiteClient是site-to-site協(xié)議的接口，分為兩種實(shí)現(xiàn)SocketClient和HttpClient兩種，具體對(duì)應(yīng)Remote ProcessGroup界面配置中的兩種協(xié)議掉缺，這里主要討論SocketClient的實(shí)現(xiàn)卜录。
org.apache.nifi.remote.client.socket.SocketClient通過(guò)createTransaction（）函數(shù)來(lái)創(chuàng)建SocketClientTransaction。
@Override
public Transaction createTransaction(final TransferDirection direction) throws IOException {
...
final EndpointConnection connectionState = pool.getEndpointConnection(direction, getConfig());
if (connectionState == null) {
return null;
}
final Transaction transaction;
try {
transaction = connectionState.getSocketClientProtocol().startTransaction(
connectionState.getPeer(), connectionState.getCodec(), direction);
} catch (final Throwable t) {
pool.terminate(connectionState);
throw new IOException("Unable to create Transaction to communicate with " + connectionState.getPeer(), t);
}
...
}
EndpointConnectionPool
此處的pool為EndpointConnectionPool眶明，它是一個(gè)EndpointConnection池艰毒，保存了EndpointConnection的map。
private final ConcurrentMap<PeerDescription, BlockingQueue<EndpointConnection>> connectionQueueMap = new ConcurrentHashMap<>();
另外搜囱，它保存了一個(gè)遠(yuǎn)端鏈接信息PeerStatus的選擇器PeerSelector.
private final PeerSelector peerSelector;
EndpointConnection
EndpointConnection就是實(shí)際上與遠(yuǎn)端nifi instance的鏈接,它保存了與遠(yuǎn)端的數(shù)據(jù)通道Peer和handshake工具對(duì)象SocketClientProtocol丑瞧。
PeerSelector
它提供了獲得下一個(gè)PeerStatus函數(shù)getNextPeerStatus（）跟新鏈接信息的函數(shù)refreshPeers()。
/**
* Return status of a peer that will be used for the next communication.
* The peer with less workload will be selected with higher probability.
* @param direction the amount of workload is calculated based on transaction direction,
* for SEND, a peer with less flow files is preferred,
* for RECEIVE, a peer with more flow files is preferred
* @return a selected peer, if there is no available peer or all peers are penalized, then return null
*/
public PeerStatus getNextPeerStatus(final TransferDirection direction) {
List<PeerStatus> peerList = peerStatuses;
if (isPeerRefreshNeeded(peerList)) {
peerRefreshLock.lock();
try {
// now that we have the lock, check again that we need to refresh (because another thread
// could have been refreshing while we were waiting for the lock).
peerList = peerStatuses;
if (isPeerRefreshNeeded(peerList)) {
try {
peerList = createPeerStatusList(direction);
} catch (final Exception e) {
final String message = String.format("%s Failed to update list of peers due to %s", this, e.toString());
warn(logger, eventReporter, message);
if (logger.isDebugEnabled()) {
logger.warn("", e);
}
}
this.peerStatuses = peerList;
peerRefreshTime = systemTime.currentTimeMillis();
}
} finally {
peerRefreshLock.unlock();
}
}
if (peerList == null || peerList.isEmpty()) {
return null;
}
}
這里的createPeerStatusList會(huì)根據(jù)遠(yuǎn)端nifi instance上端口接收或者發(fā)送的flowfiles的數(shù)量和direction對(duì)peer排序蜀肘。當(dāng)是在發(fā)送數(shù)據(jù)的時(shí)候嗦篱，選取目前需要接收的數(shù)據(jù)最少的nifi instance對(duì)應(yīng)的peer來(lái)發(fā)送數(shù)據(jù)。當(dāng)是在接收數(shù)據(jù)時(shí)幌缝，選取目前需要發(fā)送的數(shù)據(jù)量最大的nifi instance對(duì)應(yīng)的peer來(lái)接收數(shù)據(jù)。通過(guò)這個(gè)機(jī)制來(lái)達(dá)到cluster的負(fù)載均衡诫欠。
private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException {
final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>();
// Look at all of the peers that we fetched last time.
final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers;
if (lastFetched != null && !lastFetched.isEmpty()) {
lastFetched.stream().map(peer -> peer.getPeerDescription())
.forEach(desc -> peersToRequestClusterInfoFrom.add(desc));
}
// Always add the configured node info to the list of peers to communicate with
peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription());
logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom);
Exception lastFailure = null;
for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) {
try {
final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription);
lastFetchedQueryablePeers = statuses.stream()
.filter(p -> p.isQueryForPeers())
.collect(Collectors.toSet());
return statuses;
} catch (final Exception e) {
logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}",
peerDescription.getHostname(), peerDescription.getPort(), e.toString());
lastFailure = e;
}
}
final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster");
if (lastFailure != null) {
ioe.addSuppressed(lastFailure);
}
throw ioe;
}
refreshPeers函數(shù)調(diào)用了fetchRemotePeerStatuses()函數(shù),這個(gè)函數(shù)中對(duì)于每一個(gè)之前記錄過(guò)的peer都會(huì)去訪問(wèn)以跟新整個(gè)集群狀態(tài)涵卵，所以只要第一次啟動(dòng)的時(shí)候配置的url nifi instance是有效的，以后就算它掛了荒叼，理論上也是不影響其他節(jié)點(diǎn)數(shù)據(jù)的正常發(fā)送和接收的轿偎。
而且refreshPeers()函數(shù)是在EndpointConnectionPool構(gòu)造的時(shí)候，去不斷調(diào)用的被廓，所以site-to-site協(xié)議會(huì)不斷的跟新cluster中節(jié)點(diǎn)的狀態(tài)坏晦，保證數(shù)據(jù)的可靠性。
taskExecutor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
peerSelector.refreshPeers();
}
}, 0, 5, TimeUnit.SECONDS);
