MySqlSourceReader类核心功能
数据分片
执行start方法向MySqlSourceEnumerator发起分片请求, 触发MySqlSourceEnumerator类的handleSplitRequest方法, 并向reader分配分片
public void start() {
if (getNumberOfCurrentlyAssignedSplits() <= 1) {
context.sendSplitRequest();
}
}
分片处理
执行addSplits方法, 处理MySqlSourceEnumerator分配的chunk分片
private void addSplits(List<MySqlSplit> splits, boolean checkTableChangeForBinlogSplit) {
// restore for finishedUnackedSplits
// 存放未处理的分片
List<MySqlSplit> unfinishedSplits = new ArrayList<>();
for (MySqlSplit split : splits) {
LOG.info("Source reader {} adds split {}", subtaskId, split);
// 全量阶段
if (split.isSnapshotSplit()) {
MySqlSnapshotSplit snapshotSplit = split.asSnapshotSplit();
// 判断该分片是否已完成, 如已完成存放到finishedUnackedSplits map中
if (snapshotSplit.isSnapshotReadFinished()) {
finishedUnackedSplits.put(snapshotSplit.splitId(), snapshotSplit);
} else if (sourceConfig
.getTableFilters()
.dataCollectionFilter()
.isIncluded(split.asSnapshotSplit().getTableId())) {
// 如匹配到配置的同步表, 放入未处理分片的列表中
unfinishedSplits.add(split);
} else {
LOG.debug(
"The subtask {} is skipping split {} because it does not match new table filter.",
subtaskId,
split.splitId());
}
} else {
// Binlog阶段
MySqlBinlogSplit binlogSplit = split.asBinlogSplit();
// When restore from a checkpoint, the finished split infos may contain some splits
// for the deleted tables.
// We need to remove these splits for the deleted tables at the finished split
// infos.
// 如果从CK恢复, 需移除可能被删除的同步表
if (checkTableChangeForBinlogSplit) {
binlogSplit =
filterOutdatedSplitInfos(
binlogSplit,
sourceConfig
.getMySqlConnectorConfig()
.getTableFilters()
.dataCollectionFilter());
}
// Try to discovery table schema once for newly added tables when source reader
// start or restore
// 是否有新增同步表
boolean checkNewlyAddedTableSchema =
!mySqlSourceReaderContext.isHasAssignedBinlogSplit()
&& sourceConfig.isScanNewlyAddedTableEnabled();
mySqlSourceReaderContext.setHasAssignedBinlogSplit(true);
// the binlog split is suspended
if (binlogSplit.isSuspended()) {
suspendedBinlogSplit = binlogSplit;
} else if (!binlogSplit.isCompletedSplit()) {
uncompletedBinlogSplits.put(binlogSplit.splitId(), binlogSplit);
requestBinlogSplitMetaIfNeeded(binlogSplit);
} else {
uncompletedBinlogSplits.remove(binlogSplit.splitId());
MySqlBinlogSplit mySqlBinlogSplit =
discoverTableSchemasForBinlogSplit(
binlogSplit, sourceConfig, checkNewlyAddedTableSchema);
unfinishedSplits.add(mySqlBinlogSplit);
}
LOG.info(
"Source reader {} received the binlog split : {}.", subtaskId, binlogSplit);
context.sendSourceEventToCoordinator(new BinlogSplitAssignedEvent());
}
}
// notify split enumerator again about the finished unacked snapshot splits
reportFinishedSnapshotSplitsIfNeed();
// add all un-finished splits (including binlog split) to SourceReaderBase
// 调用父类SourceReaderBase.addSplits方法, 添加 un-finished splits
if (!unfinishedSplits.isEmpty()) {
super.addSplits(unfinishedSplits);
} else if (suspendedBinlogSplit
!= null) { // only request new snapshot split if the binlog split is suspended
context.sendSplitRequest();
}
}
继续调用SourceReaderBase.addSplits方法
@Override
public void addSplits(List<SplitT> splits) {
LOG.info("Adding split(s) to reader: {}", splits);
// Initialize the state for each split.
splits.forEach(
s ->
splitStates.put(
s.splitId(), new SplitContext<>(s.splitId(), initializedState(s))));
// Hand over the splits to the split fetcher to start fetch.
splitFetcherManager.addSplits(splits);
}
继续调用splitFetcherManager.addSplits方法, 其中splitFetcherManager为SingleThreadFetcherManager对象, 在MySqlSourceReader构造函数中进行的初始化
@Override
public void addSplits(List<SplitT> splitsToAdd) {
// 获取正在运行的fetcher
SplitFetcher<E, SplitT> fetcher = getRunningFetcher();
if (fetcher == null) {
/**
* 创建SplitFetcher, 并加入到fetchers集合中
* */
fetcher = createSplitFetcher();
// Add the splits to the fetchers.
// 创建AddSplitsTask任务, 并加入到taskQueue中待执行
fetcher.addSplits(splitsToAdd);
// 启动执行fetcher
startFetcher(fetcher);
} else {
// 创建AddSplitsTask任务, 并加入到taskQueue中待执行
fetcher.addSplits(splitsToAdd);
}
}
继续查看SplitFetcher run方法
@Override
public void run() {
LOG.info("Starting split fetcher {}", id);
try {
while (runOnce()) {
// nothing to do, everything is inside #runOnce.
}
} catch (Throwable t) {
errorHandler.accept(t);
} finally {
try {
splitReader.close();
} catch (Exception e) {
errorHandler.accept(e);
} finally {
LOG.info("Split fetcher {} exited.", id);
// This executes after possible errorHandler.accept(t). If these operations bear
// a happens-before relation, then we can checking side effect of
// errorHandler.accept(t)
// to know whether it happened after observing side effect of shutdownHook.run().
shutdownHook.run();
}
}
}
继续调用runOnce()
boolean runOnce() {
// first blocking call = get next task. blocks only if there are no active splits and queued
// tasks.
SplitFetcherTask task;
lock.lock();
try {
if (closed) {
return false;
}
// 从taskQueue获取待执行的任务
task = getNextTaskUnsafe();
if (task == null) {
// (spurious) wakeup, so just repeat
return true;
}
LOG.debug("Prepare to run {}", task);
// store task for #wakeUp
this.runningTask = task;
} finally {
lock.unlock();
}
// execute the task outside of lock, so that it can be woken up
boolean taskFinished;
try {
// 执行SplitFetcherTask run方法
taskFinished = task.run();
} catch (Exception e) {
throw new RuntimeException(
String.format(
"SplitFetcher thread %d received unexpected exception while polling the records",
id),
e);
}
// re-acquire lock as all post-processing steps, need it
lock.lock();
try {
this.runningTask = null;
processTaskResultUnsafe(task, taskFinished);
} finally {
lock.unlock();
}
return true;
}
继续调用AddSplitsTask run方法, 这里的splitReader对象是MySqlSplitReader, 执行其handleSplitsChanges方法,
@Override
public boolean run() {
for (SplitT s : splitsToAdd) {
assignedSplits.put(s.splitId(), s);
}
splitReader.handleSplitsChanges(new SplitsAddition<>(splitsToAdd));
return true;
}
handleSplitsChanges方法只是把MySqlSplit添加到对应的snapshotSplits或者binlogSplits列表中
@Override
public void handleSplitsChanges(SplitsChange<MySqlSplit> splitsChanges) {
if (!(splitsChanges instanceof SplitsAddition)) {
throw new UnsupportedOperationException(
String.format(
"The SplitChange type of %s is not supported.",
splitsChanges.getClass()));
}
LOG.info("Handling split change {}", splitsChanges);
for (MySqlSplit mySqlSplit : splitsChanges.splits()) {
if (mySqlSplit.isSnapshotSplit()) {
snapshotSplits.add(mySqlSplit.asSnapshotSplit());
} else {
binlogSplits.add(mySqlSplit.asBinlogSplit());
}
}
}
在前面createSplitFetcher方法创建SplitFetcher对象时, 其构造函数中会创建FetchTask
SplitFetcher(
int id,
FutureCompletingBlockingQueue<RecordsWithSplitIds<E>> elementsQueue,
SplitReader<E, SplitT> splitReader,
Consumer<Throwable> errorHandler,
Runnable shutdownHook,
Consumer<Collection<String>> splitFinishedHook,
boolean allowUnalignedSourceSplits) {
this.id = id;
this.elementsQueue = checkNotNull(elementsQueue);
this.splitReader = checkNotNull(splitReader);
this.errorHandler = checkNotNull(errorHandler);
this.shutdownHook = checkNotNull(shutdownHook);
this.allowUnalignedSourceSplits = allowUnalignedSourceSplits;
// 创建fetchTask, 其中run 方法中会调用splitReader.fetch()方法
this.fetchTask =
new FetchTask<>(
splitReader,
elementsQueue,
ids -> {
ids.forEach(assignedSplits::remove);
splitFinishedHook.accept(ids);
LOG.info("Finished reading from splits {}", ids);
},
id);
}
最终调用MySqlSplitReader.fetch()方法去拉取数据, 由DebeziumReader读取数据
