@SlaveId
public class CangoRequestDto {
    /**
     * The constant serialVersionUID.
     */
    private static final long serialVersionUID = -3153523853123787010L;

    /**
     * The Name.
     *
     * @描述:canal实例名称
     * @字段:name VARCHAR(50)
     */
    @NotEmpty
    private String name;


    /**
     * The Host.
     *
     * @描述:源库主机地址
     * @字段:host VARCHAR(50)
     */
    @NotEmpty
    private String host;


    /**
     * The Port.
     *
     * @描述:源库主机端口
     * @字段:port INT(10)
     */
    @NotNull
    private Integer port;


    /**
     * The User name.
     *
     * @描述:源库用户名
     * @字段:user_name VARCHAR(100)
     */
    @NotEmpty
    private String userName;


    /**
     * The Password.
     *
     * @描述:源库密码
     * @字段:password VARCHAR(32)
     */
    @NotEmpty
    private String password;

    /**
     * @描述:源库名称
     * @字段:db_name VARCHAR(50)
     */
    @NotEmpty
    private String dbName;

    /**
     * 该Canal实例上所有感兴趣的表.
     */
    private Set<String> tableNames;

    /**
     * The Db type.
     *
     */
    @NotNull
    private DbType dbType;


    /**
     * 模拟mysql从库
     *
     */
    @NotNull
    private Integer slaveId;

    /**
     * 需要过滤的表库.
     *
     */
    private String blackTables;

//    /**
//     * binlog解析时间.
//     */
//    private Date logParseTime;
//
//    /**
//     * Gets log parse time.
//     *
//     * @return the log parse time
//     *
//     */
//    public Date getLogParseTime() {
//        return logParseTime;
//    }
//
//    /**
//     * Sets log parse time.
//     *
//     * @param logParseTime the log parse time
//     *
//     */
//    public CangoRequestDto setLogParseTime(Date logParseTime) {
//        this.logParseTime = logParseTime;
//        return this;
//    }

    /**
     * Gets db type.
     *
     * @return the db type
     */
    public DbType getDbType() {
        return dbType;
    }

    /**
     * Sets db type.
     *
     * @param dbType the db type
     * @return the db type
     */
    public CangoRequestDto setDbType(DbType dbType) {
        this.dbType = dbType;
        return this;
    }

    /**
     * Gets host.
     *
     * @return the host
     */
    public String getHost() {
        return host;
    }

    /**
     * Sets host.
     *
     * @param host the host
     * @return the host
     */
    public CangoRequestDto setHost(String host) {
        this.host = host;
        return this;
    }

    /**
     * Gets name.
     *
     * @return the name
     */
    public String getName() {
        return name;
    }

    /**
     * Sets name.
     *
     * @param name the name
     * @return the name
     */
    public CangoRequestDto setName(String name) {
        this.name = name;
        return this;
    }

    /**
     * Gets password.
     *
     * @return the password
     */
    public String getPassword() {
        return password;
    }

    /**
     * Sets password.
     *
     * @param password the password
     * @return the password
     */
    public CangoRequestDto setPassword(String password) {
        this.password = password;
        return this;
    }

    /**
     * Gets port.
     *
     * @return the port
     */
    public Integer getPort() {
        return port;
    }

    /**
     * Sets port.
     *
     * @param port the port
     * @return the port
     */
    public CangoRequestDto setPort(Integer port) {
        this.port = port;
        return this;
    }

    /**
     * Gets slave id.
     *
     * @return the slave id
     */
    public Integer getSlaveId() {
        return slaveId;
    }

    /**
     * Sets slave id.
     *
     * @param slaveId the slave id
     * @return the slave id
     * @date :2016-04-21 11:40:42
     */
    public CangoRequestDto setSlaveId(Integer slaveId) {
        this.slaveId = slaveId;
        return this;
    }

    /**
     * Gets user name.
     *
     * @return the user name
     */
    public String getUserName() {
        return userName;
    }

    /**
     * Sets user name.
     *
     * @param userName the user name
     * @return the user name
     */
    public CangoRequestDto setUserName(String userName) {
        this.userName = userName;
        return this;
    }

    /**
     * Gets db name.
     *
     * @return the db name
     */
    public String getDbName() {
        return dbName;
    }

    /**
     * Sets db name.
     *
     * @param dbName the db name
     * @return the db name
     * @date :2016-05-03 13:51:16
     */
    public CangoRequestDto setDbName(String dbName) {
        this.dbName = dbName;
        return this;
    }

    /**
     * Gets table names.
     *
     * @return the table names
     */
    public Set<String> getTableNames() {
        return tableNames;
    }

    /**
     * Sets table names.
     *
     * @param tableNames the table names
     * @return the table names
     */
    public CangoRequestDto setTableNames(Set<String> tableNames) {
        this.tableNames = tableNames;
        return this;
    }

