flow-center/flow-center-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 提交流程中心后端代码
build / build-push (push) Successful in 28s
Details

Browse Source 
Spring Boot 流程引擎服务：流程定义/实例/任务、审批人解析、
回调分发、oapi 接口及单元测试。
保留远程 .gitea/workflows/build.yml 与 docker/Dockerfile，
新增 .gitignore 排除 target/ 构建产物。
This commit is contained in:
chenxing
2026-09-15 10:42:56 +08:00
parent 0296b2b5dd
commit c1103aeb29
65 changed files with 7119 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/yunhai/flowcenter/callback/CallbackDispatcher.java
+165
View File
@@ -0,0 +1,165 @@
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));
Map<String, Object> data = new HashMap<>();
if (event.getData() != null) {
data.putAll(event.getData());
}
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;
}
}
src/main/java/com/yunhai/flowcenter/callback/CallbackLogStore.java
+27
View File
@@ -0,0 +1,27 @@
package com.yunhai.flowcenter.callback;
import com.yunhai.flowcenter.entity.FlwCallbackLog;
import com.yunhai.flowcenter.repository.FlwCallbackLogRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/**
* 回调日志落库（独立 bean 保证 REQUIRES_NEW 生效）：AFTER_COMMIT 监听阶段原事务已提交
* 但事务资源仍挂在线程上，REQUIRED 会"加入"已完结事务并抛 no transaction is in progress——
* 必须显式开新事务写日志。落库成功即视为主业务完成，推送失败不影响审批。
*/
@Service
@RequiredArgsConstructor
public class CallbackLogStore {
private final FlwCallbackLogRepository logRepository;
/** @return 落库后的日志行 id */
@Transactional(propagation = Propagation.REQUIRES_NEW)
public Long persist(FlwCallbackLog callbackLog) {
FlwCallbackLog saved = logRepository.saveAndFlush(callbackLog);
return saved.getId();
}
}
src/main/java/com/yunhai/flowcenter/callback/CallbackProperties.java
+38
View File
@@ -0,0 +1,38 @@
package com.yunhai.flowcenter.callback;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 回调推送配置（application.yml flow-center.callback.*）。
*/
@Data
@Component
@ConfigurationProperties(prefix = "flow-center.callback")
public class CallbackProperties {
/** 推送线程池大小（概览 §8.4：单机小并发 5） */
private int pushPoolSize = 5;
/**
* 退避序列（分钟）：1m→5m→30m→2h→6h；越过序列后按 {@link #capBackoffMinutes} 封顶持续重试，
* 不自动死信（持续重试语义：消费方恢复即自动追平，消费方免对账）。
*/
private List<Long> retryBackoffMinutes = Arrays.asList(1L, 5L, 30L, 120L, 360L);
/**
* 挂起告警阈值：retryCount 首次等于该值时 UserMsg 告警恰一次（retryCount 单调递增保证）。
* 默认 6 = 越过退避表（5 项）后的第一次失败。
*/
private int hangAlertRetries = 6;
/** 越过退避表后的封顶重试间隔（分钟），默认 60 */
private long capBackoffMinutes = 60;
/** 告警接收人（本校管理员 userId，逗号分隔；空=不告警只记日志）——挂起告警与死信告警共用 */
private String alertUserIds = "";
}
src/main/java/com/yunhai/flowcenter/callback/CallbackPushService.java
+243
View File
@@ -0,0 +1,243 @@
package com.yunhai.flowcenter.callback;
import com.yunhai.flowcenter.entity.FlwCallbackLog;
import com.yunhai.flowcenter.entity.FlwCallbackReg;
import com.yunhai.flowcenter.enums.FlwEnums;
import com.yunhai.flowcenter.repository.FlwCallbackLogRepository;
import com.yunhai.flowcenter.repository.FlwCallbackRegRepository;
import com.yunhai.flowcenter.notify.UserMsgClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.Base64;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
/**
* 回调推送执行（概览 §8.4 协议）：POST callback_url + X-Flw-* 四头
* （Signature = Base64(HmacSHA256(secret, timestamp + "." + nonce + "." + body))）。
* 2xx 即成功（不解析响应体）；失败按退避序列排下次重试，越过序列后封顶 cap-backoff-minutes
* <b>持续重试不终止</b>（T1.1：消费方只承诺"瞬时失败回 5xx"，恢复后自动追平，消费方免对账）；
* 挂起（retryCount 首达 hang-alert-retries）时 UserMsg 告警恰一次；
* 死信（status=2）仅两个入口——注册停用、管理端手工标记，重试次数永不自动触发。
* 推送一律在新事务里落库（AFTER_COMMIT 监听器/调度线程均无外层事务）。
* 推送本身与落库解耦：payload_json 是完整快照，可重放。
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class CallbackPushService {
private static final Set<Long> IN_FLIGHT = new HashSet<>();
private final FlwCallbackLogRepository logRepository;
private final FlwCallbackRegRepository regRepository;
private final CallbackProperties properties;
private final UserMsgClient userMsgClient;
/** 推送指定日志行（幂等入口：成功/终态/在途跳过） */
public void pushAsync(Long logId) {
synchronized (IN_FLIGHT) {
if (!IN_FLIGHT.add(logId)) {
return;
}
}
try {
doPush(logId);
} finally {
synchronized (IN_FLIGHT) {
IN_FLIGHT.remove(logId);
}
}
}
private void doPush(Long logId) {
FlwCallbackLog callbackLog = logRepository.findById(logId).orElse(null);
if (callbackLog == null
|| callbackLog.getStatus() != FlwEnums.CallbackLogStatus.PENDING.getCode()) {
return;
}
FlwCallbackReg reg = regRepository.findById(callbackLog.getRegId()).orElse(null);
if (reg == null || reg.getStatus() != 0) {
// 注册已停用：保留日志待查，不再推送（置死信便于管理端一次性清理）
markDead(callbackLog, "registration disabled");
return;
}
String body = callbackLog.getPayloadJson();
String ts = String.valueOf(System.currentTimeMillis());
String nonce = UUID.randomUUID().toString().replace("-", "").substring(0, 16);
try {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
headers.set("X-Flw-EventId", callbackLog.getEventId());
headers.set("X-Flw-Timestamp", ts);
headers.set("X-Flw-Nonce", nonce);
headers.set("X-Flw-Signature", sign(reg.getSecret(), ts, nonce, body));
RestTemplate template = rawTemplate();
ResponseEntity<String> resp = template.exchange(reg.getCallbackUrl(), HttpMethod.POST,
new HttpEntity<>(body, headers), String.class);
int status = resp.getStatusCodeValue();
if (status >= 200 && status < 300) {
markSuccess(callbackLog, status);
} else {
markRetryOrFail(callbackLog, "HTTP " + status, String.valueOf(status));
}
} catch (Exception e) {
markRetryOrFail(callbackLog, rootMessage(e), null);
}
}
private void markSuccess(FlwCallbackLog callbackLog, int httpStatus) {
callbackLog.setStatus(FlwEnums.CallbackLogStatus.SUCCESS.getCode());
callbackLog.setRetryCount(callbackLog.getRetryCount() == null ? 0 : callbackLog.getRetryCount());
callbackLog.setLastHttpStatus(String.valueOf(httpStatus));
callbackLog.setLastError(null);
callbackLog.setPushTime(LocalDateTime.now());
saveQuietly(callbackLog);
log.debug("[flow-center] 回调推送成功：eventId={} http={}", callbackLog.getEventId(), httpStatus);
}
/**
* 失败落定（持续重试语义）：越过退避表后按 {@link CallbackProperties#getCapBackoffMinutes()}
* 封顶继续排期，<b>绝不自动死信</b>；retryCount 首次等于挂起阈值时告警恰一次。
*/
private void markRetryOrFail(FlwCallbackLog callbackLog, String error, String httpStatus) {
int retryCount = (callbackLog.getRetryCount() == null ? 0 : callbackLog.getRetryCount()) + 1;
callbackLog.setRetryCount(retryCount);
callbackLog.setLastHttpStatus(httpStatus);
callbackLog.setLastError(truncate(error));
java.util.List<Long> backoff = properties.getRetryBackoffMinutes();
long backoffMinutes = retryCount <= backoff.size()
? backoff.get(retryCount - 1)
: Math.max(1, properties.getCapBackoffMinutes());
callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode());
callbackLog.setNextRetryTime(LocalDateTime.now().plusMinutes(backoffMinutes));
saveQuietly(callbackLog);
log.warn("[flow-center] 回调推送失败（第 {} 次重试，{} 分钟后）：eventId={} {}",
retryCount, backoffMinutes, callbackLog.getEventId(), error);
// 挂起告警恰一次：retryCount 每失败 +1 单调递增，等于阈值的时机只有一次；
// msgNo("hang:"+eventId) 幂等双保险（管理端 rePush 重置 retryCount 后再挂起可再告警，属人工重启语义）
if (retryCount == properties.getHangAlertRetries()) {
alertHung(callbackLog, error, backoffMinutes);
}
}
/** 挂起告警本校管理员（持续重试越过退避表=消费方疑似停机；UserMsg 不可达只记日志，不影响重试） */
private void alertHung(FlwCallbackLog callbackLog, String error, long capMinutes) {
java.util.List<Integer> userIds = parseAlertUserIds();
if (userIds.isEmpty()) {
return;
}
try {
userMsgClient.sendUserMsg(callbackLog.getSchId().intValue(), userIds, 1,
"flow.hung", String.valueOf(callbackLog.getInstanceId()),
"回调挂起告警",
"办件 " + callbackLog.getInstanceId() + " 事件 " + callbackLog.getEvent()
+ " 推送连续失败 " + callbackLog.getRetryCount() + " 次，已转入持续重试（封顶 "
+ capMinutes + " 分钟/次）。消费方恢复后将自动追平，无需人工干预；"
+ "如确认该事件已无意义，请在管理端标记死信。最近错误：" + truncate(error),
null, "hang:" + callbackLog.getEventId(), null);
} catch (Exception e) {
log.warn("[flow-center] 挂起告警发送失败：eventId={} {}", callbackLog.getEventId(), e.toString());
}
}
/**
* 标记死信（public：管理端手工标记的唯一执行入口；重试路径已不再自动调用）。
* 仍被注册停用场景使用（doPush：注册已停用即无意义重试，置死信便于管理端清理）。
*/
public void markDead(FlwCallbackLog callbackLog, String error) {
callbackLog.setStatus(FlwEnums.CallbackLogStatus.DEAD.getCode());
callbackLog.setLastError(truncate(error));
saveQuietly(callbackLog);
log.error("[flow-center] 回调推送进入死信：eventId={} biz={}/{} error={}",
callbackLog.getEventId(), callbackLog.getBizType(), callbackLog.getBizId(), error);
alertDead(callbackLog, error);
}
private java.util.List<Integer> parseAlertUserIds() {
java.util.List<Integer> userIds = new java.util.ArrayList<>();
String alertUserIds = properties.getAlertUserIds();
if (alertUserIds == null || alertUserIds.trim().isEmpty()) {
return userIds;
}
for (String s : alertUserIds.split(",")) {
if (!s.trim().isEmpty()) {
userIds.add(Integer.valueOf(s.trim()));
}
}
return userIds;
}
/** 死信告警本校管理员（配置 alertUserIds；UserMsg 不可达只记日志，不影响死信落定） */
private void alertDead(FlwCallbackLog callbackLog, String error) {
java.util.List<Integer> userIds = parseAlertUserIds();
if (userIds.isEmpty()) {
return;
}
try {
userMsgClient.sendUserMsg(callbackLog.getSchId().intValue(), userIds, 1,
"flow.dead", String.valueOf(callbackLog.getInstanceId()),
"回调推送死信告警",
"办件 " + callbackLog.getInstanceId() + " 事件 " + callbackLog.getEvent()
+ " 推送失败进入死信：" + error,
null, "dead:" + callbackLog.getEventId(), null);
} catch (Exception e) {
log.warn("[flow-center] 死信告警发送失败：eventId={} {}", callbackLog.getEventId(), e.toString());
}
}
private void saveQuietly(FlwCallbackLog callbackLog) {
try {
logRepository.save(callbackLog);
} catch (Exception e) {
log.error("[flow-center] 回调日志落库失败：eventId={} {}", callbackLog.getEventId(), e.toString());
}
}
private static String sign(String secret, String timestamp, String nonce, String body) {
try {
Mac mac = Mac.getInstance("HmacSHA256");
mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
byte[] raw = mac.doFinal((timestamp + "." + nonce + "." + body).getBytes(StandardCharsets.UTF_8));
return Base64.getEncoder().encodeToString(raw);
} catch (Exception e) {
throw new IllegalStateException("HmacSHA256 计算失败", e);
}
}
private RestTemplate rawTemplate() {
SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
factory.setConnectTimeout(5000);
factory.setReadTimeout(10000);
return new RestTemplate(factory);
}
private static String rootMessage(Throwable e) {
Throwable cur = e;
while (cur.getCause() != null && cur.getCause() != cur) {
cur = cur.getCause();
}
return cur.toString();
}
private static String truncate(String s) {
if (s == null) {
return null;
}
return s.length() > 1000 ? s.substring(0, 1000) : s;
}
}