cloudin-datax是基于DataX开发的分布式数据同步工具，提供简单易用的操作界面，可视化定时任务配置监控和增量同步功能

package com.alibaba.datax.core.job; import com.alibaba.datax.common.constant.PluginType; import com.alibaba.datax.common.exception.DataXException; import com.alibaba.datax.common.plugin.AbstractJobPlugin; import com.alibaba.datax.common.plugin.JobPluginCollector; import com.alibaba.datax.common.spi.Reader; import com.alibaba.datax.common.spi.Writer; import com.alibaba.datax.common.statistics.PerfTrace; import com.alibaba.datax.common.statistics.VMInfo; import com.alibaba.datax.common.util.Configuration; import com.alibaba.datax.common.util.StrUtil; import com.alibaba.datax.core.AbstractContainer; import com.alibaba.datax.core.Engine; import com.alibaba.datax.core.container.util.HookInvoker; import com.alibaba.datax.core.container.util.JobAssignUtil; import com.alibaba.datax.core.job.scheduler.AbstractScheduler; import com.alibaba.datax.core.job.scheduler.processinner.StandAloneScheduler; import com.alibaba.datax.core.statistics.communication.Communication; import com.alibaba.datax.core.statistics.communication.CommunicationTool; import com.alibaba.datax.core.statistics.container.communicator.AbstractContainerCommunicator; import com.alibaba.datax.core.statistics.container.communicator.job.StandAloneJobContainerCommunicator; import com.alibaba.datax.core.statistics.plugin.DefaultJobPluginCollector; import com.alibaba.datax.core.util.ErrorRecordChecker; import com.alibaba.datax.core.util.FrameworkErrorCode; import com.alibaba.datax.core.util.container.ClassLoaderSwapper; import com.alibaba.datax.core.util.container.CoreConstant; import com.alibaba.datax.core.util.container.LoadUtil; import com.alibaba.datax.dataxservice.face.domain.enums.ExecuteMode; import com.alibaba.fastjson.JSON; import com.cyj.datax.Application; import com.cyj.datax.annotation.UserDef; import com.cyj.datax.feign.ArrangeClient; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.Validate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.ByteArrayOutputStream; import java.io.PrintStream; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; /** * Created by jingxing on 14-8-24. * <p/> * job实例运行在jobContainer容器中，它是所有任务的master，负责初始化、拆分、调度、运行、回收、监控和汇报 * 但它并不做实际的数据同步操作 */ public class JobContainer extends AbstractContainer { private static final Logger LOG = LoggerFactory .getLogger(JobContainer.class); private static final SimpleDateFormat dateFormat = new SimpleDateFormat( "yyyy-MM-dd HH:mm:ss"); private ClassLoaderSwapper classLoaderSwapper = ClassLoaderSwapper .newCurrentThreadClassLoaderSwapper(); private long jobId; private String readerPluginName; private String writerPluginName; /** * reader和writer jobContainer的实例 */ private Reader.Job jobReader; private Writer.Job jobWriter; private Configuration userConf; private long startTimeStamp; private long endTimeStamp; private long startTransferTimeStamp; private long endTransferTimeStamp; private int needChannelNumber; private int totalStage = 1; private ErrorRecordChecker errorLimit; public JobContainer(Configuration configuration) { super(configuration); errorLimit = new ErrorRecordChecker(configuration); } /** * jobContainer主要负责的工作全部在start()里面，包括init、prepare、split、scheduler、 * post以及destroy和statistics */ @UserDef @Override public void start() { LOG.info("DataX jobContainer starts job."); boolean hasException = false; boolean isDryRun = false; ArrangeClient arrangeClient = Application.applicationContext.getBean(ArrangeClient.class); try { this.startTimeStamp = System.currentTimeMillis(); isDryRun = configuration.getBool(CoreConstant.DATAX_JOB_SETTING_DRYRUN, false); if(isDryRun) { LOG.info("jobContainer starts to do preCheck ..."); this.preCheck(); } else { userConf = configuration.clone(); LOG.debug("jobContainer starts to do preHandle ..."); this.preHandle(); LOG.debug("jobContainer starts to do init ..."); this.init(); LOG.info("jobContainer starts to do prepare ..."); this.prepare(); LOG.info("jobContainer starts to do split ..."); this.totalStage = this.split(); LOG.info("jobContainer starts to do schedule ..."); this.schedule(); LOG.debug("jobContainer starts to do post ..."); this.post(); LOG.debug("jobContainer starts to do postHandle ..."); this.postHandle(); LOG.info("DataX jobId [{}] completed successfully.", this.jobId); this.invokeHooks(); } arrangeClient.finishTask(true,this.jobId,null); } catch (Throwable e) { LOG.error("Exception when job run", e); hasException = true; if (e instanceof OutOfMemoryError) { this.destroy(); System.gc(); } if (super.getContainerCommunicator() == null) { // 由于 containerCollector 是在 scheduler() 中初始化的，所以当在 scheduler() 之前出现异常时，需要在此处对 containerCollector 进行初始化 AbstractContainerCommunicator tempContainerCollector; // standalone tempContainerCollector = new StandAloneJobContainerCommunicator(configuration); super.setContainerCommunicator(tempContainerCollector); } Communication communication = super.getContainerCommunicator().collect(); // 汇报前的状态，不需要手动进行设置 // communication.setState(State.FAILED); communication.setThrowable(e); communication.setTimestamp(this.endTimeStamp); Communication tempComm = new Communication(); tempComm.setTimestamp(this.startTransferTimeStamp); Communication reportCommunication = CommunicationTool.getReportCommunication(communication, tempComm, this.totalStage); super.getContainerCommunicator().report(reportCommunication); ByteArrayOutputStream ots = new ByteArrayOutputStream(); try{ e.printStackTrace(new PrintStream(ots)); } catch (Exception exception) { } arrangeClient.finishTask(false,this.jobId,ots.toString()); throw DataXException.asDataXException( FrameworkErrorCode.RUNTIME_ERROR, e); } finally { if(!isDryRun) { this.destroy(); this.endTimeStamp = System.currentTimeMillis(); if (!hasException) { //最后打印cpu的平均消耗，GC的统计 VMInfo vmInfo = VMInfo.getVmInfo(); if (vmInfo != null) { vmInfo.getDelta(false); LOG.info(vmInfo.totalString()); } LOG.info(PerfTrace.getInstance().summarizeNoException()); this.logStatistics(); } } } } private void preCheck() { this.preCheckInit(); this.adjustChannelNumber(); if (this.needChannelNumber <= 0) { this.needChannelNumber = 1; } this.preCheckReader(); this.preCheckWriter(); LOG.info("PreCheck通过"); } private void preCheckInit() { this.jobId = this.configuration.getLong(