diff --git a/db/0pigxx_boot.sql b/db/0pigxx_boot.sql index 769f89834..e78070238 100644 --- a/db/0pigxx_boot.sql +++ b/db/0pigxx_boot.sql @@ -1857,32 +1857,6 @@ CREATE TABLE `flow_notice` ( KEY `idx_notice_due` (`status`,`scheduled_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件'; --- ---------------------------- --- Table structure for flow_callback --- ---------------------------- -DROP TABLE IF EXISTS `flow_callback`; -CREATE TABLE `flow_callback` ( - `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调事件ID，同时作为接收方幂等键', - `tenant_id` bigint NOT NULL COMMENT '所属租户ID', - `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID', - `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID', - `target_url` varchar(2000) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调目标地址', - `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调请求体 JSON', - `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '状态：PENDING 待投递，SENDING 投递中，DELIVERED 已送达，FAILED 失败', - `attempts` int NOT NULL DEFAULT '0' COMMENT '已尝试次数', - `max_attempts` int NOT NULL DEFAULT '5' COMMENT '最大尝试次数', - `next_at` timestamp(3) NOT NULL COMMENT '下次投递时间', - `history` longtext COLLATE utf8mb4_general_ci COMMENT '投递历史 JSON', - `lease_token` varchar(36) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '投递租约令牌', - `lease_until` timestamp(3) NULL DEFAULT NULL COMMENT '租约到期时间', - `last_error` varchar(500) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '最近一次失败原因', - `created_at` timestamp(3) NOT NULL COMMENT '创建时间', - `updated_at` timestamp(3) NOT NULL COMMENT '更新时间', - PRIMARY KEY (`id`), - UNIQUE KEY `uk_callback_task` (`tenant_id`,`task_id`), - KEY `idx_callback_due` (`status`,`next_at`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程节点 HTTP 回调'; - -- ---------------------------- -- Table structure for process_starter -- ---------------------------- diff --git a/db/3pigxx_flow.sql b/db/3pigxx_flow.sql index 7840937af..67c4e344d 100644 --- a/db/3pigxx_flow.sql +++ b/db/3pigxx_flow.sql @@ -330,30 +330,4 @@ CREATE TABLE `flow_notice` ( KEY `idx_notice_due` (`status`,`scheduled_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件'; --- ---------------------------- --- Table structure for flow_callback --- ---------------------------- -DROP TABLE IF EXISTS `flow_callback`; -CREATE TABLE `flow_callback` ( - `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调事件ID，同时作为接收方幂等键', - `tenant_id` bigint NOT NULL COMMENT '所属租户ID', - `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID', - `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID', - `target_url` varchar(2000) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调目标地址', - `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调请求体 JSON', - `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '状态：PENDING 待投递，SENDING 投递中，DELIVERED 已送达，FAILED 失败', - `attempts` int NOT NULL DEFAULT '0' COMMENT '已尝试次数', - `max_attempts` int NOT NULL DEFAULT '5' COMMENT '最大尝试次数', - `next_at` timestamp(3) NOT NULL COMMENT '下次投递时间', - `history` longtext COLLATE utf8mb4_general_ci COMMENT '投递历史 JSON', - `lease_token` varchar(36) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '投递租约令牌', - `lease_until` timestamp(3) NULL DEFAULT NULL COMMENT '租约到期时间', - `last_error` varchar(500) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '最近一次失败原因', - `created_at` timestamp(3) NOT NULL COMMENT '创建时间', - `updated_at` timestamp(3) NOT NULL COMMENT '更新时间', - PRIMARY KEY (`id`), - UNIQUE KEY `uk_callback_task` (`tenant_id`,`task_id`), - KEY `idx_callback_due` (`status`,`next_at`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程节点 HTTP 回调'; - SET FOREIGN_KEY_CHECKS = 1; diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java deleted file mode 100644 index e71ed1d25..000000000 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java +++ /dev/null @@ -1,23 +0,0 @@ -package com.pig4cloud.pigx.flow.constant; - -import lombok.Getter; -import lombok.RequiredArgsConstructor; - -/** 回调投递状态；编码与 flow_callback.status 保持兼容，不表示单次尝试的审计结果。 */ -@Getter -@RequiredArgsConstructor -public enum FlowCallbackStatusEnum { - - /** 等待首次投递或到期重试。 */ - PENDING("PENDING"), - /** 已领取租约，正在投递；过期后可重新领取。 */ - SENDING("SENDING"), - /** 接收方返回 HTTP 2xx，投递结束。 */ - DELIVERED("DELIVERED"), - /** 尝试耗尽或最终租约过期，需人工核对接收方后处理。 */ - FAILED("FAILED"); - - /** 持久化编码，不能随枚举重命名改变。 */ - private final String code; - -} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java deleted file mode 100644 index e723c49eb..000000000 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java +++ /dev/null @@ -1,66 +0,0 @@ -package com.pig4cloud.pigx.flow.entity; - -import com.baomidou.mybatisplus.annotation.IdType; -import com.baomidou.mybatisplus.annotation.TableId; -import com.baomidou.mybatisplus.annotation.TableName; -import com.pig4cloud.pigx.common.core.util.TenantTable; -import lombok.Data; - -import java.time.LocalDateTime; - -/** 节点完成 HTTP 回调；租约保证多实例互斥，事件 ID 供接收方幂等。 */ -@Data -@TenantTable -@TableName("flow_callback") -public class FlowCallback { - - /** 回调事件 ID。 */ - @TableId(type = IdType.ASSIGN_UUID) - private String id; - - /** 所属租户 ID。 */ - private Long tenantId; - - /** 引擎任务 ID，同一任务只入队一次。 */ - private String taskId; - - /** 引擎流程实例 ID。 */ - private String instanceId; - - /** 回调目标地址。 */ - private String targetUrl; - - /** 回调请求体 JSON。 */ - private String payload; - - /** 状态，见 FlowCallbackStatusEnum。 */ - private String status; - - /** 已尝试次数；为空时使用数据库默认值 0。 */ - private Integer attempts; - - /** 最大尝试次数；为空时使用数据库默认值 5。 */ - private Integer maxAttempts; - - /** 下次投递时间。 */ - private LocalDateTime nextAt; - - /** 投递历史 JSON 数组。 */ - private String history; - - /** 当前投递租约令牌。 */ - private String leaseToken; - - /** 租约到期时间。 */ - private LocalDateTime leaseUntil; - - /** 最近一次失败原因。 */ - private String lastError; - - /** 创建时间。 */ - private LocalDateTime createdAt; - - /** 更新时间。 */ - private LocalDateTime updatedAt; - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java index 501b3bf99..bfd9260b9 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java @@ -1,6 +1,5 @@ package com.pig4cloud.pigx.flow.config; -import com.pig4cloud.pigx.flow.service.IFlowCallbackService; import com.pig4cloud.pigx.flow.service.IFlowNoticeService; import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; import lombok.RequiredArgsConstructor; @@ -13,6 +12,8 @@ import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import java.util.concurrent.ThreadPoolExecutor; import java.util.ArrayList; import java.util.Objects; import org.flowable.common.engine.api.delegate.event.FlowableEventListener; @@ -26,8 +27,6 @@ public class FlowDurableConfiguration { private final ObjectProvider noticeService; - private final ObjectProvider callbackService; - /** * 注册持久化监听器，保留引擎已配置的监听器。 * @return 引擎配置器 @@ -44,7 +43,7 @@ public class FlowDurableConfiguration { } /** - * 使用独立的两个调度线程，慢回调不会阻塞超时提醒，也不替换应用默认调度器。 + * 使用独立的两个调度线程，分别处理站内信投递和超时提醒，也不替换应用默认调度器。 * @return 本模块专用调度器 */ @Bean(name = "flowDeliveryScheduler", defaultCandidate = false) @@ -55,18 +54,16 @@ public class FlowDurableConfiguration { return scheduler; } - /** 每轮最多处理 20 条回调，持久记录支持重启后继续。 */ - @Scheduled(fixedDelayString = "${flow.delivery.scan-delay:10000}", scheduler = "flowDeliveryScheduler") - public void scanCallbacks() { - IFlowCallbackService service = callbackService.getObject(); - for (var row : service.listDue()) { - try { - service.deliver(row.getId()); - } - catch (Exception exception) { - log.warn("流程回调投递暂未完成，事件 {}", row.getId(), exception); - } - } + /** HTTP 节点通知独立限流；队列满时拒绝，不占用提醒线程或调用线程。 */ + @Bean(name = "flowNodeNotificationExecutor", defaultCandidate = false) + public ThreadPoolTaskExecutor flowNodeNotificationExecutor() { + var executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(2); + executor.setQueueCapacity(20); + executor.setThreadNamePrefix("flow-node-notify-"); + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); + return executor; } /** 独立扫描最多 100 条超时任务。 */ diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java deleted file mode 100644 index 9b3e22951..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java +++ /dev/null @@ -1,11 +0,0 @@ -package com.pig4cloud.pigx.flow.mapper; - -import com.baomidou.mybatisplus.core.mapper.BaseMapper; -import com.pig4cloud.pigx.flow.entity.FlowCallback; -import org.apache.ibatis.annotations.Mapper; - -/** 节点 HTTP 回调表 flow_callback 数据访问。 */ -@Mapper -public interface FlowCallbackMapper extends BaseMapper { - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java deleted file mode 100644 index e8dc9b0a3..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java +++ /dev/null @@ -1,33 +0,0 @@ -package com.pig4cloud.pigx.flow.service; - -import com.baomidou.mybatisplus.extension.service.IService; -import com.pig4cloud.pigx.flow.entity.FlowCallback; -import org.flowable.task.service.impl.persistence.entity.TaskEntity; - -import java.util.List; - -/** 节点完成 HTTP 回调的至少一次投递；租约用于多实例互斥，接收方使用事件 ID 实现幂等。 */ -public interface IFlowCallbackService extends IService { - - /** - * 在引擎事务内为已完成任务入队回调；同一任务只入队一次。 - * @param task 已完成的引擎任务 - * @param tenant 任务所属租户 - * @param target 节点配置的 HTTP(S) 回调地址 - * @throws IllegalArgumentException 回调地址不是 HTTP(S) - */ - void enqueue(TaskEntity task, long tenant, String target); - - /** - * 将最后一次尝试租约已过期的回调置为失败，并返回最多 20 条到期回调。 - * @return 到期或租约过期的回调，只包含 ID - */ - List listDue(); - - /** - * 尝试投递一条到期回调，不在网络请求期间持有数据库事务。 - * @param id 扫描得到的回调 ID - */ - void deliver(String id); - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java deleted file mode 100644 index 366004128..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java +++ /dev/null @@ -1,183 +0,0 @@ -package com.pig4cloud.pigx.flow.service.impl; - -import cn.hutool.core.convert.Convert; -import cn.hutool.core.util.StrUtil; -import cn.hutool.http.HttpRequest; -import cn.hutool.http.HttpResponse; -import cn.hutool.json.JSONArray; -import cn.hutool.json.JSONObject; -import cn.hutool.json.JSONUtil; -import com.baomidou.mybatisplus.extension.plugins.pagination.Page; -import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; -import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum; -import com.pig4cloud.pigx.flow.dto.ProcessNodeRecordAssignUserParamDto; -import com.pig4cloud.pigx.flow.entity.FlowCallback; -import com.pig4cloud.pigx.flow.mapper.FlowCallbackMapper; -import com.pig4cloud.pigx.flow.service.IFlowCallbackService; -import com.pig4cloud.pigx.flow.support.utils.NodeUtil; -import lombok.RequiredArgsConstructor; -import org.flowable.task.service.impl.persistence.entity.TaskEntity; -import org.springframework.dao.DuplicateKeyException; -import org.springframework.stereotype.Service; -import org.springframework.transaction.support.TransactionTemplate; -import java.time.Instant; -import java.time.LocalDateTime; -import java.util.List; -import java.util.Objects; -import java.util.UUID; - -/** HTTP 回调持久化与投递；租约令牌条件保证只有持有租约的实例能写回结果和历史。 */ -@Service -@RequiredArgsConstructor -public class FlowCallbackServiceImpl extends ServiceImpl - implements IFlowCallbackService { - - private final TransactionTemplate transactions; - - /** - * {@inheritDoc} - */ - @Override - public void enqueue(TaskEntity task, long tenant, String target) { - if (!NodeUtil.isCallbackUrl(target)) - throw new IllegalArgumentException("节点回调地址必须使用 HTTP(S)"); - ProcessNodeRecordAssignUserParamDto payload = new ProcessNodeRecordAssignUserParamDto(); - payload.setFlowId(NodeUtil.getFlowId(task.getProcessDefinitionId())); - payload.setProcessDefinitionId(task.getProcessDefinitionId()); - payload.setProcessInstanceId(task.getProcessInstanceId()); - payload.setNodeId(task.getTaskDefinitionKey()); - payload.setTaskId(task.getId()); - payload.setExecutionId(task.getExecutionId()); - payload.setUserId(Convert.toLong(task.getAssignee())); - payload.setNodeName(task.getName()); - payload.setTaskType("COMPLETE"); - payload.setApproveDesc(Convert.toStr(task.getVariableLocal("approveDesc"))); - payload.setData(JSONUtil.toJsonStr(task.getVariables())); - payload.setLocalData(JSONUtil.toJsonStr(task.getVariablesLocal())); - LocalDateTime now = LocalDateTime.now(); - FlowCallback callback = new FlowCallback(); - callback.setTenantId(tenant); - callback.setTaskId(task.getId()); - callback.setInstanceId(task.getProcessInstanceId()); - callback.setTargetUrl(target); - callback.setPayload(JSONUtil.toJsonStr(payload)); - callback.setStatus(FlowCallbackStatusEnum.PENDING.getCode()); - callback.setNextAt(now); - callback.setCreatedAt(now); - callback.setUpdatedAt(now); - try { - save(callback); - } - catch (DuplicateKeyException ignored) { - // 唯一键保证同一任务只入队一次；MySQL 唯一键冲突只回滚当前语句，不影响引擎事务。 - } - } - - /** - * {@inheritDoc} - */ - @Override - public List listDue() { - LocalDateTime now = LocalDateTime.now(); - // 最后一次尝试进程崩溃时，结果未知，保留审计并置失败供人工核对。 - lambdaUpdate().set(FlowCallback::getStatus, FlowCallbackStatusEnum.FAILED.getCode()) - .set(FlowCallback::getLastError, "租约过期，投递结果未知") - .set(FlowCallback::getLeaseToken, null) - .set(FlowCallback::getLeaseUntil, null) - .set(FlowCallback::getUpdatedAt, now) - .eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode()) - .lt(FlowCallback::getLeaseUntil, now) - .apply("attempts >= max_attempts") - .update(); - return lambdaQuery().select(FlowCallback::getId) - .and(w -> w - .nested(due -> due.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.PENDING.getCode()) - .le(FlowCallback::getNextAt, now)) - .or(expired -> expired.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode()) - .lt(FlowCallback::getLeaseUntil, now))) - .orderByAsc(FlowCallback::getNextAt) - .page(new Page<>(1, 20, false)) - .getRecords(); - } - - /** - * {@inheritDoc} - */ - @Override - public void deliver(String id) { - String lease = UUID.randomUUID().toString(); - FlowCallback record = transactions.execute(tx -> { - LocalDateTime now = LocalDateTime.now(); - boolean claimed = lambdaUpdate().set(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode()) - .set(FlowCallback::getLeaseToken, lease) - .set(FlowCallback::getLeaseUntil, now.plusSeconds(60)) - .set(FlowCallback::getUpdatedAt, now) - .setSql("attempts = attempts + 1") - .eq(FlowCallback::getId, id) - .and(w -> w - .nested(due -> due.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.PENDING.getCode()) - .le(FlowCallback::getNextAt, now)) - .or(expired -> expired.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode()) - .lt(FlowCallback::getLeaseUntil, now))) - .apply("attempts < max_attempts") - .update(); - if (!claimed) - return null; - FlowCallback row = lambdaQuery().eq(FlowCallback::getId, id).eq(FlowCallback::getLeaseToken, lease).one(); - // 历史不截断，保留每次领取租约的审计记录。 - JSONArray history = StrUtil.isBlank(row.getHistory()) ? new JSONArray() - : JSONUtil.parseArray(row.getHistory()); - history.add(new JSONObject().set("id", lease) - .set("attempt_no", row.getAttempts()) - .set("result", "STARTED") - .set("created_at", Instant.now().toString())); - row.setHistory(history.toString()); - lambdaUpdate().set(FlowCallback::getHistory, row.getHistory()) - .eq(FlowCallback::getId, id) - .eq(FlowCallback::getLeaseToken, lease) - .update(); - return row; - }); - if (Objects.isNull(record)) - return; - boolean success = false; - String detail; - try (HttpResponse response = HttpRequest.post(record.getTargetUrl()) - .header("X-Flow-Event-Id", id) - .header("Idempotency-Key", id) - .body(record.getPayload()) - .timeout(10000) - .setFollowRedirects(false) - .execute()) { - success = response.isOk(); - detail = "HTTP " + response.getStatus(); - } - catch (Exception exception) { - // 只记录异常类型，避免响应正文、URL 参数或表单数据进入日志。 - detail = exception.getClass().getSimpleName(); - } - int attempts = record.getAttempts(); - FlowCallbackStatusEnum state = success ? FlowCallbackStatusEnum.DELIVERED - : attempts >= record.getMaxAttempts() ? FlowCallbackStatusEnum.FAILED : FlowCallbackStatusEnum.PENDING; - JSONArray history = JSONUtil.parseArray(record.getHistory()); - JSONObject attempt = history.stream() - .map(value -> (JSONObject) value) - .filter(value -> lease.equals(value.getStr("id"))) - .findFirst() - .orElseThrow(() -> new IllegalStateException("回调尝试记录缺失")); - attempt.set("result", success ? "DELIVERED" : "ERROR"); - attempt.set("detail", detail); - LocalDateTime now = LocalDateTime.now(); - lambdaUpdate().set(FlowCallback::getStatus, state.getCode()) - .set(FlowCallback::getNextAt, now.plusSeconds(Math.min(3600, 30L * (1L << Math.min(attempts - 1, 7))))) - .set(FlowCallback::getLastError, success ? null : StrUtil.maxLength(detail, 500)) - .set(FlowCallback::getLeaseToken, null) - .set(FlowCallback::getLeaseUntil, null) - .set(FlowCallback::getHistory, history.toString()) - .set(FlowCallback::getUpdatedAt, now) - .eq(FlowCallback::getId, id) - .eq(FlowCallback::getLeaseToken, lease) - .update(); - } - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java index 585923e5e..f7e43c308 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java @@ -18,7 +18,7 @@ import com.pig4cloud.pigx.flow.dto.Node; import com.pig4cloud.pigx.flow.entity.FlowNotice; import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord; import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper; -import com.pig4cloud.pigx.flow.service.IFlowCallbackService; +import com.pig4cloud.pigx.flow.support.listeners.NodeTaskCompleteNotify; import com.pig4cloud.pigx.flow.service.IFlowNoticeService; import com.pig4cloud.pigx.flow.service.IProcessInstanceRecordService; import com.pig4cloud.pigx.flow.service.IProcessNodeDataService; @@ -39,7 +39,7 @@ import java.util.UUID; @RequiredArgsConstructor public class FlowNoticeServiceImpl extends ServiceImpl implements IFlowNoticeService { - private final IFlowCallbackService callbackService; + private final NodeTaskCompleteNotify nodeTaskCompleteNotify; private final IProcessInstanceRecordService instanceRecordService; @@ -69,7 +69,7 @@ public class FlowNoticeServiceImpl extends ServiceImpl { + try { + notificationExecutor.execute(() -> sendTaskCompleted(target, body, eventId, taskId, instanceId)); + } + catch (Exception exception) { + log.warn("节点完成通知未提交，事件 {}，任务 {}，实例 {}，原因 {}", eventId, taskId, instanceId, + exception.getClass().getSimpleName()); + } + }; + if (TransactionSynchronizationManager.isSynchronizationActive()) { + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + submit.run(); + } + }); + } + else { + submit.run(); + } + } + + private void sendTaskCompleted(String target, String body, String eventId, String taskId, String instanceId) { + try (HttpResponse response = HttpRequest.post(target) + .header("X-Flow-Event-Id", eventId) + .header("Idempotency-Key", eventId) + .body(body) + .timeout(10000) + .setFollowRedirects(false) + .execute()) { + if (!response.isOk()) + log.warn("节点完成通知失败，事件 {}，任务 {}，实例 {}，HTTP {}", eventId, taskId, instanceId, response.getStatus()); + } + catch (Exception exception) { + log.warn("节点完成通知失败，事件 {}，任务 {}，实例 {}，原因 {}", eventId, taskId, instanceId, + exception.getClass().getSimpleName()); + } + } + /** * 异步发送流程实例状态变更通知 *

diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotifyTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotifyTests.java new file mode 100644 index 000000000..e3b04f178 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotifyTests.java @@ -0,0 +1,137 @@ +package com.pig4cloud.pigx.flow.support.listeners; + +import com.pig4cloud.pigx.flow.event.FlowStatusEventPublisher; +import com.pig4cloud.pigx.flow.config.FlowDurableConfiguration; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import org.slf4j.LoggerFactory; +import java.util.concurrent.CountDownLatch; +import com.pig4cloud.pigx.flow.mapper.ProcessInstanceRecordMapper; +import com.sun.net.httpserver.HttpServer; +import org.flowable.task.service.impl.persistence.entity.TaskEntity; +import org.junit.jupiter.api.Test; +import org.springframework.transaction.support.TransactionSynchronizationManager; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class NodeTaskCompleteNotifyTests { + + @Test + void sendsOnlyAfterCommit() throws Exception { + verifyDelivery(200); + } + + @Test + void failedRequestDoesNotEscapeOrRetry() throws Exception { + verifyDelivery(500); + } + + @Test + void saturatedExecutorDropsNotificationWithoutBlockingOrRetry() throws Exception { + var executor = new FlowDurableConfiguration(null).flowNodeNotificationExecutor(); + executor.initialize(); + var release = new CountDownLatch(1); + var started = new CountDownLatch(2); + var logs = new ListAppender(); + logs.start(); + Logger logger = (Logger) LoggerFactory.getLogger(NodeTaskCompleteNotify.class); + logger.addAppender(logs); + try { + for (int i = 0; i < 2; i++) { + executor.execute(() -> { + started.countDown(); + try { + release.await(); + } + catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + } + }); + } + assertThat(started.await(5, TimeUnit.SECONDS)).isTrue(); + for (int i = 0; i < 20; i++) + executor.execute(() -> { }); + TaskEntity task = mock(TaskEntity.class); + when(task.getId()).thenReturn("rejected-task"); + when(task.getProcessInstanceId()).thenReturn("rejected-instance"); + NodeTaskCompleteNotify notifier = new NodeTaskCompleteNotify(mock(ProcessInstanceRecordMapper.class), + mock(FlowStatusEventPublisher.class), executor); + assertThatCode(() -> notifier.notifyTaskCompleted(task, "http://127.0.0.1:1/rejected")) + .doesNotThrowAnyException(); + assertThat(executor.getPoolSize()).isEqualTo(2); + assertThat(executor.getThreadPoolExecutor().getQueue()).hasSize(20); + assertThat(logs.list).hasSize(1); + assertThat(logs.list.get(0).getFormattedMessage()) + .contains("节点完成通知未提交", "rejected-task", "rejected-instance", "TaskRejectedException") + .doesNotContain("http://"); + release.countDown(); + executor.getThreadPoolExecutor().shutdown(); + assertThat(executor.getThreadPoolExecutor().awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(executor.getThreadPoolExecutor().getCompletedTaskCount()).isEqualTo(22); + } + finally { + release.countDown(); + executor.shutdown(); + logger.detachAppender(logs); + logs.stop(); + } + } + + private void verifyDelivery(int status) throws Exception { + var requests = new LinkedBlockingQueue(); + var executor = new FlowDurableConfiguration(null).flowNodeNotificationExecutor(); + executor.initialize(); + var logs = new ListAppender(); + logs.start(); + Logger logger = (Logger) LoggerFactory.getLogger(NodeTaskCompleteNotify.class); + logger.addAppender(logs); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/callback", exchange -> { + try (exchange) { + String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + exchange.sendResponseHeaders(status, -1); + requests.add(body); + } + }); + server.start(); + TransactionSynchronizationManager.initSynchronization(); + try { + TaskEntity task = mock(TaskEntity.class); + when(task.getProcessDefinitionId()).thenReturn("flow1:1:definition1"); + when(task.getId()).thenReturn("task1"); + when(task.getProcessInstanceId()).thenReturn("instance1"); + NodeTaskCompleteNotify notifier = new NodeTaskCompleteNotify(mock(ProcessInstanceRecordMapper.class), + mock(FlowStatusEventPublisher.class), executor); + notifier.notifyTaskCompleted(task, "http://127.0.0.1:" + server.getAddress().getPort() + "/callback"); + assertThat(requests.poll(200, TimeUnit.MILLISECONDS)).isNull(); + var synchronizations = TransactionSynchronizationManager.getSynchronizations(); + assertThatCode(() -> synchronizations.forEach(sync -> sync.afterCommit())).doesNotThrowAnyException(); + assertThat(requests.poll(5, TimeUnit.SECONDS)).contains("\"taskId\":\"task1\"", "\"taskType\":\"COMPLETE\""); + assertThat(requests.poll(200, TimeUnit.MILLISECONDS)).isNull(); + executor.getThreadPoolExecutor().shutdown(); + assertThat(executor.getThreadPoolExecutor().awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + if (status == 500) { + assertThat(logs.list).hasSize(1); + assertThat(logs.list.get(0).getFormattedMessage()) + .contains("任务 task1", "实例 instance1", "HTTP 500").doesNotContain("http://", "taskType"); + } + } + finally { + TransactionSynchronizationManager.clearSynchronization(); + server.stop(0); + executor.shutdown(); + logger.detachAppender(logs); + logs.stop(); + } + } + +}