package com.product.data.center.service;
|
|
import cn.hutool.core.util.IdUtil;
|
import com.product.admin.service.PublicService;
|
import com.product.common.lang.StringUtils;
|
import com.product.core.config.CoreConst;
|
import com.product.core.entity.DataTableEntity;
|
import com.product.core.entity.FieldSetEntity;
|
import com.product.core.exception.BaseException;
|
import com.product.core.service.support.AbstractBaseService;
|
import com.product.core.spring.context.SpringMVCContextHolder;
|
import com.product.core.transfer.Transactional;
|
import com.product.data.center.config.CmnConst;
|
import com.product.data.center.config.ErrorCode;
|
import com.product.data.center.service.ide.ICommonService;
|
import com.product.data.center.utils.CallBackReturnValue;
|
import com.product.quartz.service.impl.SysJobService;
|
import com.product.util.BaseUtil;
|
import org.springframework.beans.factory.annotation.Value;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.text.DecimalFormat;
|
import java.util.Date;
|
import java.util.regex.Matcher;
|
import java.util.regex.Pattern;
|
|
/**
|
* @Author cheng
|
* @Date 2022/7/22 9:43
|
* @Desc
|
*/
|
@Service("dataCommonService")
|
public class CommonService extends AbstractBaseService implements ICommonService {
|
|
@Resource
|
SysJobService sysJobService;
|
|
@Resource
|
PublicService publicService;
|
|
@Resource
|
SyncDelRecordService syncDelRecordService;
|
|
@Resource
|
MesExternalService mesExternalService;
|
|
@Value("${data.system.name}")
|
private String dataSystemName;
|
|
public void deleteCenterLog() {
|
this.getBaseDao().executeUpdate("DELETE FROM product_sys_data_center_log WHERE uuid in (select * from product_sys_data_center_log_del_v) limit 10000");
|
}
|
|
|
/**
|
* 创建定时任务
|
*
|
* @param taskName 任务名称
|
* @param isUsed 是否启用
|
* @param corn 执行时间
|
* @param executeExpression 执行表达式（bean.method)
|
*/
|
private String createTimeTask(String timeTaskUuid, String taskName, boolean isUsed, String corn, String executeExpression) {
|
FieldSetEntity fse = new FieldSetEntity();
|
try {
|
fse.setTableName("product_sys_timed_task");
|
fse.setValue("uuid", timeTaskUuid);
|
fse.setValue("job_name", taskName);//任务名称
|
fse.setValue("job_group", "system");//分组
|
fse.setValue("invoke_target", executeExpression);//调用目标字符串
|
fse.setValue("cron_expression", corn);//cron表达式
|
fse.setValue("misfire_policy", "3");//错误执行策略 只执行一次
|
fse.setValue("concurrent", 0);//不允许并发执行
|
fse.setValue("remark", taskName);
|
fse.setValue("created_by", SpringMVCContextHolder.getCurrentUser().getUser_id());
|
fse.setValue("created_utc_datetime", new Date());
|
fse.setValue("status", isUsed ? 1 : 0);
|
fse.setValue("is_conceal", 1);
|
|
if (!StringUtils.isEmpty(timeTaskUuid)) {
|
BaseUtil.createCreatorAndCreationTime(fse);
|
sysJobService.updateJob(fse);
|
} else {
|
sysJobService.insertJob(fse);
|
}
|
return fse.getUUID();
|
} catch (Exception e) {
|
e.printStackTrace();
|
throw new BaseException(ErrorCode.CRATED_TIMED_TASK_FAIL);
|
}
|
}
|
|
/**
|
* 通用保存
|
*
|
* @param fse
|
* @param t
|
*/
|
private void commonSave(FieldSetEntity fse, CallBackReturnValue<String, FieldSetEntity> t) {
|
boolean isAdd = false;
|
BaseUtil.createCreatorAndCreationTime(fse);
|
if (StringUtils.isEmpty(fse.getUUID())) {
|
fse.setValue(CmnConst.UUID, IdUtil.randomUUID());
|
isAdd = true;
|
} else if (!StringUtils.isEmpty(fse.getUUID()) && "add".equals(fse.getString(CoreConst.SYSTEM_DATA_OPERATE_TYPE))) {
|
isAdd = true;
|
}
|
String timeTasksUid = t.invokeMethod(fse);
|
fse.setValue(CmnConst.TIME_TASK_UUID, timeTasksUid);
|
if (isAdd) {
|
getBaseDao().add(fse);
|
} else {
|
getBaseDao().update(fse);
|
}
|
}
|
|
|
/**
|
* 保存MES数据同步配置
|
*
|
* @param fse
|
* @return
|
* @throws BaseException
|
*/
|
@Transactional
|
@Override
|
public String saveMesSyncDataConfig(FieldSetEntity fse) throws BaseException {
|
DataTableEntity subDt = fse.getSubDataTable("product_sys_data_sync_mes_sub");
|
//固定uuid
|
fse.setValue("uuid", 1);
|
DataTableEntity dt = getBaseDao().listTable(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG, "del_type=1");
|
DataTableEntity subDataTable = null;
|
if (!DataTableEntity.isEmpty(subDt)) {
|
subDataTable = subDt.clones();
|
int b = 0;
|
for (int i = 0; i < subDt.getRows(); i++) {
|
FieldSetEntity subFse = subDt.getFieldSetEntity(i);
|
String dataType = subFse.getString("~type~");
|
String uuid = subFse.getString(CmnConst.UUID);
|
if ("del".equals(dataType)) {
|
String timeTaskUid = subFse.getString(CmnConst.TIME_TASK_UUID);
|
FieldSetEntity task = getBaseDao().getFieldSetEntity("product_sys_timed_task", timeTaskUid, false);
|
if (task != null) {
|
cancelTimeTask(task);
|
}
|
subDataTable.setFieldValue(b, "config_uuid", subFse.getUUID());
|
} else {
|
subFse.setValue("master_uuid", fse.getUUID());
|
commonSave(subFse, fs -> createTimeTask(fs.getString(CmnConst.TIME_TASK_UUID), "SYNC:" + fs.getString("table_name"), fs.getBoolean(CmnConst.IS_USED), fs.getString("execution_time"), "dataSyncService.runTimeTask('" + fs.getUUID() + "')"));
|
subDt.removeFieldSetEntity(i);
|
subDataTable.setFieldValue(b, "uuid", subFse.getUUID());
|
i--;
|
}
|
b++;
|
}
|
if (!DataTableEntity.isEmpty(subDt)) {
|
Object[] objects = subDt.getData().stream().map(item -> item.getUUID()).toArray();
|
getBaseDao().delete("product_sys_data_sync_mes_sub", objects);
|
}
|
}
|
fse.removeSubData("product_sys_data_sync_mes_sub");
|
getBaseDao().saveFieldSetEntity(fse);
|
//创建触发器
|
// //1 主库删除后同步删除子库
|
syncDelRecordService.createTrigger(subDataTable, fse.getString("data_source"), fse.getString("target_data_source"));
|
return fse.getUUID();
|
}
|
|
/**
|
* 取消定时任务
|
*
|
* @param fse
|
* @throws BaseException
|
*/
|
private void cancelTimeTask(FieldSetEntity fse) throws BaseException {
|
try {
|
this.sysJobService.deleteJob(fse);
|
} catch (Exception e) {
|
e.printStackTrace();
|
SpringMVCContextHolder.getSystemLogger().error(e);
|
throw new BaseException(ErrorCode.CANCEL_TIME_TASK_FAIL);
|
}
|
}
|
|
/**
|
* 采集配置保存
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
@Transactional
|
public synchronized String saveCollectConfig(FieldSetEntity fse) throws BaseException {
|
boolean isAdd = false;
|
if (StringUtils.isEmpty(fse.getUUID())) {
|
if (StringUtils.isEmpty(dataSystemName)) {
|
//数据来源系统名称不能为空
|
throw new BaseException(ErrorCode.DATA_SYSTEM_NAME_CAN_NOT_EMPTY);
|
}
|
Pattern r = Pattern.compile("^[A-Za-z-]+$");
|
Matcher m = r.matcher(dataSystemName);
|
if (!m.matches()) {
|
//数据来源系统名称格式不符
|
throw new BaseException(ErrorCode.DATA_SYSTEM_NAME_FORMAT_UNQUALIFIED);
|
}
|
//新增
|
FieldSetEntity fs = getBaseDao().getFieldSetEntityBySQL("SELECT\n" + "\t ifnull(MAX( id ),0)+1 id " + "FROM\n" + "\t( SELECT CONVERT(REPLACE ( id, ?, '' ),UNSIGNED int) id FROM product_sys_data_collect WHERE id LIKE concat(?,'%') ) a", new Object[]{dataSystemName, dataSystemName}, false);
|
if (!FieldSetEntity.isEmpty(fs) && !StringUtils.isEmpty(fs.getString("id"))) {
|
String id = fs.getString("id");
|
DecimalFormat decimalFormat = new DecimalFormat("000000");
|
id = decimalFormat.format(Integer.valueOf(id));
|
fse.setValue("id", dataSystemName + id);
|
} else {
|
throw new BaseException(ErrorCode.UNIQUE_COLLECT_CREATE_FAIL);
|
}
|
isAdd = true;
|
}
|
|
commonSave(fse, (fs) -> createTimeTask(fs.getString(CmnConst.TIME_TASK_UUID), "DC:" + fs.getString("name"), fs.getBoolean(CmnConst.IS_USED), fs.getString("frequency"), "dataCollectService.dataCollect('" + fs.getUUID() + "')"));
|
saveDelRecordConfig(1, fse.getUUID());
|
if (!"ch-kt".equals(dataSystemName)) {
|
//子库保存后发送到主服务
|
fse.setValue(CoreConst.SYSTEM_DATA_OPERATE_TYPE, isAdd ? "add" : "update");
|
mesExternalService.remoteSaveCollectConfig(fse);
|
}
|
return fse.getUUID();
|
}
|
|
/**
|
* 采集配置详情
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
public FieldSetEntity findCollectConfig(FieldSetEntity fse) throws BaseException {
|
return publicService.getFieldSetEntity(fse, false);
|
}
|
|
/**
|
* 采集配置删除
|
*
|
* @param fse
|
*/
|
@Override
|
@Transactional
|
public void delCollectConfig(FieldSetEntity fse) throws BaseException {
|
String uuids = fse.getUUID();
|
String[] split = uuids.split(",");
|
DataTableEntity dt = getBaseDao().listTable("product_sys_timed_task", "uuid in " + "(SELECT time_task_uuid FROM product_sys_data_collect WHERE " + BaseUtil.buildQuestionMarkFilter("uuid", split, true) + ")");
|
if (!DataTableEntity.isEmpty(dt)) {
|
for (int i = 0; i < dt.getRows(); i++) {
|
cancelTimeTask(dt.getFieldSetEntity(i));
|
}
|
}
|
for (String s : split) {
|
cancelDelRecordConfig(1, s);
|
}
|
publicService.delete(fse);
|
}
|
|
/**
|
* 提取配置保存
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
@Transactional
|
public String saveExtractConfig(FieldSetEntity fse) throws BaseException {
|
boolean isAdd = !StringUtils.isEmpty(fse.getUUID());
|
commonSave(fse, (fs) -> createTimeTask(fs.getString(CmnConst.TIME_TASK_UUID), "DE:" + fs.getString("extract_name"), fs.getBoolean(CmnConst.IS_USED), fs.getString("extract_time"), "dataExtractService.startExtractData('" + fs.getUUID() + "')"));
|
saveDelRecordConfig(2, fse.getUUID());
|
if ("ch-kt".equals(dataSystemName)) {
|
//主服务保存后分发到子服务中
|
fse.setValue(CoreConst.SYSTEM_DATA_OPERATE_TYPE, isAdd ? "add" : "update");
|
mesExternalService.remoteSaveExtractConfig(fse);
|
}
|
return fse.getUUID();
|
}
|
|
/**
|
* 提取配置查询
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
public FieldSetEntity findExtractConfig(FieldSetEntity fse) throws BaseException {
|
return publicService.getFieldSetEntity(fse, true);
|
}
|
|
/**
|
* 提取配置删除
|
*
|
* @param fse
|
*/
|
@Override
|
@Transactional
|
public void delExtractConfig(FieldSetEntity fse) throws BaseException {
|
String uuids = fse.getUUID();
|
String[] split = uuids.split(",");
|
DataTableEntity dt = getBaseDao().listTable("product_sys_timed_task", "uuid in " + "(SELECT time_task_uuid FROM product_sys_data_extract_config WHERE " + BaseUtil.buildQuestionMarkFilter("uuid", split, true) + ")");
|
if (!DataTableEntity.isEmpty(dt)) {
|
for (int i = 0; i < dt.getRows(); i++) {
|
cancelTimeTask(dt.getFieldSetEntity(i));
|
}
|
}
|
for (String s : split) {
|
cancelDelRecordConfig(2, s);
|
}
|
publicService.delete(fse);
|
}
|
|
/**
|
* 归档配置保存
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
@Transactional
|
public String saveArchivingConfig(FieldSetEntity fse) throws BaseException {
|
String sourceDataSource = fse.getString("source_data_source");
|
String targetDataSource = fse.getString("target_data_source");
|
String[] dataSource = (sourceDataSource + "," + targetDataSource).split(",");
|
StringBuilder sql = new StringBuilder();
|
sql.append("SELECT db_type num FROM product_sys_data_sync_manager WHERE ");
|
sql.append(BaseUtil.buildQuestionMarkFilter("uuid", dataSource.length, true));
|
sql.append(" GROUP BY db_type ");
|
DataTableEntity dt = getBaseDao().listTable(sql.toString(), dataSource);
|
if (DataTableEntity.isEmpty(dt)) {
|
throw new BaseException(ErrorCode.VERIFY_DATA_SOURCE_TYPE_FAIL);
|
}
|
/* 2022年10月10日 15:33:47 6c 需要能够支持不同数据库间的归档
|
if (dt.getRows() != 1) {
|
throw new BaseException(ErrorCode.DATA_SOURCE_INCONFORMITY);
|
}*/
|
if (StringUtils.equalsAny(dt.getString(0, "num"), "0", "1", "5")) {
|
boolean isAdd = !StringUtils.isEmpty(fse.getUUID());
|
commonSave(fse, (fs) -> createTimeTask(fs.getString(CmnConst.TIME_TASK_UUID), "DA:" + fs.getString("name"), fs.getBoolean(CmnConst.IS_USED), fs.getString("execute_time"), "dataArchivingService.dataArchivingEntry('" + fs.getUUID() + "')"));
|
if ("ch-kt".equals(dataSystemName)) {
|
//主服务保存后分发到子服务中
|
fse.setValue(CoreConst.SYSTEM_DATA_OPERATE_TYPE, isAdd ? "add" : "update");
|
mesExternalService.remoteSaveArchiveConfig(fse);
|
}
|
}
|
|
return fse.getUUID();
|
}
|
|
/**
|
* 归档配置详情
|
*
|
* @param fse
|
* @return
|
*/
|
@Override
|
public FieldSetEntity findArchivingConfig(FieldSetEntity fse) throws BaseException {
|
return publicService.getFieldSetEntity(fse, false);
|
}
|
|
/**
|
* 归档配置删除
|
*
|
* @param fse
|
*/
|
@Override
|
@Transactional
|
public void delArchivingConfig(FieldSetEntity fse) throws BaseException {
|
String uuids = fse.getUUID();
|
String[] split = uuids.split(",");
|
DataTableEntity dt = getBaseDao().listTable("product_sys_timed_task", "uuid in " + "(SELECT time_task_uuid FROM product_sys_data_archiving_config WHERE " + BaseUtil.buildQuestionMarkFilter("uuid", split, true) + ")");
|
if (!DataTableEntity.isEmpty(dt)) {
|
for (int i = 0; i < dt.getRows(); i++) {
|
cancelTimeTask(dt.getFieldSetEntity(i));
|
}
|
}
|
publicService.delete(fse);
|
}
|
|
/**
|
* 订单数据验证配置保存
|
*
|
* @param fse
|
* @return
|
*/
|
@Transactional
|
@Override
|
public String saveOrderDataVerification(FieldSetEntity fse) {
|
fse.setValue(CmnConst.UUID, "1");
|
commonSave(fse, fs -> createTimeTask(fs.getString(CmnConst.TIME_TASK_UUID), "订单数据同步验证", fs.getBoolean(CmnConst.IS_USED), fs.getString("frequency_verification"), "orderDataValidationService.verificationEntryPoint"));
|
return fse.getUUID();
|
}
|
|
/**
|
* 保存删除记录配置
|
*
|
* @param type
|
* @param uuid
|
*/
|
public void saveDelRecordConfig(int type, String uuid) {
|
String sql = getDelRecordSql(type);
|
|
DataTableEntity dt = getBaseDao().listTable(sql, new Object[]{uuid, dataSystemName});
|
for (int i = 0; i < dt.getRows(); i++) {
|
FieldSetEntity fse = dt.getFieldSetEntity(i);
|
fse.setTableName(CmnConst.PRODUCT_MES_DEL_RECORD_CONFIG);
|
fse.setValue("delete_sync_content", 1);
|
syncDelRecordService.createTrigger(fse);
|
}
|
|
}
|
|
private String getDelRecordSql(int type) {
|
StringBuffer sql = new StringBuffer();
|
sql.append("\n SELECT ");
|
sql.append("\n c.uuid, ");
|
sql.append("\n c.random_suffix,");
|
sql.append("\n collect.uuid config_uuid, ");
|
sql.append("\n collect.data_source, ");
|
sql.append("\n extract.extract_target_source target_data_source, ");
|
sql.append("\n collect.auto_field master_key_field, ");
|
sql.append("\n collect.source_table table_name, ");
|
sql.append("\n collect.id collect_id,2 del_type, ");
|
sql.append("\n extract.collect_source_field collect_field,");
|
sql.append("\n extract.extract_unique_field pre_master_field");
|
sql.append("\n FROM ");
|
sql.append("\n product_sys_data_collect collect ");
|
sql.append("\n JOIN product_sys_data_extract_config extract ON upper( case when length(collect.target_table)>0 then collect.target_table else collect.source_table end ) = upper( extract.extract_source_table ) ");
|
sql.append("\n LEFT JOIN product_mes_del_record_config c ON collect.uuid=c.config_uuid and c.del_type=2 ");
|
sql.append(" WHERE ");
|
if (1 == type) {
|
//采集
|
sql.append("\n collect.uuid=? ");
|
} else {
|
//提取
|
sql.append("\n extract.uuid=? ");
|
}
|
sql.append(" and collect.id like concat(?,'%')");
|
return sql.toString();
|
}
|
|
public void cancelDelRecordConfig(int type, String uuid) {
|
FieldSetEntity fieldSetEntity = new FieldSetEntity();
|
fieldSetEntity.setTableName("temp");
|
if (type == 1) {
|
fieldSetEntity.setValue("uuid", uuid);
|
syncDelRecordService.deleteTrigger(fieldSetEntity);
|
} else {
|
//提取
|
DataTableEntity dt = getBaseDao().listTable(CmnConst.PRODUCT_SYS_DATA_COLLECT, "UPPER(source_table) in (select upper(extract_source_table) from product_sys_data_extract_config where uuid =?) ", new Object[]{uuid});
|
if (DataTableEntity.isEmpty(dt)) {
|
return;
|
}
|
Object[] uuids = dt.getUuids();
|
for (Object o : uuids) {
|
fieldSetEntity.setValue("uuid", o);
|
syncDelRecordService.deleteTrigger(fieldSetEntity);
|
}
|
}
|
|
}
|
}
|
