由于nifi cluster的數(shù)據(jù)分發(fā)只能使用Remote ProcessGroup，然后通過site-to-site協(xié)議來分發(fā)數(shù)據(jù)到各個(gè)nifi instance。所以看了Remote ProcessGroup的源碼，現(xiàn)將理解記錄下來。

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ù)用來傳輸數(shù)據(jù)到另外的nifi instance，此處用到了Transaction來傳輸數(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通過createTransaction（）函數(shù)來創(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來發(fā)送數(shù)據(jù)。當(dāng)是在接收數(shù)據(jù)時(shí)，選取目前需要發(fā)送的數(shù)據(jù)量最大的nifi instance對(duì)應(yīng)的peer來接收數(shù)據(jù)。通過這個(gè)機(jī)制來達(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è)之前記錄過的peer都會(huì)去訪問以跟新整個(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);