package com.highdatas.mdm.service.impl;
|
|
import com.baomidou.mybatisplus.mapper.EntityWrapper;
|
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
|
import com.highdatas.mdm.entity.*;
|
import com.highdatas.mdm.mapper.SysAssembleMapper;
|
import com.highdatas.mdm.mapper.TableInfoMapper;
|
import com.highdatas.mdm.pojo.*;
|
import com.highdatas.mdm.pojo.kettle.BigDataDataSourceInfo;
|
import com.highdatas.mdm.pojo.kettle.DataSourceInfo;
|
import com.highdatas.mdm.pojo.kettle.UnBigDataDataSourceInfo;
|
import com.highdatas.mdm.service.*;
|
import com.highdatas.mdm.util.Constant;
|
import com.highdatas.mdm.util.ContentBuilder;
|
import com.highdatas.mdm.util.DbUtils;
|
import com.highdatas.mdm.util.RuleClient;
|
import com.xxl.job.core.log.XxlJobLogger;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import org.mybatis.spring.SqlSessionTemplate;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.annotation.Transactional;
|
|
import javax.annotation.Resource;
|
import java.io.ByteArrayInputStream;
|
import java.io.InputStream;
|
import java.io.UnsupportedEncodingException;
|
import java.sql.*;
|
import java.text.MessageFormat;
|
import java.util.*;
|
import java.util.Date;
|
import java.util.stream.Collectors;
|
|
/**
|
* <p>
|
* 服务实现类
|
* </p>
|
*
|
* @author kimi
|
* @since 2020-02-20
|
*/
|
@Slf4j
|
@Service
|
public class SysAssembleServiceImpl extends ServiceImpl<SysAssembleMapper, SysAssemble> implements ISysAssembleService {
|
@Autowired
|
ISysAssembleRuleService ruleService;
|
@Autowired
|
ISysAssembleDbService dbService;
|
@Autowired
|
ISysAssembleApiService apiService;
|
@Autowired
|
ISysAssembleDbTableService tableService;
|
@Autowired
|
ISysAssembleDbFieldService fieldService;
|
@Autowired
|
ISysAssembleParamsService paramsService;
|
@Autowired
|
ISysAssembleLogService logService;
|
@Autowired
|
IMenuMappingService menuMappingService;
|
@Autowired
|
ActivitiService activitiService;
|
@Autowired
|
IMaintainService maintainService;
|
@Autowired
|
IFlowsService flowsService;
|
@Resource
|
SqlSessionTemplate sqlSessionTemplate;
|
@Autowired
|
TableInfoMapper tableInfoMapper;
|
@Autowired
|
UnBigDataDataSourceInfo unBigDataDataSourceInfo;
|
@Autowired
|
BigDataDataSourceInfo bigDataDataSourceInfo;
|
@Autowired
|
MasterDataService masterDataService;
|
|
@Autowired
|
RuleClient ruleClient;
|
@Autowired
|
IMasterModifiedService masterModifiedService;
|
@Autowired
|
ISysViewService viewService;
|
@Autowired
|
ISysMenuService menuService;
|
@Autowired
|
IMasterAuthorSubscribeService subscribeService;
|
/**
|
*
|
* @description: 通过汇集id执行汇集任务
|
* @param id 汇集id
|
* @return: 执行结果
|
*
|
*/
|
@Override
|
public Result run(String id) {
|
XxlJobLogger.log("log info: run" );
|
if (StringUtils.isEmpty(id)) {
|
return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
|
}
|
|
SysAssemble assemble = selectById(id);
|
if (assemble == null) {
|
XxlJobLogger.log("assemble not found" );
|
return Result.error(CodeMsg.ERROR_PARAMS_NOT_MATHED);
|
}
|
|
assemble.setPreTime(new Date());
|
assemble.setPreStatus(SysAssembleRunStatus.working);
|
assemble.updateById();
|
|
if (!assemble.getStatus().equals(SysAssembleStatus.working)) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreCnt(0);
|
XxlJobLogger.log("当前任务不在工作中,状态:" + assemble.getStatus() );
|
assemble.setPreMsg("当前任务不在工作中,状态:" + assemble.getStatus());
|
assemble.updateById();
|
return Result.error(new CodeMsg(6009,"当前任务不在工作中,状态:" + assemble.getStatus()));
|
|
}
|
|
String flowId = assemble.getFlowId();
|
if (!StringUtils.isEmpty(flowId)){
|
Flows flows = flowsService.selectById(flowId);
|
if (flows != null) {
|
ActivitiStatus status = flows.getStatus();
|
if (!status.equals(ActivitiStatus.close) && !status.equals(ActivitiStatus.open)) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreCnt(0);
|
XxlJobLogger.log("当前有流程正在运行,暂时无法汇集下次数据:" + status.toString() );
|
assemble.setPreMsg("当前有流程正在运行,暂时无法汇集下次数据:" + status.toString());
|
assemble.updateById();
|
return Result.error(new CodeMsg(6009,"当前有流程正在运行,暂时无法汇集下次数据:" + status.toString()));
|
|
}
|
}
|
}
|
|
String tableNameByMenu = menuMappingService.getTableNameByMenu(assemble.getMenuId());
|
boolean canAct = maintainService.getCanAct(tableNameByMenu);
|
if (!canAct) {
|
XxlJobLogger.log("当前有流程正在运行,暂时无法汇集下次数据");
|
assemble.setPreMsg("当前有流程正在运行,暂时无法汇集下次数据");
|
assemble.updateById();
|
return Result.error(new CodeMsg(6009,"当前有流程正在运行,暂时无法汇集下次数据"));
|
}
|
|
Boolean bigData = assemble.getBigdata();
|
Date scheduleDate = new Date();
|
XxlJobLogger.log("log info: start assemble");
|
//1 load from db
|
try{
|
Result result = runByDb(assemble);
|
Date dbDate = new Date();
|
XxlJobLogger.log("log info db time:" + (dbDate.getTime() - scheduleDate.getTime()) + "ms");
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.updateById();
|
return result;
|
}
|
|
//2 load from api
|
result = runByApi(id);
|
Date apiDate = new Date();
|
XxlJobLogger.log("log info api time:" + (apiDate.getTime() - dbDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.updateById();
|
return result;
|
}
|
//3check temp table;
|
result = checkTempTable(assemble);
|
Date checkDate = new Date();
|
XxlJobLogger.log("log info check time:" + (checkDate.getTime() - apiDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
XxlJobLogger.log("检测主数据临时表出现错误"+ result.getMessage());
|
|
assemble.setPreMsg("检测主数据临时表出现错误:" + result.getMessage());
|
assemble.updateById();
|
return result;
|
}
|
|
//4 purge data
|
String purgeSql = assemble.getPurgeSql();
|
purgeSql = DbUtils.replaceEscape(purgeSql);
|
XxlJobLogger.log("assemble db purgeSql sql:" + purgeSql);
|
result = purgeData(purgeSql, assemble.getId(), bigData);
|
Date purgeDate = new Date();
|
XxlJobLogger.log("log info purge time:" + (purgeDate.getTime() - checkDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("运行清洗sql出现错误: "+purgeSql + "," + result.getMessage());
|
XxlJobLogger.log("运行清洗sql出现错误: "+purgeSql + "," + result.getMessage() );
|
assemble.updateById();
|
return result;
|
}
|
|
//5 check temp data
|
result = checkTempData(assemble);
|
Date checkTempDate = new Date();
|
XxlJobLogger.log("log info check temp time:" + (checkTempDate.getTime() - purgeDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("质量检验环节出现错误:" + result.getMessage());
|
XxlJobLogger.log("质量检验环节出现错误" + result.getMessage());
|
assemble.updateById();
|
return result;
|
}
|
//6 temp 2 record
|
result = needRollBack(assemble, result, checkTempDate);
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
XxlJobLogger.log("临时数据链接主数据系统错误：" +result.getMessage());
|
assemble.setPreMsg("临时数据链接主数据系统错误：" +result.getMessage());
|
}else {
|
assemble.setPreMsg("汇集成功");
|
XxlJobLogger.log("汇集成功");
|
assemble.setPreStatus(SysAssembleRunStatus.success);
|
}
|
assemble.updateById();
|
return result;
|
}
|
catch (Exception e) {
|
XxlJobLogger.log("汇集任务出现错误："+e.getMessage());
|
assemble.setPreMsg("汇集任务出现错误："+e.getMessage());
|
assemble.setPreStatus(SysAssembleRunStatus.fail).updateById();
|
e.printStackTrace();
|
|
return Result.error(new CodeMsg(6009, e.getMessage()));
|
}
|
}
|
|
@Transactional(rollbackFor = {Exception.class, Error.class})
|
private Result needRollBack(SysAssemble assemble, Result result, Date checkTempDate) {
|
result = temp2record(assemble);
|
Date temp2RecordDate = new Date();
|
XxlJobLogger.log("log info temp2Record time:" + (temp2RecordDate.getTime() - checkTempDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
XxlJobLogger.log("数据搬运至主数据记录环节出现错误;" + result.getMessage() );
|
assemble.setPreMsg("数据搬运至主数据记录环节出现错误:" + result.getMessage());
|
assemble.updateById();
|
return result;
|
}
|
|
|
//7 record 链接 版本
|
|
result = linkMaintain(assemble);
|
Date linkMaintainDate = new Date();
|
XxlJobLogger.log("log info link maintain time:" + (linkMaintainDate.getTime() - temp2RecordDate.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
XxlJobLogger.log("联接版本出现错误:" + result.getMessage());
|
|
assemble.setPreMsg("联接版本出现错误:" + result.getMessage());
|
assemble.updateById();
|
return result;
|
}
|
|
//8 更新 参数
|
result = updateParams(assemble);
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
XxlJobLogger.log("更新参数错误:" + result.getMessage());
|
|
assemble.setPreMsg("更新参数错误:" + result.getMessage());
|
assemble.updateById();
|
return result;
|
}
|
return result;
|
}
|
|
|
private Result linkMaintain(SysAssemble assemble) {
|
try{
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
SysAssembleUpdateType updateType = assemble.getUpdateType();
|
boolean isBigVersion = false;
|
if (updateType.equals(SysAssembleUpdateType.All)) {
|
isBigVersion = true;
|
}
|
Maintain maintain = masterDataService.uploadedData(tableName, updateType, assemble.getUserId(),isBigVersion);
|
XxlJobLogger.log("log info create now maintain:" + maintain.getId());
|
Date before2Record = new Date();
|
|
Result result = temp2recordUpdate(assemble, maintain);
|
Date after2Record = new Date();
|
XxlJobLogger.log("log info temp2Record time:" + (after2Record.getTime() - before2Record.getTime()) + "ms");
|
|
if (!result.getSuccess()) {
|
assemble.setPreStatus(SysAssembleRunStatus.fail);
|
assemble.setPreMsg("搬运更新字段出现错误" );
|
assemble.updateById();
|
return result;
|
}
|
Boolean audit = menuMapping.getAudit();
|
if (audit == null){
|
//默认需要审批
|
audit = true;
|
}
|
// 处理关联人
|
|
|
if (audit) {
|
String chargeId = menuMapping.getChargeId();
|
TUser user = DbUtils.getUserById(chargeId);
|
if (user == null) {
|
return Result.error(new CodeMsg(6009, "找不到对应的负责人:" + chargeId));
|
}
|
activitiService.setUser(user);
|
Flows flows = activitiService.start("process", null, maintain.getId(), ActivitiBusinessType.maintain);
|
assemble.setFlowId(flows.getId()).updateById();
|
maintain.setFlowId(flows.getId());
|
maintain.setDesp("启动汇集流程");
|
maintain.updateById();
|
masterModifiedService.dealAssemble(maintain.getId(), assemble.getUserId(), true);
|
}else {
|
//直接运行
|
Flows flows = new Flows().setStatus(ActivitiStatus.open).setBusinessId(maintain.getId()).setId(DbUtils.getUUID()).setCreateTime(new Date());
|
flows.setBusinessType(ActivitiBusinessType.maintain);
|
flows.insert();
|
maintain.setFlowId(flows.getId());
|
maintain.updateById();
|
maintainService.dealFlow(maintain.getId(), flows.getStatus());
|
log.info("flow-maintainService end");
|
viewService.dealFlow(maintain.getId(), flows.getStatus());
|
log.info("flow-viewService end");
|
subscribeService.dealFlow(maintain.getId(), flows.getStatus());
|
log.info("flow-subscribeService end");
|
menuService.dealFlow(maintain.getId(), flows.getStatus(), flows.getUserId());
|
log.info("flow-menuService end");
|
masterModifiedService.dealAssemble(maintain.getId(), assemble.getUserId(), false);
|
}
|
return Result.success(null);
|
}catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6009, e.getMessage()));
|
}
|
|
}
|
|
private Result updateParams(SysAssemble assemble) {
|
Connection conn = null;
|
try{
|
String id = assemble.getId();
|
Boolean bigdata = assemble.getBigdata();
|
|
DataSourceInfo dataSourceInfo;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
conn = dataSourceInfo.conn();
|
List<SysAssembleParams> paramsList = paramsService.selectList(new EntityWrapper<SysAssembleParams>().eq("parent_id", id));
|
|
for (SysAssembleParams sysAssembleParams : paramsList) {
|
String updateSql = sysAssembleParams.getUpdateSql();
|
if (StringUtils.isEmpty(updateSql)){
|
continue;
|
}
|
|
PreparedStatement preparedStatement = conn.prepareStatement(updateSql, ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_READ_ONLY);
|
ResultSet resultSet = preparedStatement.executeQuery();
|
//默认只有一个结果
|
if (resultSet.first()) {
|
String val = resultSet.getString(1);
|
sysAssembleParams.setVal(val);
|
sysAssembleParams.updateById();
|
}
|
|
}
|
return Result.success(null) ;
|
}
|
catch (Exception e) {
|
|
return Result.error(new CodeMsg(6006, e.getMessage())) ;
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6006, e.getMessage())) ;
|
}
|
}
|
}
|
|
}
|
|
private Result checkTempTable(SysAssemble assemble) {
|
try {
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
Boolean bigdata = assemble.getBigdata();
|
DataSourceInfo dataSourceInfo;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
boolean exists = dataSourceInfo.checkTableExists(tempTableName);
|
if (!exists){
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable == null) {
|
return Result.error(new CodeMsg(6003, "查询正式表的字段:" + tableName));
|
}
|
dataSourceInfo.createTable(tempTableName, fieldsFromTable);
|
}
|
//drop temp data
|
dataSourceInfo.truncateData(tempTableName);
|
return Result.success(null);
|
}catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6004, e.getMessage()));
|
}
|
}
|
private Result temp2recordUpdate(SysAssemble assemble, Maintain maintain) {
|
Connection conn = null;
|
try {
|
String menuId = assemble.getMenuId();
|
Boolean bigdata = assemble.getBigdata();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
String recordTableName = tableName + Constant.RECORD;
|
DataSourceInfo dataSourceInfo;
|
String updateSql = null;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
updateSql = Constant.MYSQLJoinUpdateSql;
|
}
|
conn = dataSourceInfo.conn();
|
SysAssembleUpdateType updateType = assemble.getUpdateType();
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable.contains(Constant.ID)) {
|
fieldsFromTable.remove(Constant.ID);
|
}
|
|
List<String> fieldFromRecordTable = new ArrayList<>(fieldsFromTable);
|
fieldFromRecordTable.add(Constant.ID);
|
fieldFromRecordTable.add(Constant.STD_ID);
|
fieldFromRecordTable.add(Constant.DEAL);
|
|
String recordFieldStr = StringUtils.join(fieldFromRecordTable, Constant.COMMA);
|
int cnt = 0;
|
if(updateType.equals(SysAssembleUpdateType.Increment)) {
|
String updateFields = assemble.getUpdateFields();
|
String[] split = updateFields.split(Constant.SEMICOLON);
|
ArrayList<String> unionCodeFields = new ArrayList<String>(Arrays.asList(split));
|
String joinStr = getJoinFieldParse(unionCodeFields);
|
List<String> otherFields = fieldsFromTable.stream().filter(s -> !unionCodeFields.contains(s)).filter(s -> !s.equalsIgnoreCase(Constant.ID)).collect(Collectors.toList());
|
if (otherFields.size() != 0) {
|
String insertFieldStr = fieldsFromTable.stream()
|
.map(s -> MessageFormat.format(Constant.Alias,Constant.T1,s))
|
.collect(Collectors.joining(Constant.COMMA));
|
|
String updatedSql = MessageFormat.format(updateSql, recordTableName, recordFieldStr,insertFieldStr, tempTableName, tableName, joinStr);
|
PreparedStatement updatedPreparedStatement = conn.prepareStatement(updatedSql);
|
int updateCnt = updatedPreparedStatement.executeUpdate();
|
cnt = updateCnt;
|
tableInfoMapper.insertMatintainDetailFromTemp(DbUtils.quotedStr(maintain.getId()), maintain.getTableName() + Constant.RECORD , DbUtils.quotedStr(Operate.update.toString()));
|
}
|
}
|
|
//tableInfoMapper.updateStdId( maintain.getTableName() + Constant.RECORD);
|
|
tableInfoMapper.tempDeal(maintain.getTableName() + Constant.RECORD, DbUtils.quotedStr(maintain.getId()));
|
|
assemble.setPreCnt(assemble.getPreCnt() + cnt);
|
assemble.updateById();
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
}
|
}
|
}
|
|
private Result temp2record(SysAssemble assemble) {
|
Connection conn = null;
|
try {
|
String menuId = assemble.getMenuId();
|
Boolean bigdata = assemble.getBigdata();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
String recordTableName = tableName + Constant.RECORD;
|
DataSourceInfo dataSourceInfo;
|
String transSql;
|
String updateSql = null;
|
String insertSql = null;
|
if (bigdata) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
transSql = Constant.Temp2RecordHbaseTemplate;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
transSql = Constant.Temp2RecordMySQLTemplate;
|
updateSql = Constant.MYSQLJoinUpdateSql;
|
insertSql = Constant.MYSQLJoinAddSql;
|
}
|
conn = dataSourceInfo.conn();
|
SysAssembleUpdateType updateType = assemble.getUpdateType();
|
List<String> fieldsFromTable = dataSourceInfo.getFieldsFromTable(tableName);
|
if (fieldsFromTable.contains(Constant.ID)) {
|
fieldsFromTable.remove(Constant.ID);
|
}
|
List<String> fieldFromRecordTable = new ArrayList<>(fieldsFromTable);
|
fieldFromRecordTable.add(Constant.ID);
|
fieldFromRecordTable.add(Constant.STD_ID);
|
fieldFromRecordTable.add(Constant.DEAL);
|
|
String recordFieldStr = StringUtils.join(fieldFromRecordTable, Constant.COMMA);
|
String fieldStr = fieldsFromTable.stream()
|
.collect(Collectors.joining(Constant.COMMA));
|
|
int cnt = 0;
|
if (updateType.equals(SysAssembleUpdateType.All)) {
|
String sql = MessageFormat.format(transSql, recordTableName, recordFieldStr, fieldStr, tempTableName);
|
XxlJobLogger.log("update sql:" + sql);
|
PreparedStatement preparedStatement = conn.prepareStatement(sql);
|
cnt = preparedStatement.executeUpdate();
|
|
}else if(updateType.equals(SysAssembleUpdateType.Increment)) {
|
String updateFields = assemble.getUpdateFields();
|
String[] split = updateFields.split(Constant.SEMICOLON);
|
ArrayList<String> unionCodeFields = new ArrayList<String>(Arrays.asList(split));
|
String joinStr = getJoinFieldParse(unionCodeFields);
|
|
// insert
|
String insertFieldStr = fieldsFromTable.stream()
|
.map(s -> MessageFormat.format(Constant.Alias,Constant.T1,s))
|
.collect(Collectors.joining(Constant.COMMA));
|
|
String insetSql = MessageFormat.format(insertSql, recordTableName, recordFieldStr,insertFieldStr, tempTableName, tableName, joinStr);
|
log.info(insertSql);
|
XxlJobLogger.log("update Increment sql:" + insertSql);
|
PreparedStatement preparedStatement = conn.prepareStatement(insetSql);
|
int insertCnt = preparedStatement.executeUpdate();
|
cnt += insertCnt;
|
|
}
|
XxlJobLogger.log("update sql cnt:" + cnt);
|
assemble.setPreCnt(cnt);
|
assemble.updateById();
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6005, e.getMessage()));
|
}
|
}
|
}
|
}
|
|
private String getOthetFieldParse(List<String> unionCodeFields) {
|
ContentBuilder builder = new ContentBuilder(Constant.AND);
|
for (String s : unionCodeFields) {
|
builder.append(MessageFormat.format(Constant.MYSQLJoinParse, s));
|
}
|
return builder.toString();
|
}
|
|
private String getJoinFieldParse(ArrayList<String> updateFields) {
|
|
ContentBuilder builder = new ContentBuilder(Constant.AND);
|
for (String s : updateFields) {
|
builder.append(MessageFormat.format(Constant.MYSQLJoinParse, s));
|
}
|
return builder.toString();
|
}
|
|
private Result checkTempData(SysAssemble assemble) {
|
try {
|
SysAssembleCheckType checkType = assemble.getCheckType();
|
if (checkType == null) {
|
return Result.error(new CodeMsg(6009, "规则校验类型为空"));
|
}
|
if (checkType.equals(SysAssembleCheckType.unlimited)) {
|
return Result.success(null);
|
}
|
|
String menuId = assemble.getMenuId();
|
MenuMapping menuMapping = menuMappingService.selectOne(new EntityWrapper<MenuMapping>().eq("menu_id", menuId));
|
String tableName = menuMapping.getTableName();
|
String tempTableName = Constant.Temp + tableName;
|
HashMap<String, Boolean> fieldResultSet = ruleClient.execuImmeForCollect(tempTableName, assemble.getUserId());
|
switch (checkType) {
|
case successAdd:
|
String unSuccessFields = fieldResultSet.keySet().stream().filter(s -> !fieldResultSet.get(s)).collect(Collectors.joining(Constant.COMMA));
|
if (fieldResultSet.keySet().contains(false)) {
|
return Result.error(new CodeMsg(6009, "规则校验不通过 字段:" + unSuccessFields));
|
}
|
break;
|
case partSuccessAdd:
|
String checkFields = assemble.getCheckFields();
|
String[] split = checkFields.split(Constant.SEMICOLON);
|
for (String s : split) {
|
Boolean checked = fieldResultSet.get(s);
|
if (checked == null || !checked) {
|
return Result.error(new CodeMsg(6009, "规则校验不通过 字段:" + s));
|
}
|
}
|
break;
|
}
|
|
|
return Result.success(null);
|
} catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(3009, e.getMessage()));
|
}
|
}
|
|
|
private Result purgeData(String purgeSql, String assembleId, boolean bigData) {
|
Connection connection = null;
|
Savepoint purge = null;
|
try {
|
DataSourceInfo dataSourceInfo = null;
|
if (bigData) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
connection = dataSourceInfo.conn();
|
connection.setAutoCommit(false);
|
purge = connection.setSavepoint("purge");
|
List<SysAssembleParams> paramsList = paramsService.selectList(new EntityWrapper<SysAssembleParams>().eq("parent_id", assembleId));
|
HashMap<String, String> paramsMap = new HashMap<>();
|
Set<String> matcher = DbUtils.matcher(purgeSql);
|
for (String s : matcher) {
|
List<SysAssembleParams> collect = paramsList.stream().filter(sysAssembleParams -> sysAssembleParams.getCode().equalsIgnoreCase(s)).collect(Collectors.toList());
|
if (collect.size() == 0) {
|
return Result.error(new CodeMsg(6006, "未匹配到参数:" + s));
|
}else if (collect.size() > 1) {
|
return Result.error(new CodeMsg(6006, "匹配到多个参数:" + s));
|
}else {
|
SysAssembleParams sysAssembleParams = collect.get(0);
|
String val = sysAssembleParams.getVal();
|
if (StringUtils.isEmpty(val)) {
|
String initSql = sysAssembleParams.getInitSql();
|
if (StringUtils.isEmpty(initSql)) {
|
return Result.error(new CodeMsg(6006, "初始化参数:" + s + "失败"));
|
}
|
PreparedStatement preparedStatement = connection.prepareStatement(initSql);
|
ResultSet resultSet = preparedStatement.executeQuery();
|
//默认只有一个结果
|
val = resultSet.getString(1);
|
}
|
paramsMap.put(s, val);
|
}
|
}
|
for (String key : paramsMap.keySet()) {
|
String val = paramsMap.get(key);
|
String preParams = MessageFormat.format(Constant.ParamsShell, key);
|
purgeSql = purgeSql.replace(preParams, val);
|
}
|
XxlJobLogger.log("purgeSql:" +purgeSql);
|
List<String> split = DbUtils.split(purgeSql, Constant.SEMICOLON);
|
XxlJobLogger.log("split size:" + split.size());
|
PreparedStatement preparedStatement = null;
|
if (!split.isEmpty()) {
|
for (String oneSql : split) {
|
XxlJobLogger.log("one sql:" + oneSql);
|
if (preparedStatement == null) {
|
preparedStatement = connection.prepareStatement(oneSql);
|
preparedStatement.addBatch(oneSql);
|
} else {
|
preparedStatement.addBatch(oneSql);
|
}
|
}
|
}
|
int[] ints = preparedStatement.executeBatch();
|
for (int i = 0; i < ints.length; i++) {
|
XxlJobLogger.log("sqls count:" + ints[i]);
|
}
|
|
connection.commit();
|
|
return Result.success(null);
|
|
} catch (SQLException e) {
|
e.printStackTrace();
|
if (connection != null) {
|
try {
|
if (purge != null) {
|
connection.rollback(purge);
|
} else {
|
connection.rollback();
|
}
|
|
} catch (SQLException e1) {
|
e1.printStackTrace();
|
return Result.error(new CodeMsg(6004, "回滚sql出错," + e.getMessage()));
|
}
|
}
|
return Result.error(new CodeMsg(6004, "运行清洗sql出错," + e.getMessage()));
|
} finally {
|
if (connection != null) {
|
try {
|
connection.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6003, "本地connection 无法关闭"));
|
}
|
}
|
}
|
}
|
|
private Result runByApi(String id) {
|
//TODO
|
return Result.success(null);
|
}
|
|
private Result runByDb(SysAssemble assemble) {
|
try {
|
String id = assemble.getId();
|
Boolean bigdata = assemble.getBigdata();
|
|
List<SysAssembleDb> dbList = dbService.selectList(new EntityWrapper<SysAssembleDb>().eq(Constant.PARENT_ID, id).eq(Constant.Active, true));
|
for (SysAssembleDb sysAssembleDb : dbList) {
|
String dbId = sysAssembleDb.getId();
|
Result result = loadOneDb(sysAssembleDb, bigdata, id);
|
XxlJobLogger.log("assemble db load one db:" + sysAssembleDb.getDatasourceUrl());
|
if (!result.getSuccess()) {
|
assemble.setPreMsg("导入源数据错误:" + sysAssembleDb.getDatabaseName()).updateById();
|
return result;
|
}
|
}
|
return Result.success(null);
|
} catch (Exception e) {
|
return Result.error(new CodeMsg(3009, e.getMessage()));
|
}
|
}
|
|
private Result loadOneDb(SysAssembleDb sysAssembleDb, boolean bigData, String assembleId) {
|
String dbId = sysAssembleDb.getId();
|
Connection conn = dbService.getConnection(dbId);
|
if (conn == null) {
|
return Result.error(new CodeMsg(6002, MessageFormat.format("未能连接到源 id: {0}", dbId)));
|
}
|
Connection dataSourceConn = null;
|
PreparedStatement statement = null;
|
try {
|
List<SysAssembleDbTable> tableList = tableService.selectList(new EntityWrapper<SysAssembleDbTable>().eq(Constant.PARENT_ID, dbId).eq(Constant.Active, true));
|
for (SysAssembleDbTable dbTable : tableList) {
|
String tableId = dbTable.getId();
|
String tempTableName = dbTable.getTempTableName();
|
XxlJobLogger.log("assemble table" + tempTableName);
|
List<SysAssembleDbField> fieldList = fieldService.selectList(new EntityWrapper<SysAssembleDbField>().eq(Constant.PARENT_ID, tableId));
|
String fields = fieldList.stream().map(SysAssembleDbField::getField).collect(Collectors.joining(Constant.COMMA));
|
XxlJobLogger.log("assemble fields" + fields);
|
SysAssembleTableType type = dbTable.getType();
|
String tableName = null;
|
String assembleTempTableName = tempTableName;
|
if (type.equals(SysAssembleTableType.sql)) {
|
tableName = dbTable.getSql();
|
} else if (type.equals(SysAssembleTableType.table)) {
|
tableName = dbTable.getTableName();
|
}
|
XxlJobLogger.log("assemble source table:" + tableName);
|
//TODO assembleTempTableName 可能会超长 64字节 后续修改
|
DataSourceInfo dataSourceInfo = null;
|
if (!bigData) {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}
|
|
// drop temp data
|
boolean b = dataSourceInfo.truncateData(tempTableName);
|
XxlJobLogger.log("assemble truncateData :" + b);
|
fixAssembleTempTable(dataSourceInfo, assembleTempTableName, fieldList, dbId);
|
XxlJobLogger.log("assemble fixAssembleTempTable :" );
|
//checkAssembleTempTableExists(assembleTempTableName);
|
String filter = dbTable.getFilter();
|
XxlJobLogger.log("assemble raw filter:" + filter);
|
if (!StringUtils.isEmpty(filter)) {
|
Set<String> matcher = DbUtils.matcher(filter);
|
for (String code : matcher) {
|
SysAssembleParams sysAssembleParams = paramsService.selectOne(new EntityWrapper<SysAssembleParams>().eq(Constant.PARENT_ID, assembleId).eq(Constant.Code, code));
|
if (sysAssembleParams == null){
|
return Result.error(new CodeMsg(6009, assembleId + "有变量未匹配到:"+ code));
|
}
|
String val = sysAssembleParams.getVal();
|
if (StringUtils.isEmpty(val)) {
|
return Result.error(new CodeMsg(6009, assembleId + "有变量未获取到值:"+ code));
|
}
|
val = DbUtils.quotedStr(val);
|
filter = filter.replace(DbUtils.assemblParam(code), val);
|
}
|
}else {
|
filter = Constant.WHERE_DEFAULT;
|
}
|
|
XxlJobLogger.log("assemble db filter:" + filter);
|
String runSqlTemplate = null;
|
if (type.equals(SysAssembleTableType.table)){
|
runSqlTemplate = Constant.selectFieldTableTemplate;
|
}else if(type.equals(SysAssembleTableType.sql)){
|
runSqlTemplate = Constant.selectFieldSqlTemplate;
|
}
|
XxlJobLogger.log("assemble db select sql template :" + runSqlTemplate);
|
String sql = MessageFormat.format(runSqlTemplate, fields, tableName, filter);
|
XxlJobLogger.log("assemble db select sql:" + sql);
|
PreparedStatement ps = conn.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
|
ps.setFetchSize(Integer.MIN_VALUE);
|
ps.setFetchDirection(ResultSet.FETCH_REVERSE);
|
ResultSet resultSet = ps.executeQuery();
|
ResultSetMetaData metaData = resultSet.getMetaData();
|
int columnCount = metaData.getColumnCount();
|
|
String insertMySqlSql = assembleSql(unBigDataDataSourceInfo.getDbName(), assembleTempTableName, fieldList);
|
XxlJobLogger.log("assemble db insert sql:" + insertMySqlSql);
|
int cnt = 0;
|
dataSourceConn = dataSourceInfo.conn();
|
dataSourceConn.setAutoCommit(false);
|
statement = dataSourceConn.prepareStatement(insertMySqlSql);
|
int count = 0;
|
while (resultSet.next()) {
|
count++;
|
for (int i = 1; i <= columnCount; i++) {
|
statement.setString(i, resultSet.getString(i));
|
}
|
statement.addBatch();
|
cnt++;
|
if (cnt == 5000) {
|
statement.executeBatch();
|
}
|
}
|
statement.executeBatch();
|
XxlJobLogger.log("assemble db select sql count:" + count);
|
}
|
|
dataSourceConn.commit();
|
return Result.success(null);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
return Result.error(new CodeMsg(6004, e.getMessage()));
|
}finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
if (statement != null) {
|
try {
|
statement.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
if (dataSourceConn != null) {
|
try {
|
dataSourceConn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
}
|
|
private void setOneBatch(boolean bigData, String insertMySqlSql, List<List<String>> branchList) {
|
Connection conn = null;
|
try {
|
DataSourceInfo dataSourceInfo = null;
|
if (bigData) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
|
conn = dataSourceInfo.conn();
|
PreparedStatement statement = conn.prepareStatement(insertMySqlSql);
|
int result = 0;
|
if (branchList.isEmpty()) {
|
return;
|
}
|
for (List<String> stringList : branchList) {
|
for (int i = 0; i < stringList.size(); i++) {
|
statement.setString(i + 1, stringList.get(i));
|
}
|
statement.addBatch();
|
}
|
statement.executeBatch();
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
}
|
|
private void runOneBatch(boolean bigData, String insertMySqlSql, StringBuilder builder) throws UnsupportedEncodingException, SQLException {
|
byte[] bytes = builder.toString().getBytes("UTF-8");
|
if (bytes.length > 0) {
|
InputStream stream = new ByteArrayInputStream(bytes);
|
bulkLoadFromInputStream(insertMySqlSql, stream, bigData);
|
}
|
//清空bulider
|
builder.delete(0, builder.length());
|
}
|
|
private void fixAssembleTempTable(DataSourceInfo dataSourceInfo, String assembleTempTableName, List<SysAssembleDbField> dbFieldList, String dbId) {
|
boolean exists = dataSourceInfo.checkTableExists(assembleTempTableName);
|
List<String> fieldList = dbFieldList.stream().map(SysAssembleDbField::getField).collect(Collectors.toList());
|
if (!exists) {
|
dataSourceInfo.createTable(assembleTempTableName, fieldList);
|
}else {
|
//check 字段是否相同
|
boolean checked = dataSourceInfo.checkFieldList(assembleTempTableName, fieldList);
|
if (!checked){
|
assembleTempTableName = reSetTableName(assembleTempTableName, dbId);
|
dataSourceInfo.createTable(assembleTempTableName, fieldList);
|
|
}
|
}
|
}
|
|
private String reSetTableName(String assembleTempTableName, String tableId) {
|
//TODO Db
|
SysAssembleDbTable dbTable = tableService.selectById(tableId);
|
String suffix = null;
|
suffix = DbUtils.getUUID(16);
|
String tempTableName = null;
|
if (dbTable.getType().equals(SysAssembleTableType.table)) {
|
tempTableName = Constant.AssembleTempTable + dbTable.getTableName() + suffix;
|
}else {
|
tempTableName = Constant.AssembleTempSql+ DbUtils.getUUID(5)+ suffix;
|
}
|
|
dbTable.setTempTableName(tempTableName).updateById();
|
return tempTableName;
|
}
|
|
public void builderEnd(StringBuilder builder, Object object) {
|
builder.append(object);
|
builder.append("\n");
|
}
|
|
|
public void builderAppend(StringBuilder builder, Object object) {
|
builder.append(object);
|
builder.append(",");
|
}
|
|
|
public String assembleSql(String dataBaseName, String tableName, List<SysAssembleDbField> fieldList) {
|
String fields = fieldList.stream().map(SysAssembleDbField::getField).collect(Collectors.joining(Constant.COMMA));
|
ContentBuilder builder = new ContentBuilder(Constant.COMMA);
|
for (int i = 0; i < fieldList.size(); i++) {
|
builder.append(Constant.QUESTION);
|
}
|
String sql = "INSERT INTO " + dataBaseName + "." + tableName + "(" + fields + ") values(" + builder.toString() + ")";
|
return sql;
|
}
|
|
private int bulkLoadFromInputStream(String sql, InputStream dataStream, boolean bigData) {
|
if (null == dataStream) {
|
log.error("输入流为NULL，没有数据导入。");
|
return 0;
|
}
|
Connection conn = null;
|
try {
|
DataSourceInfo dataSourceInfo = null;
|
if (bigData) {
|
dataSourceInfo = bigDataDataSourceInfo;
|
}else {
|
dataSourceInfo = unBigDataDataSourceInfo;
|
}
|
|
conn = dataSourceInfo.conn();
|
PreparedStatement statement = conn.prepareStatement(sql);
|
int result = 0;
|
if (statement.isWrapperFor(java.sql.Statement.class)) {
|
// com.mysql.jdbc.PreparedStatement mysqlStatement = statement.unwrap(com.mysql.jdbc.PreparedStatement.class);
|
// mysqlStatement.setLocalInfileInputStream(dataStream);
|
// statement.
|
// result = mysqlStatement.executeUpdate();
|
}
|
return result;
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
}
|
finally {
|
if (conn != null) {
|
try {
|
conn.close();
|
} catch (SQLException e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
return 0;
|
}
|
}
|
