chenxing
f7819b14e2
Local review candidate only. Explicitly selected inherited prerequisites and prior repair patches; unrelated WIP and independent MQ fixes are excluded. No push or deployment authorized.
175 lines
7.8 KiB
Java
175 lines
7.8 KiB
Java
package com.yunhai.flowcenter.callback;
|
||
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.yunhai.flowcenter.engine.FlowEvent;
|
||
import com.yunhai.flowcenter.entity.FlwCallbackLog;
|
||
import com.yunhai.flowcenter.entity.FlwCallbackReg;
|
||
import com.yunhai.flowcenter.entity.FlwDef;
|
||
import com.yunhai.flowcenter.entity.FlwInstance;
|
||
import com.yunhai.flowcenter.enums.FlwEnums;
|
||
import com.yunhai.flowcenter.repository.FlwCallbackLogRepository;
|
||
import com.yunhai.flowcenter.repository.FlwCallbackRegRepository;
|
||
import com.yunhai.flowcenter.repository.FlwDefRepository;
|
||
import com.yunhai.flowcenter.repository.FlwInstanceRepository;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.scheduling.annotation.Scheduled;
|
||
import org.springframework.stereotype.Component;
|
||
import org.springframework.transaction.event.TransactionPhase;
|
||
import org.springframework.transaction.event.TransactionalEventListener;
|
||
|
||
import java.time.LocalDateTime;
|
||
import java.time.format.DateTimeFormatter;
|
||
import java.util.HashMap;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
import java.util.UUID;
|
||
import java.util.concurrent.ExecutorService;
|
||
import java.util.concurrent.Executors;
|
||
|
||
/**
|
||
* 回调分发器（概览 §8.4 可靠性设计）：引擎 AFTER_COMMIT 事件 → 匹配 flw_callback_reg
|
||
* （flowGuid + 订阅事件）→ 落 flw_callback_log(status=0) → 异步首推。
|
||
* 落库成功即视为主业务完成，推送失败不影响审批（异步解耦是特性）。
|
||
* @Scheduled(fixedDelay=30s) 扫 (status=0, next_retry_time<=now) 重推；死信留管理端手工重推。
|
||
*/
|
||
@Slf4j
|
||
@Component
|
||
@RequiredArgsConstructor
|
||
public class CallbackDispatcher {
|
||
|
||
private static final DateTimeFormatter TIME = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
||
|
||
private final FlwInstanceRepository instanceRepository;
|
||
private final FlwDefRepository defRepository;
|
||
private final FlwCallbackRegRepository regRepository;
|
||
private final FlwCallbackLogRepository logRepository;
|
||
private final CallbackLogStore logStore;
|
||
private final CallbackPushService pushService;
|
||
private final CallbackProperties properties;
|
||
private final ObjectMapper objectMapper;
|
||
|
||
private volatile ExecutorService pushExecutor;
|
||
|
||
private ExecutorService executor() {
|
||
if (pushExecutor == null) {
|
||
synchronized (this) {
|
||
if (pushExecutor == null) {
|
||
pushExecutor = Executors.newFixedThreadPool(Math.max(1, properties.getPushPoolSize()));
|
||
}
|
||
}
|
||
}
|
||
return pushExecutor;
|
||
}
|
||
|
||
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, fallbackExecution = true)
|
||
public void onFlowEvent(FlowEvent event) {
|
||
// URGE 仅站内信通道，不进回调事件流（概览 §8.3 事件枚举固定六种）
|
||
if (event.getType() == FlowEvent.Type.URGE) {
|
||
return;
|
||
}
|
||
try {
|
||
dispatch(event);
|
||
} catch (Exception e) {
|
||
// 分发失败只影响回调（对账兜底会追平），绝不能反噬引擎主流程
|
||
log.error("[flow-center] 回调分发异常：instance={} event={} {}",
|
||
event.getInstanceId(), event.getType().getCode(), e.toString(), e);
|
||
}
|
||
}
|
||
|
||
private void dispatch(FlowEvent event) {
|
||
FlwInstance instance = instanceRepository.findById(event.getInstanceId()).orElse(null);
|
||
if (instance == null) {
|
||
return;
|
||
}
|
||
FlwDef def = defRepository.findById(instance.getDefId()).orElse(null);
|
||
if (def == null) {
|
||
return;
|
||
}
|
||
String eventCode = event.getType().getCode();
|
||
List<FlwCallbackReg> regs = regRepository.findBySchIdAndFlowGuidAndStatus(
|
||
instance.getSchId(), def.getFlowGuid(), 0);
|
||
for (FlwCallbackReg reg : regs) {
|
||
if (!subscribed(reg.getEventTypes(), eventCode)) {
|
||
continue;
|
||
}
|
||
persistAndPush(reg, instance, def, event);
|
||
}
|
||
}
|
||
|
||
private void persistAndPush(FlwCallbackReg reg, FlwInstance instance, FlwDef def, FlowEvent event) {
|
||
try {
|
||
String eventId = UUID.randomUUID().toString().replace("-", "").toLowerCase();
|
||
Map<String, Object> payload = new HashMap<>();
|
||
payload.put("eventId", eventId);
|
||
payload.put("event", event.getType().getCode());
|
||
payload.put("schId", instance.getSchId());
|
||
payload.put("flowGuid", def.getFlowGuid());
|
||
payload.put("instanceId", instance.getId());
|
||
payload.put("instanceVersion", instance.getVersion());
|
||
payload.put("bizType", instance.getBizType());
|
||
payload.put("bizId", instance.getBizId());
|
||
payload.put("bizTitle", instance.getBizTitle());
|
||
payload.put("occurredTime", LocalDateTime.now().format(TIME));
|
||
// 2026-09-20 信封升级：任务定位与操作人（均可空，办件级事件/历史重放体为 null——
|
||
// 契约见《子应用接入流程中心规范.md》§6.3/§6.4）
|
||
payload.put("taskId", event.getTaskId());
|
||
payload.put("operatorUserId", event.getOperatorId());
|
||
payload.put("operatorName", event.getOperatorName());
|
||
Map<String, Object> data = new HashMap<>();
|
||
if (event.getData() != null) {
|
||
data.putAll(event.getData());
|
||
}
|
||
// 身份在事件发布线程中冻结，不能用 AFTER_COMMIT 重读到的新轮给旧事件重新贴标。
|
||
if (event.getRoundIdentity() != null) {
|
||
data.putAll(event.getRoundIdentity());
|
||
}
|
||
payload.put("data", data);
|
||
String payloadJson = objectMapper.writeValueAsString(payload);
|
||
|
||
FlwCallbackLog callbackLog = new FlwCallbackLog();
|
||
callbackLog.setSchId(instance.getSchId());
|
||
callbackLog.setRegId(reg.getId());
|
||
callbackLog.setEventId(eventId);
|
||
callbackLog.setEvent(event.getType().getCode());
|
||
callbackLog.setInstanceId(instance.getId());
|
||
callbackLog.setBizType(instance.getBizType());
|
||
callbackLog.setBizId(instance.getBizId());
|
||
callbackLog.setPayloadJson(payloadJson);
|
||
callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
|
||
callbackLog.setRetryCount(0);
|
||
callbackLog.setNextRetryTime(LocalDateTime.now());
|
||
// REQUIRES_NEW 落库（afterCommit 阶段原事务已完结，REQUIRED 会挂空事务）
|
||
Long logId = logStore.persist(callbackLog);
|
||
// 异步首推：失败由退避重试接手（1m 起）
|
||
executor().submit(() -> pushService.pushAsync(logId));
|
||
} catch (Exception e) {
|
||
log.error("[flow-center] 回调日志落库失败（对账兜底会追平）：instance={} event={} {}",
|
||
instance.getId(), event.getType().getCode(), e.toString(), e);
|
||
}
|
||
}
|
||
|
||
/** 重推扫描（fixedDelay 串行防重入；推拉结合里 Webhook 永远只是加速器） */
|
||
@Scheduled(fixedDelay = 30_000)
|
||
public void scanRetry() {
|
||
List<FlwCallbackLog> due = logRepository.findTop50ByStatusAndNextRetryTimeLessThanEqualOrderByNextRetryTimeAsc(
|
||
FlwEnums.CallbackLogStatus.PENDING.getCode(), LocalDateTime.now());
|
||
for (FlwCallbackLog callbackLog : due) {
|
||
executor().submit(() -> pushService.pushAsync(callbackLog.getId()));
|
||
}
|
||
}
|
||
|
||
/** 订阅匹配：空=全部；逗号分隔事件码包含即命中 */
|
||
private static boolean subscribed(String eventTypes, String eventCode) {
|
||
if (eventTypes == null || eventTypes.trim().isEmpty()) {
|
||
return true;
|
||
}
|
||
for (String t : eventTypes.split(",")) {
|
||
if (eventCode.equals(t.trim())) {
|
||
return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
}
|