    /**
     * Gets black tables.
     *
     * @return the black tables
     */
    public String getBlackTables() {
        return blackTables;
    }

    /**
     * Sets black tables.
     *
     * @param blackTables the black tables
     * @return the black tables
     */
    public CangoRequestDto setBlackTables(String blackTables) {
        this.blackTables = blackTables;
        return this;
    }
} public class CangoResponseDto implements Serializable {

    @Override
    public String toString() {
        return "CanalRspDto [ failMsg=" + failMsg + ", msg=" + msg + "]";
    }

    private static final long serialVersionUID = -5348804938984020686L;

    /**
     * 失败信息.
     */
    private String failMsg;

    /**
     * 成功的响应信息.
     */
    private String msg;

    private CangoRspStatus status;

    /**
     * Gets fail msg.
     *
     * @return the fail msg
     * @date :2016-04-25 14:14:31
     */
    public String getFailMsg() {
        return failMsg;
    }

    /**
     * Sets fail msg.
     *
     * @param failMsg the fail msg
     * @date :2016-04-25 14:14:31
     */
    public void setFailMsg(String failMsg) {
        this.failMsg = failMsg;
    }


    /**
     * Gets msg.
     *
     * @return the msg
     * @date :2016-04-27 18:36:03
     */
    public String getMsg() {
        return msg;
    }

    /**
     * Sets msg.
     *
     * @param msg the msg
     * @return the msg
     * @date :2016-04-27 18:36:03
     */
    public CangoResponseDto setMsg(String msg) {
        this.msg = msg;
        return this;
    }

    public CangoRspStatus getStatus() {
        return status;
    }

    public CangoResponseDto setStatus(CangoRspStatus status) {
        this.status = status;
        return this;
    }
} public class CangoInstances {

    /**
     * @描述:主键
     * @字段:id BIGINT(19)
     */
    private Long id;


    /**
     * @描述:cango实例名称
     * @字段:name VARCHAR(50)
     */
    @NotEmpty(message = "实例name不能为空")
    private String name;


    /**
     * @描述:源库主机地址
     * @字段:host VARCHAR(50)
     */
    @NotEmpty(message = "数据库host不能为空")
    private String host;


    /**
     * @描述:源库主机端口
     * @字段:port INT(10)
     */
    @Min(value = 1, message = "数据库port不能为空")
    private Integer port;

    @NotEmpty(message = "数据库名不能为空")
    private String dbName;

    private List<String> tableNames;

    @NotEmpty(message = "用户名不能为空")
    private String userName;

    @NotEmpty(message = "密码不能为空")
    private String password;

    @NotNull(message = "数据库类型不能为空")
    private Integer dbType;

    private Integer slaveId;

    /**
     * 需要过滤的表黑名单.
     */
    private String blackTables;

    private Integer state;

    private Date createTime;

    private Date updateTime;

    public CangoInstances() {
    }

    public CangoInstances(Long id) {
        this.id = id;
    }

    public Long getId() {
        return this.id;
    }

    public CangoInstances setId(Long id) {
        this.id = id;
        return this;
    }

    public String getName() {
        return this.name;
    }

    public CangoInstances setName(String name) {
        this.name = name;
        return this;
    }

    public String getHost() {
        return this.host;
    }

    public CangoInstances setHost(String host) {
        this.host = host;
        return this;
    }

    public Integer getPort() {
        return this.port;
    }

    public CangoInstances setPort(Integer port) {
        this.port = port;
        return this;
    }

    public String getDbName() {
        return this.dbName;
    }

    public CangoInstances setDbName(String dbName) {
        this.dbName = dbName;
        return this;
    }

    public List<String> getTableNames() {
        return tableNames;
    }

    public CangoInstances setTableNames(List<String> tableNames) {
        this.tableNames = tableNames;
        return this;
    }

    public String getUserName() {
        return this.userName;
    }

    public CangoInstances setUserName(String userName) {
        this.userName = userName;
        return this;
    }

    public String getPassword() {
        return this.password;
    }

    public CangoInstances setPassword(String password) {
        this.password = password;
        return this;
    }
    public Integer getDbType() {
        return this.dbType;
    }

    public CangoInstances setDbType(Integer dbType) {
        this.dbType = dbType;
        return this;
    }

    public Integer getSlaveId() {
        return this.slaveId;
    }

    public CangoInstances setSlaveId(Integer slaveId) {
        this.slaveId = slaveId;
        return this;
    }

    public Integer getState() {
        return this.state;
    }

    public CangoInstances setState(Integer state) {
        this.state = state;
        return this;
    }

    public Date getCreateTime() {
        return this.createTime;
    }

    public CangoInstances setCreateTime(Date createTime) {
        this.createTime = createTime;
        return this;
    }

    public Date getUpdateTime() {
        return this.updateTime;
    }

    public CangoInstances setUpdateTime(Date updateTime) {
        this.updateTime = updateTime;
        return this;
    }

    public String getBlackTables() {
        return blackTables;
    }

    public void setBlackTables(String blackTables) {
        this.blackTables = blackTables;
    }
} public class CangoTable {

    private Long id;

    private String instancesName;

    private String tableName;

    private Date createTime;

    private Date updateTime;

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getInstancesName() {
        return instancesName;
    }

    public CangoTable setInstancesName(String instancesName) {
        this.instancesName = instancesName;
        return this;
    }

    public String getTableName() {
        return tableName;
    }

    public CangoTable setTableName(String tableName) {
        this.tableName = tableName;
        return this;
    }

    public Date getCreateTime() {
        return createTime;
    }

    public CangoTable setCreateTime(Date createTime) {
        this.createTime = createTime;
        return this;
    }

    public Date getUpdateTime() {
        return updateTime;
    }

    public CangoTable setUpdateTime(Date updateTime) {
        this.updateTime = updateTime;
        return this;
    }
} public enum CangoRspStatus {
    SUCCESS,
    FAILURE;
} public enum DbType {
    MYSQL(1, "com.mysql.jdbc.Driver"),
    ORACLE(2, "oracle.jdbc.driver.OracleDriver"),
    DRDS(3, "com.mysql.jdbc.Driver");

    private Integer code;
    private String driver;

    DbType(Integer type, String driver) {
        this.code = type;
    }

    public Integer getCode() {
        return code;
    }

    public String getDriver() {
        return driver;
    }

    public boolean isOracle() {
        return this.equals(ORACLE);
    }


    public boolean isMysql() {
        return this.equals(MYSQL);
    }

    public boolean isDRDS() {
        return this.equals(DRDS);
    }

    public DbType parse(Integer code) {
        switch (code) {
            case 1:
                return MYSQL;
            case 2:
                return ORACLE;
            case 3:
                return DRDS;
            default:
                return null;
        }
    }
} public enum OracleParamEnum {
    INSTANCE_NAME,
    DB_USERNAME,
    DB_PASSWORD,
    DB_URL,
    DB_NAME,
    DB_HOST,
    DB_PORT,
    BLACK_TABLES;
} public enum State {

    /**
     * 启动实例.
     */
    START(1, "启动"),

    /**
     * 停用实例
     */
    STOP(2, "停用"),

    /**
     * 禁用实例
     */
    DISABLE(3, "禁用");

    /**
     * The Code.
     */
    private Integer code;

    /**
     * The Desc.
     */
    private String desc;

    /**
     * Instantiates a new Canal rsp status.
     *
     * @param code the code
     * @param desc the desc
     */
    private State(Integer code, String desc) {
        this.code = code;
        this.desc = desc;
    }

    /**
     * Value of canal rsp status.
     *
     * @param code the code
     * @return the canal rsp status
     * @date :2016-04-25 14:13:28
     */
    public static State valueOf(Integer code) {
        if (code != null) {
            State[] values = values();
            for (State value : values) {
                if (code.equals(value.code)) {
                    return value;
                }
            }

        }
        return null;
    }

    /**
     * Gets code.
     *
     * @return the code
     * @date :2016-04-25 14:13:28
     */
    public Integer getCode() {
        return code;
    }

    /**
     * Sets code.
     *
     * @param code the code
     */
    public void setCode(Integer code) {
        this.code = code;
    }

    /**
     * Gets desc.
     *
     * @return the desc
     */
    public String getDesc() {
        return desc;
    }

    /**
     * Sets desc.
     *
     * @param desc the desc
     */
    public void setDesc(String desc) {
        this.desc = desc;
    }

} public class CangoInstanceCache {
    private static final ConcurrentHashMap<String, MysqlInstance> mysqlInstanceMap = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, OracleInstance> oracleInstanceMap = new ConcurrentHashMap<>();

    public static void addOracleInstance(String key, OracleInstance oracleInstance) {
        oracleInstanceMap.putIfAbsent(key, oracleInstance);
    }

    public static void addMysqlInstance(String key, MysqlInstance mysqlInstance) {
        mysqlInstanceMap.putIfAbsent(key, mysqlInstance);
    }

    public static Map<String, MysqlInstance> getMysqlInstanceMap() {
        return Collections.unmodifiableMap(mysqlInstanceMap);
    }

    public static Map<String, OracleInstance> getOracleInstanceMap() {
        return Collections.unmodifiableMap(oracleInstanceMap);
    }


    public static void removeOracleInstance(String key) {
        oracleInstanceMap.remove(key);
    }

    public static void removeMysqlInstance(String key) {
        mysqlInstanceMap.remove(key);
    }

    public static void clearCache() {
        oracleInstanceMap.clear();
        mysqlInstanceMap.clear();
    }

    public static String createKey(String dbHost, int dbPort) {
        return String.format("%s_%d", dbHost, dbPort);
    }

    public static MysqlInstance getMysqlInstance(String key) {
        return mysqlInstanceMap.get(key);
    }

    public static OracleInstance getOracleInstance(String key) {
        return oracleInstanceMap.get(key);
    }
} public class MysqlInstance extends CanalInstanceWithManager {
    private static Logger LOGGER = LoggerFactory.getLogger(MysqlInstance.class);

    private String name;

    private Set<String> fullTableList = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());

    public MysqlInstance(Canal canal) {
        super(canal, null);
        this.name = canal.getName();
    }

    @Override
    public synchronized void start() {
        super.start();
    }

    @Override
    protected void initEventStore() {
        LOGGER.debug("init eventStore");
        eventStore = new KafkaMysqlStore(this.parameters);
    }

    @Override
    protected void initEventSink() {
        super.initEventSink();
    }

    @Override
    public synchronized void stop() {
        super.stop();
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Set<String> getFullTableList() {
        return Collections.unmodifiableSet(fullTableList);
    }

    public void addTables(Collection<String> tables) {
        if (CollectionUtils.isNotEmpty(tables)) {
            fullTableList.addAll(tables);
        }
    }

} public class MysqlParameter extends CanalParameter {
    private String canalName;

    public String getCanalName() {
        return canalName;
    }

    public void setCanalName(String canalName) {
        this.canalName = canalName;
    }
} public class OracleInstance extends AbstractYuGongLifeCycle {
    private static Logger logger = LoggerFactory.getLogger(OracleInstance.class);

    private final Map<OracleParamEnum, Object> params;
    private final RunMode runMode = RunMode.ALL;
    private final DbType sourceDbType = DbType.ORACLE;
    private final int noUpdateThreshold = 3;
    private final int retryTimes = 3;
    private final int retryInterval = 1000;
    private final int threadSize;
    private final AtomicInteger count;  // 对添加table计数，均匀hash到tableHolderMap
    private final Semaphore[] semaphores;
    private final Map<String, Set<TableHolder>> tableHolderMap;
    private YuGongContext globalContext;
    private DataSourceFactory dataSourceFactory = new DataSourceFactory();
    private String name;   // 实例名称
    private volatile boolean extractorDump = true;
    private volatile boolean applierDump = false;
    private YuGongException exception = null;
    private RecordExtractor extractor;
    private RecordApplier applier;
    private String executorName;
    private ThreadPoolExecutor threadPoolExecutor;

    private OracleInstance() {
        this(new HashMap(), null);
        throw new UnsupportedOperationException();
    }

    public OracleInstance(Map params, String name) {
        this.params = params;
        this.name = name;
        this.count = new AtomicInteger(0);
        this.threadSize = 1;
//        this.threadSize = Runtime.getRuntime().availableProcessors();
        this.tableHolderMap = new ConcurrentHashMap<String, Set<TableHolder>>() {
            {
                for (int i = 0; i < threadSize; i++) {
                    put(String.valueOf(i), new ConcurrentSkipListSet<TableHolder>());
                }
            }
        };
        this.semaphores = new Semaphore[threadSize];
        Arrays.fill(semaphores, new Semaphore(0));
    }

    public void start() {
        super.start();

        try {
            if (!dataSourceFactory.isStart()) {
                dataSourceFactory.start();
            }

            globalContext = initGlobalContext();
            extractor = initExtractor(globalContext, runMode);
            applier = initApplier(globalContext);

            if (!extractor.isStart()) {
                extractor.start();
            }

            if (!applier.isStart()) {
                applier.start();
            }

            // 1.将同步的表哈希到tableHolderMap不同的value中
            // 2.开启一个线程池，分别遍历不同key下的tableHolderMap
            executorName = this.getClass().getSimpleName() + "-" + globalContext.getName() + "-" + globalContext.getDbName();
            threadPoolExecutor = new ThreadPoolExecutor(threadSize,
                    threadSize,
                    60,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<Runnable>(threadSize * 2),
                    new NamedThreadFactory(executorName), new ThreadPoolExecutor.CallerRunsPolicy());
            for (int i = 0; i < threadSize; i++) {
                final int temp = i;
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        String key = String.valueOf(temp);
                        label:
                        while (true) {
                            try {
                                Set<TableHolder> tableHolders = tableHolderMap.get(key);
                                if (CollectionUtils.isEmpty(tableHolders)) {
                                    if (OracleInstance.this.isStop()) {
                                        break label;
                                    }
                                    semaphores[temp].acquire();

                                    if (OracleInstance.this.isStop()) {
                                        break label;
                                    }
                                }

                                for (TableHolder tableHolder : tableHolders) {
                                    if (OracleInstance.this.isStop()) {
                                        break label;
                                    }

                                    try {
                                        Table tableMeta = tableHolder.table;
                                        processTable(globalContext, tableMeta, extractor, applier);
                                    } catch (Throwable e) {
                                        if (OracleInstance.this.isStop()) {
                                            logger.info("shutdown OracleInstance thread:[{}]", Thread.currentThread().getName());
                                            break label;
                                        } else {
                                            logger.error("synchronize table[{}] happened exception. caused by {}", tableHolder.table.getName(),
                                                    ExceptionUtils.getFullStackTrace(e));
                                        }
                                    }
                                }

                                Thread.sleep(retryInterval);
                            } catch (InterruptedException e) {
                                logger.info("shutdown OracleInstance thread:[{}], end the while loop", Thread.currentThread().getName());
                                break label;
                            } catch (Throwable e) {
                                logger.error("OracleInstance happened exception. caused by {}", ExceptionUtils.getFullStackTrace(e));
                            }
                        }
                    }
                });
            }
        } catch (Throwable e) {
            logger.error("OracleInstance[{}] start failed caused by {}",
                    name,
                    ExceptionUtils.getFullStackTrace(e));
        }
    }

    /**
     * 暂时的方案是处理每张表到没有数据
     *
     * @param context
     * @param table
     * @param extractor
     * @param applier
     */
    private void processTable(YuGongContext context, Table table, RecordExtractor extractor, RecordApplier applier) {
        try {
            MDC.put(YuGongConstants.MDC_TABLE_SHIT_KEY, table.getFullName());
            ExtractStatus status = ExtractStatus.NORMAL;
//            AtomicLong batchId = new AtomicLong(0);
            long tpsLimit = context.getTpsLimit();
            int noUpdateTimes = 0;
            do {
                long start = System.currentTimeMillis();
                // 提取数据
                Map<String, Object> rowIdsAndRecords = extractor.extract(table);
                String mlogCleanSql = (String) rowIdsAndRecords.get("mlogCleanSql");
                List<ColumnValue> rowsIds = (List<ColumnValue>) rowIdsAndRecords.get("rowIds");
                List<Record> records = (List<Record>) rowIdsAndRecords.get("kafkaIncrementRecords");

                List<Record> ackRecords = records;// 保留ack引用


                // 判断是否记录日志
//                RecordDumper.dumpExtractorInfo(batchId.incrementAndGet(),
//                        ackRecords,
//                        extractorDump);

                if (YuGongUtils.isEmpty(records)) {
                    status = extractor.status();
                } else {
                    // 载入数据
                    Throwable applierException = null;
                    for (int i = 0; i < retryTimes; i++) {
                        try {
                            applier.apply(records, table);
                            applierException = null;
                            break;
                        } catch (Throwable e) {
                            applierException = e;
                            if (processException(context, table, extractor, e)) {
                                break;
                            }
                        }
                    }

                    if (applierException != null) {
                        throw applierException;
                    } else { // 如果extractor,applier整个过程无误，就可以清理MLOG$_{TABLE}中的数据
                        extractor.clearMlog(rowsIds, mlogCleanSql);
                    }
                }

                // 判断是否记录日志
//                RecordDumper.dumpApplierInfo(batchId.get(), ackRecords, records, applierDump);

                long end = System.currentTimeMillis();

                if (tpsLimit > 0) {
                    tpsControl(ackRecords, start, end, tpsLimit);
                }

                Thread.sleep(retryInterval);
                // 控制一下增量的退出
                if (status == ExtractStatus.NO_UPDATE) {
                    noUpdateTimes++;
                    if (noUpdateThreshold > 0 && noUpdateTimes > noUpdateThreshold) {
                        break;
                    }
                }
            } while (status != ExtractStatus.TABLE_END);
            table.getColumns().clear();
            table.getPrimaryKeys().clear();
            logger.info("table[{}] is end by {}", table.getFullName(), status);
            table = null;
        } catch (InterruptedException e) {
            throw new YuGongException(e);
        } catch (Throwable e) {
            throw new YuGongException(e);
        }
    }

    private boolean processException(YuGongContext context, Table table, RecordExtractor extractor, Throwable e) {
        if (ExceptionUtils.getRootCause(e) instanceof InterruptedException) {
            // interrupt事件，响应退出
            logger.info("table[{}] is interrpt ,current status:{} !", table
                    .getFullName(), extractor.status());
            return true;
        } else if (OracleInstance.this.isStop()) {
            return true;
        } else {
            logger.error("retry, something error happened. caused by {}",
                    ExceptionUtils.getFullStackTrace(e));
            logger.info("table[{}] is error , current status:{} !", table
                    .getFullName(), extractor.status());
            try {
                Thread.sleep(retryInterval);
            } catch (InterruptedException e1) {
                exception = new YuGongException(e1);
                Thread.currentThread().interrupt();
                return true;
            }
        }
        return false;
    }


    private RecordApplier initApplier(YuGongContext context) {
        return new KafkaRecordApplier(context);
    }

    private RecordExtractor initExtractor(YuGongContext context, RunMode runMode) {
        if (context.getSourceDbType() == DbType.ORACLE) {
            if (runMode == RunMode.MARK) {
                return new KafkaRecRecordExtractor(context);
            } else if (runMode == RunMode.INC) {
                KafkaIncRecordExtractor recordExtractor = new KafkaIncRecordExtractor(context);
                recordExtractor.setSleepTime(1000L);
                return recordExtractor;
            } else {
                AbstractRecordExtractor incExtractor = (AbstractRecordExtractor) initExtractor(
                        context,
                        RunMode.INC);
                KafkaAllRecordExtractor allExtractor = new KafkaAllRecordExtractor(context);
                allExtractor.setKafkaIncExtractor((AbstractOracleRecordExtractor) incExtractor);
                return allExtractor;
            }
        } else {
            throw new YuGongException("Unsupported " + sourceDbType);
        }
    }

    private YuGongContext initGlobalContext() {
        YuGongContext context = new YuGongContext();
        logger.info("check source database connection ...");
        context.setSourceDs(initDataSource());
        logger.info("check source database is ok");
        context.setSourceEncoding("UTF-8");
        context.setBatchApply(true);
        context.setOnceCrawNum(200);
        context.setTpsLimit(2000);
        context.setIgnoreSchema(true);
        context.setSkipApplierException(false);
        context.setRunMode(runMode);

        // 手动添加的参数
        context.setName((String) params.get(OracleParamEnum.INSTANCE_NAME));
        context.setDbName((String) params.get(OracleParamEnum.DB_NAME));
        context.setDbHost((String) params.get(OracleParamEnum.DB_HOST));
        context.setDbPort((Integer) params.get(OracleParamEnum.DB_PORT));
        return context;
    }

    private DataSource initDataSource() {
        String username = (String) params.get(OracleParamEnum.DB_USERNAME);
        String password = (String) params.get(OracleParamEnum.DB_PASSWORD);
        String url = (String) params.get(OracleParamEnum.DB_URL);
        DbType dbType = DbType.ORACLE;
        String poolSize = "30";

        Properties properties = new Properties();
        if (poolSize != null) {
            properties.setProperty("maxActive", poolSize);
        } else {
            properties.setProperty("maxActive", "200");
        }

        DataSourceConfig dsConfig = new DataSourceConfig(url, username, password, dbType, properties);
        return dataSourceFactory.getDataSource(dsConfig);
    }

    private List<TableHolder> initTable(String tableName) {
        logger.info("check source tables read privileges ...");
        String[] strArr = StringUtils.split(tableName, ".");
        List<Table> whiteTables = null;
        List<String> blackTables = (List<String>) params.get(OracleParamEnum.BLACK_TABLES);
        boolean ignoreSchema = false;

        if (strArr.length == 2 && !isBlackTable(tableName, blackTables)) {
            if (tableName.indexOf("*") == -1) {  // 某个表
                whiteTables = TableMetaGenerator.getTableMetasWithoutColumn(globalContext.getSourceDs(),
                        strArr[0],
                        strArr[1]);
            } else {                             // 某个库
                whiteTables = TableMetaGenerator.getTableMetasWithoutColumn(globalContext.getSourceDs(),
                        strArr[0],
                        null);
            }

        } else {
            throw new YuGongException("table[" + tableName + "] is not valid");
        }

        if (whiteTables.isEmpty()) {
            throw new YuGongException("table[" + tableName + "] is not found");
        }

        List<TableHolder> tableHolders = Lists.newArrayList();
        for (Table table : whiteTables) {
            // 根据实际表名处理一下
            if (!isBlackTable(table.getName(), blackTables)) {
                TableMetaGenerator.buildColumns(globalContext.getSourceDs(), table);
                TableHolder holder = new TableHolder(table);
                holder.ignoreSchema = ignoreSchema;
                tableHolders.add(holder);
            }
        }

        logger.info("check source tables is ok.");

        return tableHolders;
    }

    private boolean isBlackTable(String table, List tableBlackList) {
        for (Object tableBlack : tableBlackList) {
            if (LikeUtil.isMatch((String) tableBlack, table)) {
                return true;
            }
        }

        return false;
    }

    private boolean addTableName(String tableName) {
        if (checkTableName(tableName)) {
            int hash = count.getAndIncrement();
            hash %= threadSize;
            Set<TableHolder> tableHolders = tableHolderMap.get(String.valueOf(hash));
            boolean result = tableHolders.addAll(initTable(tableName));

            if (!result) { // 如果添加失败，count就减一还原
                count.decrementAndGet();
            }

            if (result && semaphores[hash].availablePermits() == 0) {
                semaphores[hash].release();
            }
            return result;
        } else {
            return false;
        }
    }

    /**
     * 除非全部的tableNames均为空，返回false，否则为true
     *
     * @param tableNames
     * @return 返回新添加的表
     */
    public synchronized List<String> addTableName(List<String> tableNames) {
        if (CollectionUtils.isEmpty(tableNames)) {
            return null;
        } else {
            List<String> addedTables = new ArrayList<>();
            for (int i = 0, len = tableNames.size(); i < len; i++) {
                String table = tableNames.get(i);
                if (addTableName(table)) {
                    addedTables.add(table);
                }
            }
            return addedTables;
        }
    }

    private boolean checkTableName(String tableName) {
        if (StringUtils.isEmpty(tableName)) {
            logger.warn("tableName should not be null!");
            return false;
        }

        if (!tableName.contains(".")) {
            logger.warn("schema should not be null, please set schema for table[" + tableName + "]");
            return false;
        }

        for (int i = 0; i < threadSize; i++) {
            for (TableHolder tableHolder : tableHolderMap.get(String.valueOf(i))) {
                // 忽略区分大小写
                if (tableHolder.table.getName().equalsIgnoreCase(tableName)) {
                    return false;
                }
            }
        }

        return true;
    }


    public void stop() {
        super.stop();
        if (!extractor.isStop()) {
            extractor.stop();
        }

        if (!applier.isStop()) {
            applier.stop();
        }

        threadPoolExecutor.shutdownNow();

        releaseSemaphores(semaphores);

        while (!threadPoolExecutor.isTerminated()) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                // ignore
            }
        }

        if (dataSourceFactory.isStart()) {
            dataSourceFactory.stop();
        }

        exception = null;

        tableHolderMap.clear();
        logger.info("OracleInstance [{}] stop successful. ", name);
    }

    private void releaseSemaphores(Semaphore[] semaphores) {
        for (Semaphore semaphore : semaphores) {
            semaphore.release();
        }
    }

    private void tpsControl(List<Record> result, long start, long end, long tps) throws InterruptedException {
        long expectTime = (result.size() * 1000) / tps;
        long runTime = expectTime - (end - start);
        if (runTime > 0) {
            Thread.sleep(runTime);
        }
    }

    public String getName() {
        return name;
    }

    private static class TableHolder {
        Table table;
        boolean ignoreSchema = false;

        public TableHolder(Table table) {
            this.table = table;
        }
    }
} public class YuGongException extends NestableRuntimeException {

    private static final long serialVersionUID = -654893533794556357L;

    public YuGongException(String errorCode) {
        super(errorCode);
    }

    public YuGongException(String errorCode, Throwable cause) {
        super(errorCode, cause);
    }

    public YuGongException(String errorCode, String errorDesc) {
        super(errorCode + ":" + errorDesc);
    }

    public YuGongException(String errorCode, String errorDesc, Throwable cause) {
        super(errorCode + ":" + errorDesc, cause);
    }

    public YuGongException(Throwable cause) {
        super(cause);
    }


} public interface CangoInstancesService {
    void save(CangoInstances cangoInstances);

    void updateState(CangoInstances cangoInstances);

    void delete(CangoInstances cangoInstances);

    CangoInstances findByName(CangoInstances cangoInstances);

    void deleteAll();

    /**
     * Find by condition list, including HOST, PORT, STATE, DBTYPE.
     *
     * @param cangoInstances the cango instances
     * @return the list
     */
    List<CangoInstances> findByCondition(CangoInstances cangoInstances);

    void batchUpdateState(List<CangoInstances> instancesList);

    void saveOrUpdate(CangoInstances cangoInstances);
} public interface CangoManagerService {

    CangoResponseDto start(CangoRequestDto cangoRequestDto);

    CangoResponseDto add(CangoRequestDto cangoRequestDto);

    CangoResponseDto stop(CangoRequestDto cangoRequestDto);

    CangoResponseDto enable(CangoRequestDto cangoRequestDto);

    CangoResponseDto disable(CangoRequestDto cangoRequestDto);

    CangoResponseDto check(CangoRequestDto canalReqDto);

    CangoResponseDto startAll();

    CangoResponseDto stopAll();

    CangoResponseDto shutdown();

    CangoResponseDto clearAll();
} public interface CangoTableService {
    void save(CangoTable cangoTable);

    CangoTable query(CangoTable cangoTable);

    void deleteByInstancesName(CangoTable cangoTable);

    List<CangoTable> findByInstancesName(CangoTable cangoTable);

    void deleteAll();

    void batchSave(List<CangoTable> cangoTables);
}
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.instance.manager.model.Canal;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
import com.bella.cango.dto.CangoRequestDto;
import com.bella.cango.dto.CangoResponseDto;
import com.bella.cango.entity.CangoInstances;
import com.bella.cango.entity.CangoTable;
import com.bella.cango.enums.CangoRspStatus;
import com.bella.cango.enums.DbType;
import com.bella.cango.enums.OracleParamEnum;
import com.bella.cango.enums.State;
import com.bella.cango.instance.cache.CangoInstanceCache;
import com.bella.cango.instance.mysql.MysqlInstance;
import com.bella.cango.instance.mysql.MysqlParameter;
import com.bella.cango.instance.oracle.OracleInstance;
import com.bella.cango.instance.oracle.exception.YuGongException;
import com.bella.cango.service.CangoInstancesService;
import com.bella.cango.service.CangoManagerService;
import com.bella.cango.service.CangoTableService;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.net.InetSocketAddress;
import java.text.MessageFormat;
import java.util.*;
            return tableNames;
        }
        return null;
    }

    /**
     * Filter table list set.
     * note: 必须满足schema.table的格式
     *
     * @param tableNames the table names
     * @return the set
     */
    private Set<String> filterTableList(List<String> tableNames) {
        if (!CollectionUtils.isEmpty(tableNames)) {
            Set<String> filteredTables = new HashSet<>();
            for (String tableName : tableNames) {
                if (StringUtils.isNotBlank(tableName)
                        && tableName.contains(".")
                        && !tableName.startsWith(".")
                        && !tableName.endsWith(".")
                        && tableName.indexOf(".") == tableName.lastIndexOf(".")) {
                    filteredTables.add(tableName);
                }
            }

            return filteredTables;
        } else {
            return null;
        }
    }

    /**
     * Build canal parameter canal parameter.
     *
     * @param cangoInstances the canal instances
     * @return the canal parameter
     * @date :2016-04-26 17:20:18
     */
    private CanalParameter buildCanalParameter(CangoInstances cangoInstances) {
        MysqlParameter parameter = new MysqlParameter();
        parameter.setCanalName(cangoInstances.getName());
        parameter.setMetaMode(CanalParameter.MetaMode.ZOOKEEPER);
        parameter.setHaMode(CanalParameter.HAMode.HEARTBEAT);
        parameter.setIndexMode(CanalParameter.IndexMode.ZOOKEEPER);

        parameter.setMemoryStorageBufferSize(32 * 1024);

        parameter.setSourcingType(CanalParameter.SourcingType.MYSQL);
        parameter.setDbAddresses(Collections.singletonList(new InetSocketAddress(cangoInstances.getHost(), cangoInstances.getPort() != null ? cangoInstances.getPort() : 3306)));
        parameter.setDbUsername(cangoInstances.getUserName());
        parameter.setDbPassword(cangoInstances.getPassword());

        parameter.setSlaveId(Long.valueOf(cangoInstances.getSlaveId()));

        //zk设置
        parameter.setZkClusters(Arrays.asList(zkCluster.split(",")));

        parameter.setDefaultConnectionTimeoutInSeconds(30);
        parameter.setConnectionCharset("UTF-8");
        parameter.setConnectionCharsetNumber((byte) 33);
        parameter.setReceiveBufferSize(8 * 1024);
        parameter.setSendBufferSize(8 * 1024);

        //过滤掉库
//        parameter.setBlackFilter("xx\\..*");
        parameter.setBlackFilter(blackFilterConverter(cangoInstances.getBlackTables()));

        parameter.setDetectingEnable(false);
        parameter.setFilterTableError(true);
        parameter.setDetectingIntervalInSeconds(10);
        parameter.setDetectingRetryTimes(3);
        parameter.setDetectingSQL("SELECT 'x' FROM DUAL");

        return parameter;
    }

    /**
     * Black filter converter string.
     *
     * @param blackTables the black tables
     * @return the string
     */
    private String blackFilterConverter(String blackTables) {
        if (StringUtils.isNotBlank(blackTables)) {
            List<String> blackList = Arrays.asList(blackTables.split(","));
            Set<String> filterTableList = filterTableList(blackList);
            StringBuilder stringBuilder = new StringBuilder();
            for (String black : filterTableList) {
                if (black.contains("*")) {
                    String schema = black.substring(0, black.indexOf("."));
                    stringBuilder.append(schema).append("\\..*").append(",");
                } else {
                    String[] split = black.split("\\.");
                    stringBuilder.append(split[0]).append("\\.").append(split[1]).append(",");
                }
            }

            return stringBuilder.deleteCharAt(stringBuilder.length() - 1).toString();
        } else {
            return "";
        }
    }

    /**
     * Start yu gong.
     *
     * @param cangoInstances the canal instances
     * @date : 2016-07-05 14:05:30
     */
    private void startYuGong(CangoInstances cangoInstances) {
        //查看该数据库实例是否已启用
        String key = cangoInstances.getName();

        //该数据库实例是否已经添加过（相同的数据库主机、端口只需启动一个CanalInstance）
        OracleInstance cachedInstance = CangoInstanceCache.getOracleInstance(key);
        if (cachedInstance == null) {
            try {
                String instancesInfo = cangoInstances.getHost() + ":" + cangoInstances.getPort() + "/" + cangoInstances.getDbName();
                LOGGER.info("start the OracleInstance ({}).", instancesInfo);
                if (StringUtils.isEmpty(cangoInstances.getDbName())) {
                    throw new YuGongException("schema should not be null!");