From 4dac27393ebcd4699e716795e94b9cc5f172cb67 Mon Sep 17 00:00:00 2001 From: dzc <2388224167@qq.com> Date: Mon, 10 Aug 2026 10:21:17 +0800 Subject: [PATCH] =?UTF-8?q?fix:=E4=BC=98=E5=8C=96=E6=89=AB=E4=B8=80?= =?UTF-8?q?=E6=89=AB=E6=94=AF=E4=BB=98=E8=BD=AE=E8=AE=AD=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../job/ThirdAggregateOrderPayJob.java | 122 +++++---- .../thirdparty/job/ThirdAppPlusPayJob.java | 80 +++--- .../service/ThirdPayPollingCoordinator.java | 243 ++++++++++++++++++ .../ThirdAggregateOrderPayServiceImpl.java | 9 +- .../impl/appplus/ThirdAppPlusServiceImpl.java | 9 +- .../job/ThirdAppPlusPayJobTest.java | 40 +++ .../ThirdPayPollingCoordinatorTest.java | 48 ++++ ...ThirdAggregateOrderPayServiceImplTest.java | 10 +- ...irdAppPlusServiceImplOrderCodePayTest.java | 7 + 9 files changed, 479 insertions(+), 89 deletions(-) create mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinator.java create mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinatorTest.java diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java index 047c17608..f144293c0 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java @@ -7,6 +7,7 @@ import com.bsd.pjcloud.thirdparty.constant.pay.ThirdAggregatePaySessionStatusEnu import com.bsd.pjcloud.thirdparty.entity.pay.ThirdAggregatePaySessionEntity; import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; +import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; @@ -28,16 +29,6 @@ public class ThirdAggregateOrderPayJob { */ private static final int BATCH_SIZE = 100; - /** - * 未到期聚合会话的服务端查单间隔，单位毫秒。 - */ - private static final long PENDING_POLL_INTERVAL_MILLIS = 5000L; - - /** - * 到期聚合子订单的服务端关单补偿间隔，单位毫秒，避免继续占用五秒高频任务。 - */ - private static final long CLOSE_COMPENSATION_INTERVAL_MILLIS = 30000L; - /** * 到期后的关单补偿窗口，单位分钟；超过该窗口不再后台调用第三方支付平台。 */ @@ -54,68 +45,95 @@ public class ThirdAggregateOrderPayJob { private final ThirdAggregateOrderPayService aggregateOrderPayService; /** - * 每五秒分批处理等待扫码、平台创建中和支付中的聚合会话。 + * 服务端轮询协调器，用于在无活动会话时退避数据库扫描，并在新会话创建后恢复短间隔。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + + /** + * 分批处理等待扫码、平台创建中和支付中的聚合会话。 * *

调度线程初始没有租户上下文，单条处理前使用记录自身 tenantId 恢复上下文； - * Redis 键仍由公共基础设施自动附加租户前缀。

+ * Redis 键仍由公共基础设施自动附加租户前缀。方法每五秒轻量触发一次，但连续空结果时协调器会把 + * 真正的数据库扫描逐级退避到最长六十秒。

*/ - @Scheduled(fixedDelay = PENDING_POLL_INTERVAL_MILLIS, initialDelay = PENDING_POLL_INTERVAL_MILLIS) + @Scheduled(fixedDelay = ThirdPayPollingCoordinator.ACTIVE_TRIGGER_INTERVAL_MILLIS, + initialDelay = ThirdPayPollingCoordinator.ACTIVE_TRIGGER_INTERVAL_MILLIS) public void handlePendingSessions() { - List sessions = aggregatePaySessionService.list( - new LambdaQueryWrapper() - .in(ThirdAggregatePaySessionEntity::getStatus, - ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), - ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), - ThirdAggregatePaySessionStatusEnum.PAYING.getType()) - .orderByAsc(ThirdAggregatePaySessionEntity::getCreateTime) - .last("limit " + BATCH_SIZE)); - if (ObjectUtil.isEmpty(sessions)) { + if (!pollingCoordinator.tryStartAggregatePendingPolling()) { return; } - for (ThirdAggregatePaySessionEntity session : sessions) { - if (session.getTenantId() == null) { - log.warn("聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); - continue; + boolean hasPendingSessions = true; + try { + List sessions = aggregatePaySessionService.list( + new LambdaQueryWrapper() + .in(ThirdAggregatePaySessionEntity::getStatus, + ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), + ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), + ThirdAggregatePaySessionStatusEnum.PAYING.getType()) + .orderByAsc(ThirdAggregatePaySessionEntity::getCreateTime) + .last("limit " + BATCH_SIZE)); + hasPendingSessions = ObjectUtil.isNotEmpty(sessions); + if (!hasPendingSessions) { + return; } - try { - TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); - } catch (Exception exception) { - log.error("处理聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); + for (ThirdAggregatePaySessionEntity session : sessions) { + if (session.getTenantId() == null) { + log.warn("聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); + continue; + } + try { + TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); + } catch (Exception exception) { + log.error("处理聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); + } } + } finally { + pollingCoordinator.finishAggregatePendingPolling(hasPendingSessions); } } /** - * 每三十秒补偿处理已到期但子订单尚未确认关闭的聚合会话。 + * 补偿处理已到期但子订单尚未确认关闭的聚合会话。 * *

补偿窗口从会话原始失效时间开始计算，最多持续十分钟；因此客户端退出、断网或不再查询状态时， - * 服务端仍会独立完成有限次查单和关单，同时不会对一天前的未支付订单永久轮询。

+ * 服务端仍会独立完成有限次查单和关单。连续空结果时数据库扫描从三十秒逐级退避到最长五分钟， + * 同时不会对一天前的未支付订单永久轮询。

*/ - @Scheduled(fixedDelay = CLOSE_COMPENSATION_INTERVAL_MILLIS, initialDelay = CLOSE_COMPENSATION_INTERVAL_MILLIS) + @Scheduled(fixedDelay = ThirdPayPollingCoordinator.COMPENSATION_TRIGGER_INTERVAL_MILLIS, + initialDelay = ThirdPayPollingCoordinator.COMPENSATION_TRIGGER_INTERVAL_MILLIS) public void handleExpiredSessionCompensation() { - LocalDateTime now = LocalDateTime.now(); - List sessions = aggregatePaySessionService.list( - new LambdaQueryWrapper() - .eq(ThirdAggregatePaySessionEntity::getStatus, ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()) - .isNotNull(ThirdAggregatePaySessionEntity::getPayNo) - .isNotNull(ThirdAggregatePaySessionEntity::getExpireTime) - .between(ThirdAggregatePaySessionEntity::getExpireTime, - now.minusMinutes(CLOSE_COMPENSATION_MINUTES), now) - .orderByDesc(ThirdAggregatePaySessionEntity::getExpireTime) - .last("limit " + BATCH_SIZE)); - if (ObjectUtil.isEmpty(sessions)) { + if (!pollingCoordinator.tryStartAggregateCompensationPolling()) { return; } - for (ThirdAggregatePaySessionEntity session : sessions) { - if (session.getTenantId() == null) { - log.warn("到期聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); - continue; + boolean hasExpiredSessions = true; + try { + LocalDateTime now = LocalDateTime.now(); + List sessions = aggregatePaySessionService.list( + new LambdaQueryWrapper() + .eq(ThirdAggregatePaySessionEntity::getStatus, ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()) + .isNotNull(ThirdAggregatePaySessionEntity::getPayNo) + .isNotNull(ThirdAggregatePaySessionEntity::getExpireTime) + .between(ThirdAggregatePaySessionEntity::getExpireTime, + now.minusMinutes(CLOSE_COMPENSATION_MINUTES), now) + .orderByDesc(ThirdAggregatePaySessionEntity::getExpireTime) + .last("limit " + BATCH_SIZE)); + hasExpiredSessions = ObjectUtil.isNotEmpty(sessions); + if (!hasExpiredSessions) { + return; } - try { - TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); - } catch (Exception exception) { - log.error("补偿关闭聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); + for (ThirdAggregatePaySessionEntity session : sessions) { + if (session.getTenantId() == null) { + log.warn("到期聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); + continue; + } + try { + TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); + } catch (Exception exception) { + log.error("补偿关闭聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); + } } + } finally { + pollingCoordinator.finishAggregateCompensationPolling(hasExpiredSessions); } } } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java index 43e9b2dbe..887637176 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java @@ -11,6 +11,7 @@ import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusService; import com.bsd.pjcloud.thirdparty.service.ThirdPayOrderService; +import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; @@ -32,11 +33,6 @@ public class ThirdAppPlusPayJob { */ private static final int ORDER_CODE_BATCH_SIZE = 100; - /** - * 原生订单码的服务端查单间隔，单位毫秒。 - */ - private static final long ORDER_CODE_POLL_INTERVAL_MILLIS = 5000L; - /** * 原生订单码到期后的关单补偿窗口，单位分钟；超过窗口后停止后台查询。 */ @@ -52,6 +48,11 @@ public class ThirdAppPlusPayJob { */ private final ThirdAppPlusService thirdAppPlusService; + /** + * 服务端轮询协调器，用于在无待处理订单时退避数据库扫描，并在创建订单后恢复短间隔。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + /** * 处理,未勾起业务回调的订单 */ @@ -76,43 +77,54 @@ public class ThirdAppPlusPayJob { } /** - * 每五秒分批处理支付宝和微信云消费机订单码，最长处理到订单过期后十分钟。 + * 分批处理支付宝和微信云消费机订单码，最长处理到订单过期后十分钟。 * *

初始查询在线程无租户上下文时跨租户读取，处理单条订单前通过 TenantBroker - * 恢复订单所属租户，确保后续读取和原子更新都受正确租户条件约束。

+ * 恢复订单所属租户，确保后续读取和原子更新都受正确租户条件约束。方法每五秒轻量触发一次， + * 但连续空结果时协调器会把真正的数据库扫描逐级退避到最长六十秒。

*/ - @Scheduled(fixedDelay = ORDER_CODE_POLL_INTERVAL_MILLIS, initialDelay = ORDER_CODE_POLL_INTERVAL_MILLIS) + @Scheduled(fixedDelay = ThirdPayPollingCoordinator.ACTIVE_TRIGGER_INTERVAL_MILLIS, + initialDelay = ThirdPayPollingCoordinator.ACTIVE_TRIGGER_INTERVAL_MILLIS) public void handleOrderCodeOrders() { - LocalDateTime earliestExpireTime = LocalDateTime.now().minusMinutes(ORDER_CODE_CLOSE_COMPENSATION_MINUTES); - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .in(ThirdPayOrderEntity::getPayPlatformType, - AppPlusPlatformTypeEnum.ALI_PAY.getType(), AppPlusPlatformTypeEnum.WECHAT.getType()) - .eq(ThirdPayOrderEntity::getBizType, ThirdPayOrderBizTypeEnum.YXFJ_PAY.getType()) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) - .isNotNull(ThirdPayOrderEntity::getExpireTime) - .ge(ThirdPayOrderEntity::getExpireTime, earliestExpireTime) - .notExists("SELECT 1 FROM third_aggregate_pay_session s WHERE s.pay_no = third_pay_order.pay_no AND s.del_flag = '0'") - .orderByAsc(ThirdPayOrderEntity::getCreateTime) - .last("limit " + ORDER_CODE_BATCH_SIZE)); - if (ObjectUtil.isEmpty(orders)) { + if (!pollingCoordinator.tryStartOrderCodePolling()) { return; } - for (ThirdPayOrderEntity order : orders) { - if (order.getTenantId() == null) { - log.warn("订单码缺少租户ID,orderId:{},platform:{}", order.getId(), order.getPayPlatformType()); - continue; + boolean hasPendingOrders = true; + try { + LocalDateTime earliestExpireTime = LocalDateTime.now().minusMinutes(ORDER_CODE_CLOSE_COMPENSATION_MINUTES); + List orders = thirdPayOrderService.list(new LambdaQueryWrapper() + .in(ThirdPayOrderEntity::getPayPlatformType, + AppPlusPlatformTypeEnum.ALI_PAY.getType(), AppPlusPlatformTypeEnum.WECHAT.getType()) + .eq(ThirdPayOrderEntity::getBizType, ThirdPayOrderBizTypeEnum.YXFJ_PAY.getType()) + .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) + .isNotNull(ThirdPayOrderEntity::getExpireTime) + .ge(ThirdPayOrderEntity::getExpireTime, earliestExpireTime) + .notExists("SELECT 1 FROM third_aggregate_pay_session s WHERE s.pay_no = third_pay_order.pay_no AND s.del_flag = '0'") + .orderByAsc(ThirdPayOrderEntity::getCreateTime) + .last("limit " + ORDER_CODE_BATCH_SIZE)); + hasPendingOrders = ObjectUtil.isNotEmpty(orders); + if (!hasPendingOrders) { + return; } - try { - TenantBroker.runAs(order.getTenantId(), tenantId -> { - ThirdPayOrderEntity latestOrder = thirdPayOrderService.getById(order.getId()); - if (latestOrder != null && ObjectUtil.equals(latestOrder.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { - thirdAppPlusService.queryPayResult(latestOrder); - } - }); - } catch (Exception exception) { - log.error("处理订单码状态异常,orderId:{},tenantId:{},platform:{}", - order.getId(), order.getTenantId(), order.getPayPlatformType(), exception); + for (ThirdPayOrderEntity order : orders) { + if (order.getTenantId() == null) { + log.warn("订单码缺少租户ID,orderId:{},platform:{}", order.getId(), order.getPayPlatformType()); + continue; + } + try { + TenantBroker.runAs(order.getTenantId(), tenantId -> { + ThirdPayOrderEntity latestOrder = thirdPayOrderService.getById(order.getId()); + if (latestOrder != null && ObjectUtil.equals(latestOrder.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { + thirdAppPlusService.queryPayResult(latestOrder); + } + }); + } catch (Exception exception) { + log.error("处理订单码状态异常,orderId:{},tenantId:{},platform:{}", + order.getId(), order.getTenantId(), order.getPayPlatformType(), exception); + } } + } finally { + pollingCoordinator.finishOrderCodePolling(hasPendingOrders); } } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinator.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinator.java new file mode 100644 index 000000000..767497dd9 --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinator.java @@ -0,0 +1,243 @@ +package com.bsd.pjcloud.thirdparty.service; + +import org.springframework.stereotype.Component; + +import java.util.concurrent.TimeUnit; + +/** + * 第三方订单码服务端轮询协调器。 + * + *

支付平台回调仍是结果确认主路径；本协调器只控制数据库补偿扫描频率。有待处理记录时维持短间隔， + * 连续空扫描时指数退避，订单创建成功后由当前服务实例主动唤醒，从而避免全天执行无效 SQL。

+ */ +@Component +public class ThirdPayPollingCoordinator { + + /** + * 活动订单任务的轻量触发周期，单位毫秒；真正的数据库扫描还会受空闲退避门控制。 + */ + public static final long ACTIVE_TRIGGER_INTERVAL_MILLIS = 5000L; + + /** + * 过期补偿任务的轻量触发周期，单位毫秒；真正的数据库扫描还会受空闲退避门控制。 + */ + public static final long COMPENSATION_TRIGGER_INTERVAL_MILLIS = 30000L; + + /** + * 原生订单码和聚合活动会话连续为空时的最大数据库扫描间隔，单位毫秒。 + */ + private static final long ACTIVE_MAX_IDLE_INTERVAL_MILLIS = 60000L; + + /** + * 聚合过期补偿连续为空时的最大数据库扫描间隔，单位毫秒。 + */ + private static final long COMPENSATION_MAX_IDLE_INTERVAL_MILLIS = 300000L; + + /** + * 指数退避最大左移位数，防止连续空扫描计数和间隔计算无限增长。 + */ + private static final int MAX_BACKOFF_SHIFT = 8; + + /** + * 旧支付宝、微信原生订单码扫描门，状态仅属于当前服务实例进程。 + */ + private final AdaptivePollingGate orderCodeGate; + + /** + * 支付宝、微信聚合活动会话扫描门，状态仅属于当前服务实例进程。 + */ + private final AdaptivePollingGate aggregatePendingGate; + + /** + * 聚合会话到期关单补偿扫描门，状态仅属于当前服务实例进程。 + */ + private final AdaptivePollingGate aggregateCompensationGate; + + /** + * 创建轮询协调器并让三个扫描门在服务启动后各执行一次恢复扫描。 + */ + public ThirdPayPollingCoordinator() { + orderCodeGate = new AdaptivePollingGate(ACTIVE_TRIGGER_INTERVAL_MILLIS, ACTIVE_MAX_IDLE_INTERVAL_MILLIS); + aggregatePendingGate = new AdaptivePollingGate(ACTIVE_TRIGGER_INTERVAL_MILLIS, ACTIVE_MAX_IDLE_INTERVAL_MILLIS); + aggregateCompensationGate = new AdaptivePollingGate( + COMPENSATION_TRIGGER_INTERVAL_MILLIS, COMPENSATION_MAX_IDLE_INTERVAL_MILLIS); + } + + /** + * 尝试取得本轮原生订单码数据库扫描资格。 + * + * @return 到达下次扫描时间且当前没有同类扫描执行时返回 {@code true} + */ + public boolean tryStartOrderCodePolling() { + return orderCodeGate.tryStart(); + } + + /** + * 完成本轮原生订单码扫描并根据是否存在待处理订单安排下次数据库扫描。 + * + * @param hasPendingOrders 本轮查询是否返回待处理订单；查询异常时应传 {@code true} 以便尽快重试 + */ + public void finishOrderCodePolling(boolean hasPendingOrders) { + orderCodeGate.finish(hasPendingOrders); + } + + /** + * 原生订单码创建或复用成功后唤醒当前实例扫描门，最迟在下一个五秒触发周期执行数据库扫描。 + */ + public void wakeOrderCodePolling() { + orderCodeGate.wakeUp(); + } + + /** + * 尝试取得本轮聚合活动会话数据库扫描资格。 + * + * @return 到达下次扫描时间且当前没有同类扫描执行时返回 {@code true} + */ + public boolean tryStartAggregatePendingPolling() { + return aggregatePendingGate.tryStart(); + } + + /** + * 完成本轮聚合活动会话扫描并根据是否存在待处理会话安排下次数据库扫描。 + * + * @param hasPendingSessions 本轮查询是否返回活动会话；查询异常时应传 {@code true} 以便尽快重试 + */ + public void finishAggregatePendingPolling(boolean hasPendingSessions) { + aggregatePendingGate.finish(hasPendingSessions); + } + + /** + * 聚合会话创建或复用成功后同时唤醒活动扫描与到期补偿扫描。 + * + *

补偿扫描同步唤醒可避免它正处于五分钟空闲间隔时错过新会话的两分钟失效节点。

+ */ + public void wakeAggregatePolling() { + aggregatePendingGate.wakeUp(); + aggregateCompensationGate.wakeUp(); + } + + /** + * 尝试取得本轮聚合过期补偿数据库扫描资格。 + * + * @return 到达下次扫描时间且当前没有同类扫描执行时返回 {@code true} + */ + public boolean tryStartAggregateCompensationPolling() { + return aggregateCompensationGate.tryStart(); + } + + /** + * 完成本轮聚合过期补偿扫描并根据是否存在待补偿会话安排下次数据库扫描。 + * + * @param hasExpiredSessions 本轮查询是否返回待补偿会话；查询异常时应传 {@code true} 以便尽快重试 + */ + public void finishAggregateCompensationPolling(boolean hasExpiredSessions) { + aggregateCompensationGate.finish(hasExpiredSessions); + } + + /** + * 单类轮询任务的进程内自适应扫描门。 + * + *

扫描门不保存业务订单信息，也不替代数据库恢复；服务重启后第一次触发始终允许扫描。

+ */ + private static final class AdaptivePollingGate { + + /** + * 发现待处理记录或扫描异常时使用的基础间隔，单位毫秒。 + */ + private final long activeIntervalMillis; + + /** + * 连续空扫描允许退避到的最大间隔，单位毫秒。 + */ + private final long maxIdleIntervalMillis; + + /** + * 连续空扫描次数，用于计算指数退避倍数，唤醒或发现记录后清零。 + */ + private int consecutiveEmptyScans; + + /** + * 使用 {@link System#nanoTime()} 时间基准保存的下次允许扫描时刻。 + */ + private long nextScanAtNanos; + + /** + * 当前是否已有线程取得扫描资格，防止调度重叠执行同一类数据库扫描。 + */ + private boolean scanInProgress; + + /** + * 扫描执行期间是否收到订单创建唤醒，确保完成扫描时不会覆盖这次唤醒。 + */ + private boolean wakeUpRequested; + + /** + * 创建自适应扫描门，初始下次扫描时刻设为当前时间以支持服务启动恢复。 + * + * @param activeIntervalMillis 活动或异常状态的基础扫描间隔，单位毫秒 + * @param maxIdleIntervalMillis 连续空闲状态的最大扫描间隔，单位毫秒 + */ + private AdaptivePollingGate(long activeIntervalMillis, long maxIdleIntervalMillis) { + this.activeIntervalMillis = activeIntervalMillis; + this.maxIdleIntervalMillis = maxIdleIntervalMillis; + this.nextScanAtNanos = System.nanoTime(); + } + + /** + * 原子判断当前是否到达扫描时间，并锁定本轮扫描资格。 + * + * @return 当前线程取得扫描资格时返回 {@code true} + */ + private synchronized boolean tryStart() { + long now = System.nanoTime(); + if (scanInProgress || now - nextScanAtNanos < 0) { + return false; + } + scanInProgress = true; + return true; + } + + /** + * 释放扫描资格并根据查询结果设置下次允许扫描的时间。 + * + * @param hasPendingRecords 本轮是否存在待处理记录，异常场景按 {@code true} 处理 + */ + private synchronized void finish(boolean hasPendingRecords) { + scanInProgress = false; + if (wakeUpRequested) { + wakeUpRequested = false; + consecutiveEmptyScans = 0; + nextScanAtNanos = System.nanoTime(); + return; + } + long nextIntervalMillis = hasPendingRecords ? activeIntervalMillis : nextIdleIntervalMillis(); + if (hasPendingRecords) { + consecutiveEmptyScans = 0; + } + nextScanAtNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(nextIntervalMillis); + } + + /** + * 唤醒扫描门；扫描进行中时记录延迟唤醒，空闲时立即允许下一次调度触发扫描。 + */ + private synchronized void wakeUp() { + consecutiveEmptyScans = 0; + if (scanInProgress) { + wakeUpRequested = true; + return; + } + nextScanAtNanos = System.nanoTime(); + } + + /** + * 按连续空扫描次数计算下一次退避间隔，并限制在当前扫描门最大空闲间隔内。 + * + * @return 下一次数据库扫描等待时长，单位毫秒 + */ + private long nextIdleIntervalMillis() { + consecutiveEmptyScans = Math.min(consecutiveEmptyScans + 1, MAX_BACKOFF_SHIFT); + long multiplier = 1L << consecutiveEmptyScans; + return Math.min(activeIntervalMillis * multiplier, maxIdleIntervalMillis); + } + } +} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImpl.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImpl.java index 20d4328bf..f028dc4e0 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImpl.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImpl.java @@ -218,6 +218,11 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay */ private final StringEncryptor stringEncryptor; + /** + * 服务端轮询协调器，在聚合会话创建或复用成功后唤醒当前实例的自适应扫描任务。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + /** * 进程级强随机数生成器，用于不可预测的聚合访问令牌。 */ @@ -239,7 +244,9 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay throw new RuntimeException("聚合订单码正在处理中，请勿重复提交"); } try { - return createOrReuse(dto); + ThirdAggregateOrderCodePayVo result = createOrReuse(dto); + pollingCoordinator.wakeAggregatePolling(); + return result; } finally { releaseLock(lockKey, lockToken); } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java index 6d2c39d82..50054cd15 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java @@ -164,6 +164,11 @@ public class ThirdAppPlusServiceImpl implements ThirdAppPlusService { */ private final ThirdAggregateOrderPayService thirdAggregateOrderPayService; + /** + * 服务端轮询协调器，在原生订单码创建或复用成功后唤醒当前实例的自适应扫描任务。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + /** * Redis 字符串模板，用于同一租户业务订单的短时并发互斥。 */ @@ -535,7 +540,9 @@ public class ThirdAppPlusServiceImpl implements ThirdAppPlusService { throw new RuntimeException("订单码支付正在处理中，请勿重复提交"); } try { - return createOrReuseOrderCodePay(dto); + ThirdOrderCodePayVo result = createOrReuseOrderCodePay(dto); + pollingCoordinator.wakeOrderCodePolling(); + return result; } finally { releaseOrderCodePayLock(lockKey, lockToken); } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java index a5d2332a4..741a7355e 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java @@ -6,6 +6,7 @@ import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusService; import com.bsd.pjcloud.thirdparty.service.ThirdPayOrderService; +import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -19,25 +20,50 @@ import java.util.List; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; +/** + * 原生订单码服务端轮询任务的租户恢复和自适应扫描测试。 + */ @ExtendWith(MockitoExtension.class) class ThirdAppPlusPayJobTest { + /** + * 支付订单服务 Mock，用于返回跨租户待处理订单。 + */ @Mock private ThirdPayOrderService thirdPayOrderService; + /** + * 支付编排服务 Mock，用于验证恢复租户后执行平台查单。 + */ @Mock private ThirdAppPlusService thirdAppPlusService; + /** + * 自适应轮询协调器 Mock，用于控制本轮是否允许访问数据库。 + */ + @Mock + private ThirdPayPollingCoordinator pollingCoordinator; + + /** + * 被测原生订单码轮询任务。 + */ @InjectMocks private ThirdAppPlusPayJob job; + /** + * 清理测试期间由 TenantBroker 设置的线程租户上下文。 + */ @AfterEach void tearDown() { TenantContextHolder.setTenantId(null); } + /** + * 验证取得扫描资格后按订单租户恢复上下文，并把非空结果反馈给协调器。 + */ @Test void handleOrderCodeOrdersShouldRestoreEachOrderTenant() { ThirdPayOrderEntity order = new ThirdPayOrderEntity(); @@ -45,6 +71,7 @@ class ThirdAppPlusPayJobTest { order.setTenantId(66L); order.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); order.setExpireTime(LocalDateTime.now().plusMinutes(1)); + when(pollingCoordinator.tryStartOrderCodePolling()).thenReturn(true); when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); when(thirdPayOrderService.getById(order.getId())).thenAnswer(invocation -> { Assertions.assertEquals(66L, TenantContextHolder.getTenantId()); @@ -54,6 +81,19 @@ class ThirdAppPlusPayJobTest { job.handleOrderCodeOrders(); verify(thirdAppPlusService).queryPayResult(order); + verify(pollingCoordinator).finishOrderCodePolling(true); Assertions.assertNull(TenantContextHolder.getTenantId()); } + + /** + * 验证空闲退避尚未到期时任务直接返回，不执行任何数据库查询。 + */ + @Test + void handleOrderCodeOrdersShouldSkipDatabaseWhenBackoffIsActive() { + when(pollingCoordinator.tryStartOrderCodePolling()).thenReturn(false); + + job.handleOrderCodeOrders(); + + verifyNoInteractions(thirdPayOrderService, thirdAppPlusService); + } } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinatorTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinatorTest.java new file mode 100644 index 000000000..997dc7a97 --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinatorTest.java @@ -0,0 +1,48 @@ +package com.bsd.pjcloud.thirdparty.service; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** + * 第三方订单码自适应轮询协调器测试。 + */ +class ThirdPayPollingCoordinatorTest { + + /** + * 验证空扫描完成后立即触发会被退避门拦截，而订单创建唤醒可以立即恢复扫描资格。 + */ + @Test + void emptyScanShouldBackoffUntilOrderCreationWakesGate() { + ThirdPayPollingCoordinator coordinator = new ThirdPayPollingCoordinator(); + + Assertions.assertTrue(coordinator.tryStartOrderCodePolling()); + coordinator.finishOrderCodePolling(false); + Assertions.assertFalse(coordinator.tryStartOrderCodePolling()); + + coordinator.wakeOrderCodePolling(); + + Assertions.assertTrue(coordinator.tryStartOrderCodePolling()); + coordinator.finishOrderCodePolling(true); + } + + /** + * 验证聚合会话唤醒会同时解除活动扫描和过期补偿扫描的空闲退避。 + */ + @Test + void aggregateWakeShouldReleasePendingAndCompensationBackoff() { + ThirdPayPollingCoordinator coordinator = new ThirdPayPollingCoordinator(); + Assertions.assertTrue(coordinator.tryStartAggregatePendingPolling()); + coordinator.finishAggregatePendingPolling(false); + Assertions.assertTrue(coordinator.tryStartAggregateCompensationPolling()); + coordinator.finishAggregateCompensationPolling(false); + Assertions.assertFalse(coordinator.tryStartAggregatePendingPolling()); + Assertions.assertFalse(coordinator.tryStartAggregateCompensationPolling()); + + coordinator.wakeAggregatePolling(); + + Assertions.assertTrue(coordinator.tryStartAggregatePendingPolling()); + coordinator.finishAggregatePendingPolling(true); + Assertions.assertTrue(coordinator.tryStartAggregateCompensationPolling()); + coordinator.finishAggregateCompensationPolling(true); + } +} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImplTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImplTest.java index effef4c5d..9a3083f5c 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImplTest.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAggregateOrderPayServiceImplTest.java @@ -19,6 +19,7 @@ import com.bsd.pjcloud.thirdparty.service.ThirdAggregateAliPayService; import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; import com.bsd.pjcloud.thirdparty.service.ThirdAggregateWechatPayService; import com.bsd.pjcloud.thirdparty.service.ThirdPayOrderService; +import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; import com.bsd.pjcloud.thirdparty.vo.appplus.pay.ThirdAggregateOrderCodePayVo; import com.bsd.pjcloud.wallet.api.entity.mch.MchPaySettingEntity; import com.bsd.pjcloud.wallet.api.feign.RemoteMchService; @@ -106,6 +107,12 @@ class ThirdAggregateOrderPayServiceImplTest { @Mock private StringEncryptor stringEncryptor; + /** + * 服务端轮询协调器 Mock，用于验证聚合会话创建后主动唤醒扫描任务。 + */ + @Mock + private ThirdPayPollingCoordinator pollingCoordinator; + /** * 被测聚合支付编排服务。 */ @@ -129,7 +136,7 @@ class ThirdAggregateOrderPayServiceImplTest { AppPlusPayConfig appPlusPayConfig = completeApplicationConfig(); service = new ThirdAggregateOrderPayServiceImpl(aggregatePaySessionService, aggregateAliPayService, aggregateWechatPayService, thirdPayOrderService, remoteMchService, - remoteTenantService, appPlusPayConfig, stringRedisTemplate, stringEncryptor); + remoteTenantService, appPlusPayConfig, stringRedisTemplate, stringEncryptor, pollingCoordinator); lenient().when(stringRedisTemplate.opsForValue()).thenReturn(valueOperations); lenient().when(valueOperations.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(true); @@ -175,6 +182,7 @@ class ThirdAggregateOrderPayServiceImplTest { Assertions.assertEquals(ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), savedSession.getStatus()); Assertions.assertNull(savedSession.getSelectedPlatform()); Assertions.assertNull(savedSession.getPayNo()); + verify(pollingCoordinator).wakeAggregatePolling(); } /** diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImplOrderCodePayTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImplOrderCodePayTest.java index ffc792f41..53db7cc0a 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImplOrderCodePayTest.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImplOrderCodePayTest.java @@ -79,6 +79,12 @@ class ThirdAppPlusServiceImplOrderCodePayTest { @Mock private ThirdAggregateOrderPayService thirdAggregateOrderPayService; + /** + * 服务端轮询协调器 Mock，用于验证原生订单码创建后主动唤醒扫描任务。 + */ + @Mock + private ThirdPayPollingCoordinator pollingCoordinator; + @Mock private StringRedisTemplate stringRedisTemplate; @@ -204,6 +210,7 @@ class ThirdAppPlusServiceImplOrderCodePayTest { ArgumentCaptor lockKeyCaptor = ArgumentCaptor.forClass(String.class); verify(valueOperations).setIfAbsent(lockKeyCaptor.capture(), any(), any()); Assertions.assertEquals("thirdparty:order-code-pay:lock:20:order-1001", lockKeyCaptor.getValue()); + verify(pollingCoordinator).wakeOrderCodePolling(); } @Test