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