DataLink是一个满足各种异构数据源之间的实时增量同步分布式可扩展的数据交换平台

package com.ucar.datalink.biz.service.impl; import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; import com.google.common.collect.Lists; import com.google.common.collect.Sets; import com.ucar.datalink.biz.dal.MediaDAO; import com.ucar.datalink.biz.dal.MediaSourceDAO; import com.ucar.datalink.biz.service.MediaSourceService; import com.ucar.datalink.biz.service.SyncRelationService; import com.ucar.datalink.biz.utils.DataLinkFactory; import com.ucar.datalink.biz.utils.DataSourceFactory; import com.ucar.datalink.biz.utils.ddl.DdlUtils; import com.ucar.datalink.biz.utils.ddl.DdlSqlUtils; import com.ucar.datalink.biz.utils.ddl.SQLStatementHolder; import com.ucar.datalink.common.errors.DatalinkException; import com.ucar.datalink.domain.media.*; import com.ucar.datalink.domain.media.parameter.MediaSrcParameter; import com.ucar.datalink.domain.media.parameter.rdb.RdbMediaSrcParameter; import com.ucar.datalink.domain.media.parameter.sddl.SddlMediaSrcParameter; import com.ucar.datalink.domain.relationship.*; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang.StringUtils; import org.apache.ddlutils.model.Table; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; import javax.sql.DataSource; import java.util.*; import java.util.stream.Collectors; /** * Created by lubiao on 2017/5/23. */ @Service public class SyncRelationServiceImpl implements SyncRelationService { private static final String ALL_MAPPING = "ALL_MAPPING"; private LoadingCache<String, List<MediaMappingInfo>> mediaMappingCache = CacheBuilder.newBuilder().build( new CacheLoader<String, List<MediaMappingInfo>>() { @Override public List<MediaMappingInfo> load(String key) throws Exception { List<MediaMappingInfo> list = mediaDAO.getAllMediaMappings(); return list == null ? Lists.newArrayList() : list; } } ); private LoadingCache<Long, List<MediaMappingInfo>> taskMediaMappingCache = CacheBuilder.newBuilder().build( new CacheLoader<Long, List<MediaMappingInfo>>() { @Override public List<MediaMappingInfo> load(Long taskId) throws Exception { List<MediaMappingInfo> allList = mediaMappingCache.getUnchecked(ALL_MAPPING); return allList.stream().filter(i -> i.getTaskId().equals(taskId)).collect(Collectors.toList()); } } ); private LoadingCache<String, List<SyncNode>> treesCache = CacheBuilder.newBuilder().build( new CacheLoader<String, List<SyncNode>>() { @Override public List<SyncNode> load(String mediaName) throws Exception { return generateTrees(mediaName); } } ); @Override public void clearSyncRelationCache() { mediaMappingCache.invalidateAll(); taskMediaMappingCache.invalidateAll(); treesCache.invalidateAll(); } @Autowired MediaDAO mediaDAO; @Autowired MediaSourceDAO mediaSourceDAO; @Override public List<SyncNode> getSyncRelationTrees(Long mediaSourceId, String mediaName) { Long sddlMediaSourceId = isSDDLSubDB(mediaSourceId); Long realMediaSourceId; if (sddlMediaSourceId != null) { realMediaSourceId = sddlMediaSourceId; } else { realMediaSourceId = mediaSourceId; } List<SyncNode> list = treesCache.getUnchecked(mediaName); return list.stream().filter(i -> isDbInNodeTree(realMediaSourceId, mediaName, i, true)).collect(Collectors.toList()); } @Override public List<SqlCheckResult> checkSqls(Long mediaSourceId, String sqls) { Long sddlMediaSourceId = isSDDLSubDB(mediaSourceId); if (sddlMediaSourceId != null) { mediaSourceId = sddlMediaSourceId; } MediaSourceInfo mediaSourceInfo = mediaSourceDAO.getById(mediaSourceId); List<SQLStatementHolder> holders = DdlSqlUtils.buildSQLStatement(mediaSourceInfo.getType(), sqls); List<SqlCheckResult> list = new ArrayList<>(); if (!holders.isEmpty()) { for (SQLStatementHolder holder : holders) { SqlCheckResult sqlCheckResult = checkOneSql(mediaSourceInfo, holder); if (sqlCheckResult != null) { list.add(sqlCheckResult); } } } return list; } private SqlCheckResult checkOneSql(final MediaSourceInfo mediaSourceInfo, final SQLStatementHolder holder) { SqlCheckResult sqlCheckResult = new SqlCheckResult(holder.getSqlString()); holder.check(); if (holder.getSqlType().equals(SqlType.CreateTable)) { holder.getSqlCheckItems().forEach(i -> { List<SqlCheckTree> trees = buildSqlCheckTrees(i, mediaSourceInfo); if (!trees.isEmpty()) { sqlCheckResult.getSqlCheckTrees().addAll(trees); trees.forEach(t -> { t.setSqlExeDirection(SqlExeDirection.Negative); t.getSqlCheckNotes().add(new SqlCheckNote( "请按照同步关系的反向顺序执行脚本.", SqlCheckNote.RoleType.DBA, SqlCheckNote.NoteLevel.INFO)); if (containsSddlDataSource(t.getRootNode())) { t.getSqlCheckNotes().add(new SqlCheckNote( "SDDL数据源内部先执行非0号库，再执行0号库.", SqlCheckNote.RoleType.DBA, SqlCheckNote.NoteLevel.INFO)); } if (containsSddlDataSource(t.getRootNode())) { t.getSqlCheckNotes().add(new SqlCheckNote( "SDDL增加数据表，请在sddladmin中做对应的配置变更.", SqlCheckNote.RoleType.DLA, SqlCheckNote.NoteLevel.WARN)); } t.getSqlCheckNotes().addAll(buildSqlCheckNotes(mediaSourceInfo, i, t.getRootNode(), true)); }); } } ); } else if (holder.getSqlType().equals(SqlType.AlterTable)) { holder.getSqlCheckItems().forEach(i -> { List<SqlCheckTree> trees = buildSqlCheckTrees(i, mediaSourceInfo); if (!trees.isEmpty()) { sqlCheckResult.getSqlCheckTrees().addAll(trees); trees.forEach(t -> { if (i.isContainsTableRename()) { t.getSqlCheckNotes().add(new SqlCheckNote( "参与数据同步的表，不支持[Table-Rename]操作.", SqlCheckNote.RoleType.ALL, SqlCheckNote.NoteLevel.ERROR)); } if (i.isContainsColumnRename()) { t.getSqlCheckNotes().add(new SqlCheckNote( "参与数据同步的表，不支持[Column-Rename]操作", SqlCheckNote.RoleType.ALL, SqlCheckNo