fix(flow): 移除回调持久化重试并限制单次通知并发

(cherry picked from commit f7cd6724ee5e5a2fbb7a677f386e582ba9d92610)

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
 -- ----------------------------

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;

pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java

@@ -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;
-
-}

pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java

@@ -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;
-
-}

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<IFlowNoticeService> noticeService;
 
-	private final ObjectProvider<IFlowCallbackService> 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 条超时任务。 */

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java

@@ -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<FlowCallback> {
-
-}

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java

@@ -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<FlowCallback> {
-
-	/**
-	 * 在引擎事务内为已完成任务入队回调；同一任务只入队一次。
-	 * @param task 已完成的引擎任务
-	 * @param tenant 任务所属租户
-	 * @param target 节点配置的 HTTP(S) 回调地址
-	 * @throws IllegalArgumentException 回调地址不是 HTTP(S)
-	 */
-	void enqueue(TaskEntity task, long tenant, String target);
-
-	/**
-	 * 将最后一次尝试租约已过期的回调置为失败，并返回最多 20 条到期回调。
-	 * @return 到期或租约过期的回调，只包含 ID
-	 */
-	List<FlowCallback> listDue();
-
-	/**
-	 * 尝试投递一条到期回调，不在网络请求期间持有数据库事务。
-	 * @param id 扫描得到的回调 ID
-	 */
-	void deliver(String id);
-
-}

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java

@@ -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<FlowCallbackMapper, FlowCallback>
-		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<FlowCallback> 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();
-	}
-
-}

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<FlowNoticeMapper, FlowNotice> implements IFlowNoticeService {
 
-	private final IFlowCallbackService callbackService;
+	private final NodeTaskCompleteNotify nodeTaskCompleteNotify;
 
 	private final IProcessInstanceRecordService instanceRecordService;
 
@@ -69,7 +69,7 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 		if (FlowableEngineEventType.TASK_COMPLETED.name().equals(event)) {
 			cancelReminder(tenant, task.getId());
 			if (StrUtil.isNotBlank(node.getEventConfig()))
-				callbackService.enqueue(task, tenant, node.getEventConfig());
+				nodeTaskCompleteNotify.notifyTaskCompleted(task, node.getEventConfig());
 			return;
 		}
 		if (Objects.equals(node.getType(), NodeTypeEnum.ROOT.getValue())

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java

@@ -1,28 +1,107 @@
 package com.pig4cloud.pigx.flow.support.listeners;
 
+import cn.hutool.core.convert.Convert;
+import cn.hutool.http.HttpRequest;
+import cn.hutool.http.HttpResponse;
+import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.pig4cloud.pigx.common.security.util.NonWebTokenContextHolder;
 import com.pig4cloud.pigx.flow.dto.ProcessInstanceParamDto;
+import com.pig4cloud.pigx.flow.dto.ProcessNodeRecordAssignUserParamDto;
 import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord;
 import com.pig4cloud.pigx.flow.event.FlowStatusEventPublisher;
 import com.pig4cloud.pigx.flow.mapper.ProcessInstanceRecordMapper;
-import lombok.RequiredArgsConstructor;
+import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.stereotype.Service;
+import java.util.UUID;
+import java.util.concurrent.Executor;
+import org.springframework.beans.factory.annotation.Qualifier;
 
 /**
- * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， IFlowCallbackService 在提交后投递；此类保留现有
- * Redis 状态发布行为。
+ * 流程实例状态通知及节点 HTTP 通知；HTTP 通知在事务提交后发送一次，失败不重试。
  *
  * @author lengleng
  */
 @Service
-@RequiredArgsConstructor
+@Slf4j
 public class NodeTaskCompleteNotify {
 
 	private final ProcessInstanceRecordMapper processInstanceRecordMapper;
 
 	private final FlowStatusEventPublisher flowStatusEventPublisher;
 
+	private final Executor notificationExecutor;
+
+	public NodeTaskCompleteNotify(ProcessInstanceRecordMapper processInstanceRecordMapper,
+			FlowStatusEventPublisher flowStatusEventPublisher,
+			@Qualifier("flowNodeNotificationExecutor") Executor notificationExecutor) {
+		this.processInstanceRecordMapper = processInstanceRecordMapper;
+		this.flowStatusEventPublisher = flowStatusEventPublisher;
+		this.notificationExecutor = notificationExecutor;
+	}
+
+	/** 节点完成后发送一次 HTTP 通知，先快照任务数据，事务回滚时不发送。 */
+	public void notifyTaskCompleted(TaskEntity task, String target) {
+		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()));
+		String body = JSONUtil.toJsonStr(payload);
+		String eventId = UUID.randomUUID().toString();
+		String taskId = task.getId();
+		String instanceId = task.getProcessInstanceId();
+		Runnable submit = () -> {
+			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());
+		}
+	}
+
 	/**
 	 * 异步发送流程实例状态变更通知
 	 * <p>

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<ILoggingEvent>();
+		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<String>();
+		var executor = new FlowDurableConfiguration(null).flowNodeNotificationExecutor();
+		executor.initialize();
+		var logs = new ListAppender<ILoggingEvent>();
+		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();
+		}
+	}
+
+}