Datax 数据传输源码简单分析

数据传输大致过程：

datax采用插件模式设计，reader与reade之间，reader与writer之间完全解耦，可做到互不影响。datax有三大部分，reader，writer，channel，reader和writer间依赖channel传输数据，reader通过recordSender.sendtoWriter()往channel写入数据，writer通过recordReceiver.getFromReader()从channel拉取数据，channel的底层是一个队列，先进先出（ArrayBlockingQueue）。

作业运行设计：

Job: Job是DataX用以描述从一个源头到一个目的端的同步作业，是DataX数据同步的最小业务单元。比如：从一张mysql的表同步到odps的一个表的特定分区。

Task: Task是为最大化而把Job拆分得到的最小执行单元。比如：读一张有1024个分表的mysql分库分表的Job，拆分成1024个读Task，用若干个并发执行。

TaskGroup: 描述的是一组Task集合。在同一个TaskGroupContainer执行下的Task集合称之为TaskGroup

JobContainer: Job执行器，负责Job全局拆分、调度、前置语句和后置语句等工作的工作单元。类似Yarn中的JobTracker。

TaskGroupContainer: TaskGroup执行器，负责执行一组Task的工作单元，类似Yarn中的TaskTracker。

重点源码：

下面以MySQL为例，进行源代码分析

读取过程源码分析

// MySQL 读取 源代码
@Override
public void startRead(RecordSender recordSender) {
    int fetchSize = this.readerSliceConfig.getInt(Constant.FETCH_SIZE);

    this.commonRdbmsReaderTask.startRead(this.readerSliceConfig, recordSender,
            super.getTaskPluginCollector(), fetchSize);
}

// 调用 rdbms 的 startRead 源代码，重点看 transportOneRecord 方法
public void startRead(Configuration readerSliceConfig,
                      RecordSender recordSender,
                      TaskPluginCollector taskPluginCollector, int fetchSize) {
    ......
        while (rs.next()) {
            if (limit != null && i > limit) {
                break;
            }
            rsNextUsedTime += (System.nanoTime() - lastTime);
            this.transportOneRecord(recordSender, rs,
                    metaData, columnNumber, mandatoryEncoding, taskPluginCollector);
            lastTime = System.nanoTime();
            i++;
        }

        allResultPerfRecord.end(rsNextUsedTime);
        LOG.info("Finished read record by Sql: [{}\n] {}.",
                querySql, basicMsg);
    ......
}

// 发送到 channel 队列中
protected Record transportOneRecord(RecordSender recordSender, ResultSet rs, 
        ResultSetMetaData metaData, int columnNumber, String mandatoryEncoding, 
        TaskPluginCollector taskPluginCollector) {
    Record record = buildRecord(recordSender,rs,metaData,columnNumber,mandatoryEncoding,taskPluginCollector); 
    recordSender.sendToWriter(record);
    return record;
}

读取到队列，以及出队列 和 队列 channel 设计 源码

// 从reader中读取到数据 源代码
@Override
public void sendToWriter(Record record) {
    if(shutdown){
        throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
    }

    Validate.notNull(record, "record不能为空.");

    if (record.getMemorySize() > this.byteCapacity) {
        this.pluginCollector.collectDirtyRecord(record, new Exception(String.format("单条记录超过大小限制，当前限制为:%s", this.byteCapacity)));
        return;
    }

    boolean isFull = (this.bufferIndex >= this.bufferSize || this.memoryBytes.get() + record.getMemorySize() > this.byteCapacity);
    if (isFull) {
        flush();
    }

    this.buffer.add(record);
    this.bufferIndex++;
    memoryBytes.addAndGet(record.getMemorySize());
}

//  reader 读取时 调用 flush
@Override
public void sendToWriter(Record record) {
    if(shutdown){
        throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
    }

    Validate.notNull(record, "record不能为空.");

    if (record.getMemorySize() > this.byteCapacity) {
        this.pluginCollector.collectDirtyRecord(record, new Exception(String.format("单条记录超过大小限制，当前限制为:%s", this.byteCapacity)));
        return;
    }

    boolean isFull = (this.bufferIndex >= this.bufferSize || this.memoryBytes.get() + record.getMemorySize() > this.byteCapacity);
    if (isFull) {
        flush();
    }

    this.buffer.add(record);
    this.bufferIndex++;
    memoryBytes.addAndGet(record.getMemorySize());
}

// 写数据源代码 ，获取之前reader读取的数据 
@Override
public Record getFromReader() {
    if(shutdown){
        throw DataXException.asDataXException(CommonErrorCode.SHUT_DOWN_TASK, "");
    }
    boolean isEmpty = (this.bufferIndex >= this.buffer.size());
    if (isEmpty) {
        receive();
    }

    Record record = this.buffer.get(this.bufferIndex++);
    if (record instanceof TerminateRecord) {
        record = null;
    }
    return record;
}

// getFromReader 调用 receive 方法
private void receive() {
    this.channel.pullAll(this.buffer);
    this.bufferIndex = 0;
    this.bufferSize = this.buffer.size();
}


// 内存Channel的具体实现，底层其实是一个ArrayBlockingQueue

public class MemoryChannel extends Channel {

    private int bufferSize = 0;

    private AtomicInteger memoryBytes = new AtomicInteger(0);

    private ArrayBlockingQueue<Record> queue = null;
    ......
}

写入过程源代码分析

// MySQL 写入 源代码
public void startWrite(RecordReceiver recordReceiver) {
    this.commonRdbmsWriterTask.startWrite(recordReceiver, this.writerSliceConfig,
            super.getTaskPluginCollector());
}

// rdbms 的 startWrite方法 中的 startWriteWithConnection 子方法源代码如下
public void startWriteWithConnection(RecordReceiver recordReceiver, TaskPluginCollector taskPluginCollector, Connection connection) {
   ......
    List<Record> writeBuffer = new ArrayList<Record>(this.batchSize);
    int bufferBytes = 0;   
    try {
        Record record;
        while ((record = recordReceiver.getFromReader()) != null) {
            if (record.getColumnNumber() != this.columnNumber) {
                // 源头读取字段列数与目的表字段写入列数不相等，直接报错
                throw DataXException
                        .asDataXException(
                                DBUtilErrorCode.CONF_ERROR,
                                String.format(
                                        "列配置信息有错误. 因为您配置的任务中，源头读取字段数:%s 与 目的表要写入的字段数:%s 不相等. 请检查您的配置并作出修改.",
                                        record.getColumnNumber(),
                                        this.columnNumber));
            }

            writeBuffer.add(record);
            bufferBytes += record.getMemorySize();

            if (writeBuffer.size() >= batchSize || bufferBytes >= batchByteSize) {
                doBatchInsert(connection, writeBuffer);
                writeBuffer.clear();
                bufferBytes = 0;
            }
        }
        if (!writeBuffer.isEmpty()) {
            doBatchInsert(connection, writeBuffer);
            writeBuffer.clear();
            bufferBytes = 0;
        }
    } catch (Exception e) {
        throw DataXException.asDataXException(
                DBUtilErrorCode.WRITE_DATA_ERROR, e);
    } finally {
        writeBuffer.clear();
        bufferBytes = 0;
        DBUtil.closeDBResources(null, null, connection);
    }
   ......
}
 // record = recordReceiver.getFromReader() 获取record