Browse Source

业务数据同步

dev-V1.10.2-temp
zhuibobo 4 months ago
parent
commit
e4be119fdb
  1. 67
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/controller/BizSyncManageController.java
  2. 113
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/SyncJobStatusHolder.java
  3. 11
      zdxt-web-client/zdxt-admin-plus-ui/src/api/efcode/biz/index.ts
  4. 227
      zdxt-web-client/zdxt-admin-plus-ui/src/views/efcode/biz/index.vue
  5. 33
      zdxt-web-client/zdxt-admin-plus-ui/src/views/efcode/interface/index.vue

67
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/controller/BizSyncManageController.java

@ -9,6 +9,7 @@ import com.zdxt.enforcementcode.businessdatasyn.domain.mapper.BizSyncLogMapper;
import com.zdxt.enforcementcode.businessdatasyn.domain.mapper.BizSyncRecordMapper;
import com.zdxt.enforcementcode.businessdatasyn.domain.mapper.BizSyncTaskMapper;
import com.zdxt.enforcementcode.businessdatasyn.executor.BusinessDataSyncScheduler;
import com.zdxt.enforcementcode.businessdatasyn.executor.SyncJobStatusHolder;
import com.zdxt.enforcementcode.businessdatasyn.service.SyncRecordService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@ -16,6 +17,10 @@ import org.dromara.common.core.domain.R;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 业务数据同步 - 管理接口
* <p>权限标识:efcode:bizsync:manage</p>
@ -34,35 +39,79 @@ public class BizSyncManageController {
private final BizSyncTaskMapper taskMapper;
private final BizSyncRecordMapper recordMapper;
private final BizSyncLogMapper logMapper;
private final SyncJobStatusHolder jobStatusHolder;
/** 专用线程池:串行执行同步任务,避免多个 force 并发压垮数据库 */
private static final ExecutorService SYNC_EXECUTOR = Executors.newSingleThreadExecutor(r -> {
Thread t = new Thread(r, "biz-sync-async");
t.setDaemon(true);
return t;
});
/**
* 手工触发指定表同步
* 手工触发指定表同步(异步,立即返回 jobId)
*/
@PostMapping("/trigger/{configKey}")
public R<String> trigger(@PathVariable String configKey) {
log.info("[BizSync] 手动触发同步: {}", configKey);
scheduler.runOne(configKey);
return R.ok("触发成功: " + configKey);
String jobId = jobStatusHolder.createJob("TRIGGER_ONE", configKey);
CompletableFuture.runAsync(() -> {
try {
scheduler.runOne(configKey);
jobStatusHolder.markSuccess(jobId, "同步完成: " + configKey);
} catch (Exception e) {
jobStatusHolder.markFail(jobId, e.getMessage());
}
}, SYNC_EXECUTOR);
return R.ok(jobId);
}
/**
* 手工触发全部同步
* 手工触发全部同步(异步)
*/
@PostMapping("/trigger-all")
public R<String> triggerAll() {
log.info("[BizSync] 手动触发全部同步");
scheduler.runAll();
return R.ok("全部触发成功");
String jobId = jobStatusHolder.createJob("TRIGGER_ALL", null);
CompletableFuture.runAsync(() -> {
try {
scheduler.runAll();
jobStatusHolder.markSuccess(jobId, "全部同步完成");
} catch (Exception e) {
jobStatusHolder.markFail(jobId, e.getMessage());
}
}, SYNC_EXECUTOR);
return R.ok(jobId);
}
/**
* 强制全量同步
* 强制全量同步(异步)
*/
@PostMapping("/trigger-force")
public R<String> triggerForce() {
log.info("[BizSync] 手动强制全量同步");
scheduler.runAllForce();
return R.ok("强制全量触发成功");
String jobId = jobStatusHolder.createJob("TRIGGER_FORCE", null);
CompletableFuture.runAsync(() -> {
try {
scheduler.runAllForce();
jobStatusHolder.markSuccess(jobId, "强制全量同步完成");
} catch (Exception e) {
jobStatusHolder.markFail(jobId, e.getMessage());
}
}, SYNC_EXECUTOR);
return R.ok(jobId);
}
/**
* 查询异步任务状态(前端轮询用)
*/
@GetMapping("/job-status/{jobId}")
public R<SyncJobStatusHolder.JobStatus> jobStatus(@PathVariable String jobId) {
SyncJobStatusHolder.JobStatus status = jobStatusHolder.getJob(jobId);
if (status == null) {
return R.fail("任务不存在或已过期");
}
return R.ok(status);
}
/**

113
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/SyncJobStatusHolder.java

@ -0,0 +1,113 @@
package com.zdxt.enforcementcode.businessdatasyn.executor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
/**
* 同步任务运行状态持有器
* <p>基于内存 ConcurrentHashMap,用于前端轮询判断异步任务是否完成</p>
* <p>注意:仅供 UI 触发的同步任务使用;Quartz Job 和单元测试直接调 scheduler,不经过本类</p>
* <p>单节点部署有效,多节点部署需改为 Redis</p>
*
* @author zrb
*/
@Slf4j
@Component
public class SyncJobStatusHolder {
/** 任务过期时间(毫秒),完成后 30 分钟自动清理 */
private static final long EXPIRE_MS = 30 * 60 * 1000L;
private final Map<String, JobStatus> jobs = new ConcurrentHashMap<>();
/**
* 创建一个新任务,返回 jobId
*/
public String createJob(String jobType, String configKey) {
String jobId = UUID.randomUUID().toString().replace("-", "");
JobStatus status = new JobStatus();
status.setJobId(jobId);
status.setJobType(jobType);
status.setConfigKey(configKey);
status.setStatus("RUNNING");
status.setStartTime(LocalDateTime.now());
jobs.put(jobId, status);
log.info("[BizSync-Job] 创建任务: jobId={}, type={}, configKey={}", jobId, jobType, configKey);
cleanExpired();
return jobId;
}
/**
* 标记任务成功
*/
public void markSuccess(String jobId, String message) {
JobStatus status = jobs.get(jobId);
if (status != null) {
status.setStatus("SUCCESS");
status.setEndTime(LocalDateTime.now());
status.setMessage(message);
log.info("[BizSync-Job] 任务成功: jobId={}, msg={}", jobId, message);
}
}
/**
* 标记任务失败
*/
public void markFail(String jobId, String errorMsg) {
JobStatus status = jobs.get(jobId);
if (status != null) {
status.setStatus("FAIL");
status.setEndTime(LocalDateTime.now());
status.setMessage(errorMsg);
log.error("[BizSync-Job] 任务失败: jobId={}, err={}", jobId, errorMsg);
}
}
/**
* 获取任务状态
*/
public JobStatus getJob(String jobId) {
return jobs.get(jobId);
}
/**
* 清理过期任务(完成时间超过 EXPIRE_MS 的)
*/
private void cleanExpired() {
long now = System.currentTimeMillis();
jobs.entrySet().removeIf(entry -> {
JobStatus s = entry.getValue();
if (s.getEndTime() == null) return false;
long endMillis = s.getEndTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
return (now - endMillis) > EXPIRE_MS;
});
}
/**
* 任务状态 DTO
*/
@Data
public static class JobStatus {
/** 任务 UUID */
private String jobId;
/** 任务类型: TRIGGER_ONE / TRIGGER_ALL / TRIGGER_FORCE */
private String jobType;
/** 关联配置 key(TRIGGER_ONE 时有值) */
private String configKey;
/** 状态: RUNNING / SUCCESS / FAIL */
private String status;
/** 提示信息 */
private String message;
/** 开始时间 */
private LocalDateTime startTime;
/** 结束时间 */
private LocalDateTime endTime;
}
}

11
zdxt-web-client/zdxt-admin-plus-ui/src/api/efcode/biz/index.ts

@ -41,6 +41,17 @@ export const getTaskList = (params: { configKey?: string; pageNum?: number; page
});
};
/**
* 查询异步任务状态(前端轮询用,silent 不弹错误提示)
*/
export const getJobStatus = (jobId: string) => {
return request({
url: '/web/biz-sync/job-status/' + jobId,
method: 'get',
headers: { silent: true }
});
};
/**
* 分页查询同步日志
*/

