MySqlSourceEnumerator類核心方法
start方法
@Override
public void start() {
// 如果啟動(dòng)參數(shù)是INITIAL
// splitAssigner 是MySqlHybridSplitAssigner, 否則是MySqlBinlogSplitAssigner
// 啟動(dòng) SplitAssigner
splitAssigner.open();
// 如果是新增同步表, 重新啟動(dòng)CDC任務(wù)場景下:
//當(dāng)新添加的表的快照完成分配時(shí)外臂，請求更新binlog的分割狀態(tài)
requestBinlogSplitUpdateIfNeed();
// 周期性的執(zhí)行syncWithReaders 方法
this.context.callAsync(
this::getRegisteredReader,
this::syncWithReaders,
CHECK_EVENT_INTERVAL,
CHECK_EVENT_INTERVAL);
}
/**
* 1、檢查是否新添加的表的快照已經(jīng)完成分配蝌焚。
* 2脓鹃、如果是毡鉴，則遍歷已注冊的讀取器刁标。
* 3、對每個(gè)讀取器蚓庭，發(fā)送更新binlog分割狀態(tài)的請求事件致讥。
*/
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 的請求，對數(shù)據(jù)進(jìn)行分片,通過 assignSplits 對分片進(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的分片請求, 根據(jù)reader的taskId, 放入到readersAwaitingSplit TreeSet中, 表示等待分配split的reader請求
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)前注冊的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)閉請求, 釋放資源
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 之間來回傳遞的自定義事件嘴拢∽ぃ可以利用此機(jī)制來執(zhí)行復(fù)雜的協(xié)調(diào)任務(wù)
處理事件類型:
- FinishedSnapshotSplitsReportEvent: reader完成讀取chunk數(shù)據(jù)事件
- BinlogSplitMetaRequestEvent: reader請求分配BinlogSplit
- BinlogSplitUpdateAckEvent: BinlogSplit分配確認(rèn)ACK事件
- LatestFinishedSplitsNumberRequestEvent: 請求獲取當(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事件請求
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()
