CDP-DataX

SlavePluginCollector

FrameworkErrorCode

OracleReaderErrorCode

MetricManager

FileChannel

StandAloneSlaveContainerCollector

+ collect(): Metric+ report(Metric): void

LocalScheduler

Column

- type: Column.Type- content: Object

+ asLong(): Long+ asDouble(): Double+ asString(): String+ asDate(): Date+ asBytes(): byte[]+ asBool(): Boolean

BoolColumn

SlaveContainerCollector

ConfigParser

+ parse(String path): Configuration

SlaveContainer

- channels: List- readerPool: ThreadPool- writerPool: ThreadPool

+ start(): void //完成数据传输、同步- init(): void //初始化Channel，线程池- schedule(): void //轮询channels，分派任务- destroy(): void //销毁Channel，线程池

NumberColumn

StringColumn

Plugin

+ getName(): String+ getDeveloper(): String+ getDescription(): String+ init(): void+ destroy(): void

MysqlReaderErrorCode

Scheduler

MysqlWriterErrorCode

MasterContainer

- reader: Master- writer: Master- readJobs: List- writeJobs: List

+ start(): void- init(): void- prepare(): void- split(): List- schedule(): void- post(): void- destroy(): void

DataXException

+ code:ErrorCode

+ DataXException(Throwable)+ DataXException(Throwable)+ getErrorCode()

SlaveContainer.WorkUnit

- reader: Reader.Slave- writer: Writer.Slave- channel: Channel

+ start(): void

RecordReceiver

+ receiveFromReader(): Record

Record

Column.Type

ErrorCode

+ getCode:String+ getDescription():String

DoubleQueue

Collector

PluginCollector

ElementErrorCode

PluginLoader

+ loadMaster(String): AbstractMasterPlugin+ loadSlave(String): AbstractSlavePlugin

MemoryChannel

ArrayBlockingQueue

BytesColumn

+ setEncoding(String): void

SlaveContainerRunner

- slaveContainer

+ schedule(): void

AbstractPlugin

- pluginConf: PluginConf

Engine

+ main(String[] args)

Reader.Slave

Writer

OracleWriterrErrorCode

Configuration

Reader

RecordExchanger

Metric

RuntimeException

MasterPluginCollector

Container

StandAloneScheduler

AbstractMasterPlugin

+ prepare(): void+ split(): void+ post(): void

RecordSender

+ sendToWriter(Record): void

AbstractSlavePlugin

+ prepare(): void+ start(): void+ post(): void

BufferedRecordExchanger

+ setBufferSize();

MasterContainerCollector

ContainerCollector

Writer.Slave

Reader.Master

StandAloneSlavePluginCollector

DateColumn

Writer.Master

Channel

+ push(Record): void+ pull(): Record+ pushAll(List): void+ pullAll(List): void+ setSpeedLimit(int): void

实现全部采用 enum

StandAloneMasterContainerCollector