MySqlSourceEnumerator類核心方法

start方法

 @Override
    public void start() {
        // 如果啟動(dòng)參數(shù)是INITIAL
       // splitAssigner 是MySqlHybridSplitAssigner, 否則是MySqlBinlogSplitAssigner
        // 啟動(dòng) SplitAssigner
        splitAssigner.open();
        // 如果是新增同步表, 重新啟動(dòng)CDC任務(wù)場(chǎng)景下: 
       //當(dāng)新添加的表的快照完成分配時(shí)，請(qǐng)求更新binlog的分割狀態(tài)
        requestBinlogSplitUpdateIfNeed();
        // 周期性的執(zhí)行syncWithReaders 方法
        this.context.callAsync(
                this::getRegisteredReader,
                this::syncWithReaders,
                CHECK_EVENT_INTERVAL,
                CHECK_EVENT_INTERVAL);
    }

 /**
     * 1、檢查是否新添加的表的快照已經(jīng)完成分配。
     * 2、如果是，則遍歷已注冊(cè)的讀取器。
     * 3、對(duì)每個(gè)讀取器，發(fā)送更新binlog分割狀態(tài)的請(qǐng)求事件。
     */
    private void requestBinlogSplitUpdateIfNeed() {
        if (isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())) {
            for (int subtaskId : getRegisteredReader()) {
                LOG.info(
                        "The enumerator requests subtask {} to update the binlog split after newly added table.",
                        subtaskId);
                context.sendEventToSourceReader(subtaskId, new BinlogSplitUpdateRequestEvent());
            }
        }
    }

   /**
     * 當(dāng)SourceEnumerator恢復(fù)或者SourceEnumerator和SourceReader之間的通信失敗時(shí)，
     * 可能會(huì)錯(cuò)過一些通知事件。告訴所有的SourceReader報(bào)告它們已完成但未確認(rèn)的分片。
     */
    private void syncWithReaders(int[] subtaskIds, Throwable t) {
        if (t != null) {
            throw new FlinkRuntimeException("Failed to list obtain registered readers due to:", t);
        }
        // when the SourceEnumerator restores or the communication failed between
        // SourceEnumerator and SourceReader, it may missed some notification event.
        // tell all SourceReader(s) to report there finished but unacked splits.
        if (splitAssigner.waitingForFinishedSplits()) {
            for (int subtaskId : subtaskIds) {
                context.sendEventToSourceReader(
                        subtaskId, new FinishedSnapshotSplitsRequestEvent());
            }
        }

        requestBinlogSplitUpdateIfNeed();
    }

其中splitAssigner.open()方法后續(xù)再分析

handleSplitRequest 方法

接收來自 reader 的請(qǐng)求，對(duì)數(shù)據(jù)進(jìn)行分片,通過 assignSplits 對(duì)分片進(jìn)行分配, 注意assignSplits方法在CK完成回調(diào)notifyCheckpointComplete方法中也會(huì)調(diào)用, 如果已經(jīng)是split分配完成后的一個(gè)ck，就可以標(biāo)志著全量階段已經(jīng)完成，可以開始準(zhǔn)備下發(fā)BinlogSplit了

 @Override
    public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
        if (!context.registeredReaders().containsKey(subtaskId)) {
            // reader failed between sending the request and now. skip this request.
            return;
        }

        // 接收來自reader的分片請(qǐng)求, 根據(jù)reader的taskId, 放入到readersAwaitingSplit TreeSet中, 表示等待分配split的reader請(qǐng)求
        readersAwaitingSplit.add(subtaskId);
        // 為reader分配split, 分配完成后, 從readersAwaitingSplit集合中 remove掉
        assignSplits();
    }

     private void assignSplits() {
        final Iterator<Integer> awaitingReader = readersAwaitingSplit.iterator();

        while (awaitingReader.hasNext()) {
            int nextAwaiting = awaitingReader.next();
            // if the reader that requested another split has failed in the meantime, remove
            // it from the list of waiting readers
            // 判斷當(dāng)前注冊(cè)的reader集合是否存在
            if (!context.registeredReaders().containsKey(nextAwaiting)) {
                awaitingReader.remove();
                continue;
            }

            // 判斷當(dāng)前的Snapshot全量階段是否完成, 如果完成根據(jù)設(shè)置的closeIdleReaders參數(shù), 是否關(guān)閉空閑Reader
            /**
             *  Flink CDC 的增量快照框架有兩個(gè)主要階段: 全量階段和增量階段。
             *  這兩個(gè)階段的并行度并不相同，全量階段支持多并行度，加快大量數(shù)據(jù)的同步過程，
             *  增量階段讀取Binlog變更日志，需要使用單并發(fā)保證事件的順序和正確性。
             *  在全量階段讀取結(jié)束后，由于增量階段只需要一個(gè)并發(fā)，會(huì)出現(xiàn)大量的空閑 Reader，比較浪費(fèi)資源。
             *  2.4 版本使用增量快照連接器時(shí)，支持配置打開自動(dòng)關(guān)閉空閑 Reader 的功能來關(guān)閉這些空閑 Reader
             */
            if (splitAssigner.isStreamSplitAssigned()
                    && sourceConfig.isCloseIdleReaders()
                    && noMoreSnapshotSplits()
                    && (binlogSplitTaskId != null && !binlogSplitTaskId.equals(nextAwaiting))) {
                // close idle readers when snapshot phase finished.
                // 向reader 發(fā)送關(guān)閉請(qǐng)求, 釋放資源
                context.signalNoMoreSplits(nextAwaiting);
                awaitingReader.remove();
                LOG.info("Close idle reader of subtask {}", nextAwaiting);
                continue;
            }

            // TODO: 待分析
            Optional<MySqlSplit> split = splitAssigner.getNext();
            if (split.isPresent()) {
                final MySqlSplit mySqlSplit = split.get();
                /**
                 *  向reader 分配MySqlSplit
                 */
                context.assignSplit(mySqlSplit, nextAwaiting);
                if (mySqlSplit instanceof MySqlBinlogSplit) {
                    this.binlogSplitTaskId = nextAwaiting;
                }
                awaitingReader.remove();
                LOG.info("The enumerator assigns split {} to subtask {}", mySqlSplit, nextAwaiting);
            } else {
                // there is no available splits by now, skip assigning
                requestBinlogSplitUpdateIfNeed();
                break;
            }
        }
    }

