private void doDdl(DbLoadContext context, List<EventData> eventDatas) {
        for (final EventData data : eventDatas) {
            DataMedia dataMedia = ConfigHelper.findDataMedia(context.getPipeline(), data.getTableId());
            final DbDialect dbDialect = dbDialectFactory.getDbDialect(context.getIdentity().getPipelineId(),
                (DbMediaSource) dataMedia.getSource());
            Boolean skipDdlException = context.getPipeline().getParameters().getSkipDdlException();
            try {
                Boolean result = dbDialect.getJdbcTemplate().execute(new StatementCallback<Boolean>() {

                    public Boolean doInStatement(Statement stmt) throws SQLException, DataAccessException {
                        Boolean result = true;
                        if (dbDialect instanceof MysqlDialect && StringUtils.isNotEmpty(data.getDdlSchemaName())) {
                            // 如果mysql，执行ddl时，切换到在源库执行的schema上
                            // result &= stmt.execute("use " +
                            // data.getDdlSchemaName());

                            // 解决当数据库名称为关键字如"Order"的时候,会报错,无法同步
                            result &= stmt.execute("use `" + data.getDdlSchemaName() + "`");
                        }
                        result &= stmt.execute(data.getSql());
                        return result;
                    }
                });
                if (result) {
                    context.getProcessedDatas().add(data); // 记录为成功处理的sql
                } else {
                    context.getFailedDatas().add(data);
                }

            } catch (Throwable e) {
                if (skipDdlException) {
                    // do skip
                    logger.warn("skip exception for ddl : {} , caused by {}", data, ExceptionUtils.getFullStackTrace(e));
                } else {
                    throw new LoadException(e);
                }
            }

        }
    }
private void doDdl ( DbLoadContext context , List < EventData > eventDatas ) { for ( final EventData data : eventDatas ) { DataMedia dataMedia = ConfigHelper . findDataMedia ( context . getPipeline ( ) , data . getTableId ( ) ) ; final DbDialect dbDialect = dbDialectFactory . getDbDialect ( context . getIdentity ( ) . getPipelineId ( ) , ( DbMediaSource ) dataMedia . getSource ( ) ) ; Boolean skipDdlException = context . getPipeline ( ) . getParameters ( ) . getSkipDdlException ( ) ; try { Boolean result = dbDialect . getJdbcTemplate ( ) . execute ( new StatementCallback < Boolean > ( ) { public Boolean doInStatement ( Statement stmt ) throws SQLException , DataAccessException { Boolean result = true ; if ( dbDialect instanceof MysqlDialect && StringUtils . isNotEmpty ( data . getDdlSchemaName ( ) ) ) { result &= stmt . execute ( "use `" + data . getDdlSchemaName ( ) + "`" ) ; } result &= stmt . execute ( data . getSql ( ) ) ; return result ; } } ) ; if ( result ) { context . getProcessedDatas ( ) . add ( data ) ; } else { context . getFailedDatas ( ) . add ( data ) ; } } catch ( Throwable e ) { if ( skipDdlException ) { logger . warn ( "skip exception for ddl : {} , caused by {}" , data , ExceptionUtils . getFullStackTrace ( e ) ) ; } else { throw new LoadException ( e ) ; } } } }
执行ddl的调用，处理逻辑比较简单: 串行调用

@param context
@param eventDatas