227
zdxt-web-client/zdxt-admin-plus-ui/src/views/efcode/biz/index.vue

@ -1,12 +1,15 @@
<script setup lang="ts">
import { ref, reactive, onMounted } from 'vue';
import { ElMessage, ElMessageBox } from 'element-plus';
import { ref, reactive, onMounted, onUnmounted } from 'vue';
import { ElMessage, ElMessageBox, ElNotification, ElLoading } from 'element-plus';
import type { LoadingInstance } from 'element-plus/es/components/loading/src/loading';
import { Loading } from '@element-plus/icons-vue';
import {
triggerSync,
triggerAllSync,
triggerForceSync,
getTaskList,
getLogList
getLogList,
getJobStatus
} from '@/api/efcode/biz';
/**
@ -22,19 +25,154 @@ const syncConfigs = ref([
v2Table: 'enforcement_registration',
syncDirection: '双向',
enabled: true
},
{
configKey: 'enterprise_information',
description: '企业信息表双向同步',
v1Table: 'enterprise_information',
v2Table: 'enterprise_information',
syncDirection: '双向',
enabled: false
}
]);
const loading = ref(false);
// ============ 异步任务轮询状态 ============
interface RunningJob {
jobId: string;
jobType: string;
configKey?: string;
startAt: number;
}
const runningJob = ref<RunningJob | null>(null);
const elapsedSec = ref(0);
let elapsedTimer: any = null;
let pollTimer: any = null;
let loadingInstance: LoadingInstance | null = null;
/** 提交异步任务并开始轮询 */
const submitAsyncJob = async (
apiCall: () => Promise<any>,
jobType: string,
configKey?: string,
useMask: boolean = false,
maskText: string = '执行中,请稍候...'
) => {
loading.value = true;
try {
const res: any = await apiCall();
const jobId = res?.data || res?.msg;
if (!jobId) {
ElMessage.error('提交失败:未返回 jobId');
loading.value = false;
return;
}
runningJob.value = { jobId, jobType, configKey, startAt: Date.now() };
if (useMask) {
openFullscreenMask(maskText);
}
startElapsedTimer();
pollJobStatus(jobId);
} catch (e: any) {
ElMessage.error(e?.msg || '提交失败');
loading.value = false;
}
};
/** 开启全屏遮罩 */
const openFullscreenMask = (text: string) => {
closeFullscreenMask();
loadingInstance = ElLoading.service({
lock: true,
fullscreen: true,
text,
background: 'rgba(0, 0, 0, 0.7)',
customClass: 'biz-sync-fullscreen-loading'
});
};
/** 关闭遮罩 */
const closeFullscreenMask = () => {
if (loadingInstance) {
loadingInstance.close();
loadingInstance = null;
}
};
/** 启动计时器 */
const startElapsedTimer = () => {
elapsedSec.value = 0;
clearInterval(elapsedTimer);
elapsedTimer = setInterval(() => {
if (runningJob.value) {
elapsedSec.value = Math.floor((Date.now() - runningJob.value.startAt) / 1000);
// 同步更新遮罩文本
if (loadingInstance) {
const typeLabel = runningJob.value.jobType === 'TRIGGER_FORCE'
? '强制全量同步'
: runningJob.value.jobType === 'TRIGGER_ALL'
? '全部增量同步'
: '同步';
(loadingInstance as any).setText(
`${typeLabel}运行中... 已耗时 ${formatElapsed(elapsedSec.value)}(请不要关闭页面)`
);
}
}
}, 1000);
};
/** 停止定时器 */
const stopTimers = () => {
clearInterval(elapsedTimer);
clearTimeout(pollTimer);
elapsedTimer = null;
pollTimer = null;
};
/** 任务结束清理 */
const finishJob = () => {
stopTimers();
closeFullscreenMask();
runningJob.value = null;
loading.value = false;
};
/** 轮询任务状态 */
const pollJobStatus = async (jobId: string) => {
try {
const res: any = await getJobStatus(jobId);
const st = res?.data;
if (!st) {
ElMessage.warning('任务状态丢失,可能已过期');
finishJob();
return;
}
if (st.status === 'RUNNING') {
// 3 秒一次轮询
pollTimer = setTimeout(() => pollJobStatus(jobId), 3000);
} else if (st.status === 'SUCCESS') {
ElNotification({
title: '同步成功',
message: st.message || '任务已完成',
type: 'success',
duration: 6000
});
finishJob();
} else {
ElNotification({
title: '同步失败',
message: st.message || '任务执行失败',
type: 'error',
duration: 0
});
finishJob();
}
} catch (e) {
// 网络异常,5 秒后重试
pollTimer = setTimeout(() => pollJobStatus(jobId), 5000);
}
};
/** 格式化耗时 */
const formatElapsed = (sec: number) => {
const m = Math.floor(sec / 60);
const s = sec % 60;
return m > 0 ? `${m}分${s}秒` : `${s}秒`;
};
// ============ 任务记录弹窗 ============
const taskDialogVisible = ref(false);
const taskLoading = ref(false);
@ -62,57 +200,63 @@ const logQuery = reactive({
/** 触发单个表增量同步 */
const handleTrigger = async (row: any) => {
if (runningJob.value) {
ElMessage.warning('已有任务在运行中,请等待完成');
return;
}
try {
await ElMessageBox.confirm(
`确认触发【${row.description}】增量同步?`,
'确认',
{ type: 'warning' }
);
loading.value = true;
const res: any = await triggerSync(row.configKey);
ElMessage.success(res.msg || '触发成功');
await submitAsyncJob(() => triggerSync(row.configKey), 'TRIGGER_ONE', row.configKey);
} catch (e: any) {
if (e !== 'cancel' && e?.action !== 'cancel') {
ElMessage.error(e?.msg || '触发失败');
}
} finally {
loading.value = false;
}
};
/** 触发全部同步 */
const handleTriggerAll = async () => {
if (runningJob.value) {
ElMessage.warning('已有任务在运行中,请等待完成');
return;
}
try {
await ElMessageBox.confirm('确认触发全部表的增量同步?', '确认', { type: 'warning' });
loading.value = true;
const res: any = await triggerAllSync();
ElMessage.success(res.msg || '全部触发成功');
await submitAsyncJob(triggerAllSync, 'TRIGGER_ALL');
} catch (e: any) {
if (e !== 'cancel' && e?.action !== 'cancel') {
ElMessage.error(e?.msg || '触发失败');
}
} finally {
loading.value = false;
}
};
/** 强制全量同步 */
const handleForceSync = async () => {
if (runningJob.value) {
ElMessage.warning('已有任务在运行中,请等待完成');
return;
}
try {
await ElMessageBox.confirm(
'强制全量同步将扫描全部数据,耗时较长,确认执行?',
'强制全量同步将扫描全部数据,耗时较长(可能几分钟到十几分钟),确认执行?',
'警告',
{ type: 'error', confirmButtonText: '确定执行', cancelButtonText: '取消' }
);
loading.value = true;
const res: any = await triggerForceSync();
ElMessage.success(res.msg || '强制全量触发成功');
await submitAsyncJob(
triggerForceSync,
'TRIGGER_FORCE',
undefined,
true,
'强制全量同步启动中...'
);
} catch (e: any) {
if (e !== 'cancel' && e?.action !== 'cancel') {
ElMessage.error(e?.msg || '触发失败');
}
} finally {
loading.value = false;
}
};
@ -193,10 +337,33 @@ const formatTime = (val: string) => {
};
onMounted(() => {});
onUnmounted(() => {
stopTimers();
closeFullscreenMask();
});
</script>
<template>
<div class="app-container">
<!-- 运行中任务提示条 -->
<el-alert
v-if="runningJob"
type="warning"
:closable="false"
style="margin-bottom: 16px"
>
<template #title>
<el-icon class="is-loading" style="vertical-align: middle; margin-right: 6px">
<Loading />
</el-icon>
<span>
同步任务运行中... (类型: {{ runningJob.jobType }}
<template v-if="runningJob.configKey">, 配置: {{ runningJob.configKey }}</template>
, 已运行: {{ formatElapsed(elapsedSec) }})
</span>
</template>
</el-alert>
<el-card shadow="never">
<template #header>
<div class="card-header">
@ -207,10 +374,10 @@ onMounted(() => {});
</el-tag>
</div>
<div class="header-actions">
<el-button type="primary" :loading="loading" @click="handleTriggerAll">
<el-button type="primary" :loading="loading" :disabled="!!runningJob" @click="handleTriggerAll">
全部增量同步
</el-button>
<el-button type="danger" :loading="loading" @click="handleForceSync">
<el-button type="danger" :loading="loading" :disabled="!!runningJob" @click="handleForceSync">
强制全量同步
</el-button>
</div>
@ -245,7 +412,7 @@ onMounted(() => {});
type="primary"
link
size="small"
:disabled="!row.enabled"
:disabled="!row.enabled || !!runningJob"
@click="handleTrigger(row)"
>
增量同步

33
zdxt-web-client/zdxt-admin-plus-ui/src/views/efcode/interface/index.vue

@ -12,39 +12,6 @@ import { invokeInterface, type InterfaceItem } from '@/api/efcode/interface';
* 当前示例:AppEnfCpPlanController(/app-enf/cpPlan)
*/
const interfaceConfigs = ref<InterfaceItem[]>([
{
name: '我的计划列表',
group: 'AppEnfCpPlanController',
url: '/app-enf/cpPlan/list',
method: 'get',
description: '查询当前登录用户创建的计划(分页)',
params: { pageNum: 1, pageSize: 10 }
},
{
name: '其他计划列表',
group: 'AppEnfCpPlanController',
url: '/app-enf/cpPlan/otherList',
method: 'get',
description: '本单位全部 + 其他单位公开(仅待预约)',
params: { pageNum: 1, pageSize: 10 }
},
{
name: '预约前置检查',
group: 'AppEnfCpPlanController',
url: '/app-enf/cpPlan/reserve/check/{targetPlanId}',
method: 'get',
description: '预约前置检查,需填写 targetPlanId',
pathParams: { targetPlanId: '' }
},
{
name: '执行预约',
group: 'AppEnfCpPlanController',
url: '/app-enf/cpPlan/reserve/do',
method: 'post',
description: '执行预约,提交 targetPlanId、opMessage',
params: { targetPlanId: '', opMessage: '' },
confirm: true
},
{
name: '计划自动匹配预约',
group: 'AppEnfCpPlanController',

Loading…
Cancel
Save