handleSourceEvent方法

處理 SourceEvent，是 SplitEnumerator 和 SourceReader 之間來回傳遞的自定義事件?？梢岳么藱C(jī)制來執(zhí)行復(fù)雜的協(xié)調(diào)任務(wù)
處理事件類型:

• FinishedSnapshotSplitsReportEvent: reader完成讀取chunk數(shù)據(jù)事件
• BinlogSplitMetaRequestEvent: reader請(qǐng)求分配BinlogSplit
• BinlogSplitUpdateAckEvent: BinlogSplit分配確認(rèn)ACK事件
• LatestFinishedSplitsNumberRequestEvent: 請(qǐng)求獲取當(dāng)前最新已完成的Snapshot chunk數(shù)量
• BinlogSplitAssignedEvent: reader已分配到BinlogSplit事件

 @Override
    public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
        // 接收到reader FinishedSnapshotSplitsReportEvent 事件
        if (sourceEvent instanceof FinishedSnapshotSplitsReportEvent) {
            LOG.info(
                    "The enumerator under {} receives finished split offsets {} from subtask {}.",
                    splitAssigner.getAssignerStatus(),
                    sourceEvent,
                    subtaskId);
            FinishedSnapshotSplitsReportEvent reportEvent =
                    (FinishedSnapshotSplitsReportEvent) sourceEvent;
            Map<String, BinlogOffset> finishedOffsets = reportEvent.getFinishedOffsets();

            // 在Snapshot全量階段, 保存每個(gè)reader讀取chunk完成時(shí)候的binlog點(diǎn)位
            splitAssigner.onFinishedSplits(finishedOffsets);
            requestBinlogSplitUpdateIfNeed();

            // send acknowledge event
            // 發(fā)送ack event
            FinishedSnapshotSplitsAckEvent ackEvent =
                    new FinishedSnapshotSplitsAckEvent(new ArrayList<>(finishedOffsets.keySet()));
            context.sendEventToSourceReader(subtaskId, ackEvent);
        } else if (sourceEvent instanceof BinlogSplitMetaRequestEvent) { //  處理BinlogSplitMetaRequestEvent事件請(qǐng)求
            LOG.debug(
                    "The enumerator receives request for binlog split meta from subtask {}.",
                    subtaskId);
            sendBinlogMeta(subtaskId, (BinlogSplitMetaRequestEvent) sourceEvent);
        } else if (sourceEvent instanceof BinlogSplitUpdateAckEvent) {
            LOG.info(
                    "The enumerator receives event that the binlog split has been updated from subtask {}. ",
                    subtaskId);
            splitAssigner.onBinlogSplitUpdated();
        } else if (sourceEvent instanceof LatestFinishedSplitsNumberRequestEvent) {
            LOG.info(
                    "The enumerator receives request from subtask {} for the latest finished splits number after added newly tables. ",
                    subtaskId);
            // 告訴reader 當(dāng)前已完成的Snapshot FinishedSplitInfos size
            handleLatestFinishedSplitNumberRequest(subtaskId);
        } else if (sourceEvent instanceof BinlogSplitAssignedEvent) { // 處理BinlogSplitAssignedEvent事件
            LOG.info(
                    "The enumerator receives notice from subtask {} for the binlog split assignment. ",
                    subtaskId);
            binlogSplitTaskId = subtaskId;
        }
    }

至此已梳理MySqlSourceEnumerator類中的核心方法及作用, 但是全量階段的chunk劃分及向reader分配MySqlSplit邏輯在哪呢?

• chunk劃分: 在start方法中的splitAssigner.open() 實(shí)現(xiàn)
• MySqlSplit分配: handleSplitRequest()->assignSplits()->splitAssigner.getNext()