diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJob.java b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJob.java index 3af02e996..5d601c533 100644 --- a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJob.java +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJob.java @@ -6,12 +6,14 @@ import com.bsd.pjcloud.common.data.tenant.TenantBroker; import com.bsd.pjcloud.hard.api.constant.order.HardOrderStatusEnum; import com.bsd.pjcloud.hard.api.constant.order.HardPayTypeEnum; import com.bsd.pjcloud.hard.api.entity.order.HardOrderEntity; +import com.bsd.pjcloud.hard.service.HardAggregatePayPollingCoordinator; import com.bsd.pjcloud.hard.service.HardOrderService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import java.time.LocalDateTime; import java.util.List; /** @@ -30,25 +32,99 @@ public class HardAggregateOrderPayJob { */ private static final int BATCH_SIZE = 100; + /** + * 近期订单高频补偿窗口，单位分钟；覆盖聚合码有效期及 Thirdparty 关单补偿时间。 + */ + private static final long ACTIVE_ORDER_WINDOW_MINUTES = 15L; + + /** + * 超出近期窗口但尚未达到自动取消时间的订单低频核对间隔，单位毫秒。 + */ + private static final long STALE_RECONCILIATION_INTERVAL_MILLIS = 300000L; + + /** + * 服务启动后首次核对遗留订单的等待时间，单位毫秒，避免与启动恢复扫描同时争用资源。 + */ + private static final long STALE_INITIAL_DELAY_MILLIS = 60000L; + /** * Hardware 订单服务，复用带订单锁和事务的聚合支付状态确认及结算流程。 */ private final HardOrderService hardOrderService; /** - * 每五秒补偿仍处于创建状态且已绑定聚合会话的 Hardware 订单。 + * 近期聚合订单轮询协调器，用于在空闲时退避数据库扫描并响应新订单唤醒。 + */ + private final HardAggregatePayPollingCoordinator pollingCoordinator; + + /** + * 高频补偿最近十五分钟内仍处于创建状态且已绑定聚合会话的 Hardware 订单。 * *

调度线程无租户上下文时先跨租户读取候选记录，再按记录持久化的租户 ID 恢复上下文； - * 具体结算继续由业务服务的分布式订单锁和状态条件保证幂等。

+ * 具体结算继续由业务服务的分布式订单锁和状态条件保证幂等。方法每五秒轻量触发一次，连续空结果时 + * 真正的数据库扫描按十秒、二十秒、四十秒退避到最长一分钟。

*/ - @Scheduled(fixedDelay = 5000L, initialDelay = 5000L) + @Scheduled(fixedDelay = HardAggregatePayPollingCoordinator.TRIGGER_INTERVAL_MILLIS, + initialDelay = HardAggregatePayPollingCoordinator.TRIGGER_INTERVAL_MILLIS) public void settlePendingAggregateOrders() { - List orders = hardOrderService.list(new LambdaQueryWrapper() + if (!pollingCoordinator.tryStartPolling()) { + return; + } + boolean hasPendingOrders = true; + try { + LocalDateTime activeSince = LocalDateTime.now().minusMinutes(ACTIVE_ORDER_WINDOW_MINUTES); + List orders = listPendingOrders( + new LambdaQueryWrapper().ge(HardOrderEntity::getCreateTime, activeSince)); + hasPendingOrders = ObjectUtil.isNotEmpty(orders); + processOrders(orders); + } finally { + // 查询异常不能被当作空闲，否则数据库或远程服务恢复前会被错误放大重试间隔。 + pollingCoordinator.finishPolling(hasPendingOrders); + } + } + + /** + * 低频核对十五分钟至一小时内的遗留订单，并取消超过一小时的订单。 + * + *

一小时以内仍保留服务异常恢复能力；超过一小时后不再调用 Thirdparty 查单，而是把仍处于创建状态 + * 的 Hardware 订单原子更新为未支付取消，避免异常记录被永久轮询。第三方若在此后报告延迟支付， + * 由线下渠道核对和协商，不再自动改变 Hardware 库存及订单状态。

+ */ + @Scheduled(fixedDelay = STALE_RECONCILIATION_INTERVAL_MILLIS, initialDelay = STALE_INITIAL_DELAY_MILLIS) + public void settleStaleAggregateOrders() { + LocalDateTime now = LocalDateTime.now(); + LocalDateTime activeSince = now.minusMinutes(ACTIVE_ORDER_WINDOW_MINUTES); + LocalDateTime cancelBefore = now.minusMinutes(HardAggregatePayPollingCoordinator.MAX_AUTO_SETTLEMENT_MINUTES); + List expiredOrders = listPendingOrders( + new LambdaQueryWrapper().lt(HardOrderEntity::getCreateTime, cancelBefore)); + cancelExpiredOrders(expiredOrders, cancelBefore); + List recoverableOrders = listPendingOrders(new LambdaQueryWrapper() + .ge(HardOrderEntity::getCreateTime, cancelBefore) + .lt(HardOrderEntity::getCreateTime, activeSince)); + processOrders(recoverableOrders); + } + + /** + * 按附加时间条件读取仍处于创建状态的聚合支付 Hardware 订单。 + * + * @param timeWrapper 近期或遗留订单的创建时间条件 + * @return 按创建时间升序排列且最多一百条的待结算订单 + */ + private List listPendingOrders(LambdaQueryWrapper timeWrapper) { + return hardOrderService.list(timeWrapper .eq(HardOrderEntity::getPayType, HardPayTypeEnum.AGGREGATE_ORDER_PAY.getType()) .eq(HardOrderEntity::getOrderStatus, HardOrderStatusEnum.CREATE.getType()) .isNotNull(HardOrderEntity::getOrderNo) .orderByAsc(HardOrderEntity::getCreateTime) .last("limit " + BATCH_SIZE)); + } + + /** + * 按订单租户恢复上下文并调用已有幂等结算流程。 + * + * @param orders 本轮需要核对的 Hardware 聚合支付订单；为空时不执行远程调用 + */ + private void processOrders(List orders) { if (ObjectUtil.isEmpty(orders)) { return; } @@ -66,4 +142,33 @@ public class HardAggregateOrderPayJob { } } } + + /** + * 将超过一小时仍未结算的聚合支付订单原子更新为未支付取消状态。 + * + *

更新同时校验支付方式、创建状态和创建时间，避免与临界时刻的终端查询或支付结算互相覆盖。 + * 此处使用 {@link HardOrderStatusEnum#UNFOLLOW} 表示未支付自动取消，不能使用代表已核销完成的 + * {@link HardOrderStatusEnum#CANCEL}。

+ * + * @param orders 本轮查询到的超时 Hardware 聚合支付订单 + * @param cancelBefore 本轮统一的一小时截止时间 + */ + private void cancelExpiredOrders(List orders, LocalDateTime cancelBefore) { + if (ObjectUtil.isEmpty(orders)) { + return; + } + for (HardOrderEntity order : orders) { + if (order.getTenantId() == null) { + log.warn("超时聚合支付Hardware订单缺少持久化租户ID,orderId:{}", order.getId()); + continue; + } + try { + TenantBroker.runAs(order.getTenantId(), tenantId -> + hardOrderService.cancelExpiredAggregateOrder(order.getId(), cancelBefore)); + } catch (Exception exception) { + log.warn("取消超时聚合支付Hardware订单失败,orderId:{},tenantId:{},exceptionType:{}", + order.getId(), order.getTenantId(), exception.getClass().getSimpleName()); + } + } + } } diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinator.java b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinator.java new file mode 100644 index 000000000..27a71ed20 --- /dev/null +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinator.java @@ -0,0 +1,112 @@ +package com.bsd.pjcloud.hard.service; + +import org.springframework.stereotype.Component; + +import java.util.concurrent.TimeUnit; + +/** + * Hardware 聚合订单支付结算轮询协调器。 + * + *

协调器只控制当前服务实例访问数据库的频率，不保存业务订单状态。存在近期待结算订单或查询异常时 + * 保持五秒补偿；连续空扫描时指数退避，创建订单后主动唤醒，避免无订单时全天执行固定频率 SQL。

+ */ +@Component +public class HardAggregatePayPollingCoordinator { + + /** + * 近期聚合订单结算任务的轻量触发周期，单位毫秒；数据库扫描还会受空闲退避门控制。 + */ + public static final long TRIGGER_INTERVAL_MILLIS = 5000L; + + /** + * 聚合支付 Hardware 订单允许自动结算的最长时间，单位分钟；状态查询和后台任务必须共用该值。 + */ + public static final long MAX_AUTO_SETTLEMENT_MINUTES = 60L; + + /** + * 连续空扫描时允许达到的最大数据库扫描间隔，单位毫秒。 + */ + private static final long MAX_IDLE_INTERVAL_MILLIS = 60000L; + + /** + * 指数退避最大左移位数，防止连续空扫描计数和间隔计算无限增长。 + */ + private static final int MAX_BACKOFF_SHIFT = 8; + + /** + * 连续空扫描次数；发现订单或收到创建唤醒后清零。 + */ + private int consecutiveEmptyScans; + + /** + * 使用 {@link System#nanoTime()} 时间基准保存的下次允许扫描时刻。 + */ + private long nextScanAtNanos = System.nanoTime(); + + /** + * 当前是否已有线程取得扫描资格，用于防止调度线程重叠访问数据库。 + */ + private boolean scanInProgress; + + /** + * 扫描执行期间是否收到新订单唤醒，确保扫描结束时不会覆盖唤醒信号。 + */ + private boolean wakeUpRequested; + + /** + * 原子判断近期聚合订单是否到达下次数据库扫描时间，并锁定本轮扫描资格。 + * + * @return 到达扫描时间且当前没有同类扫描执行时返回 {@code true} + */ + public synchronized boolean tryStartPolling() { + long now = System.nanoTime(); + if (scanInProgress || now - nextScanAtNanos < 0) { + return false; + } + scanInProgress = true; + return true; + } + + /** + * 释放扫描资格并按查询结果安排下一次数据库扫描。 + * + * @param hasPendingOrders 本轮是否查询到近期待结算订单；查询异常时传 {@code true} 以保持短间隔重试 + */ + public synchronized void finishPolling(boolean hasPendingOrders) { + scanInProgress = false; + if (wakeUpRequested) { + wakeUpRequested = false; + consecutiveEmptyScans = 0; + nextScanAtNanos = System.nanoTime(); + return; + } + long nextIntervalMillis = hasPendingOrders ? TRIGGER_INTERVAL_MILLIS : nextIdleIntervalMillis(); + if (hasPendingOrders) { + consecutiveEmptyScans = 0; + } + nextScanAtNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(nextIntervalMillis); + } + + /** + * 聚合订单创建或复用成功后唤醒近期扫描，最迟在下一个五秒调度周期访问数据库。 + */ + public synchronized void wakePolling() { + 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(TRIGGER_INTERVAL_MILLIS * multiplier, MAX_IDLE_INTERVAL_MILLIS); + } +} diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardOrderService.java b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardOrderService.java index b7e62933b..473680f2c 100644 --- a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardOrderService.java +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/HardOrderService.java @@ -40,6 +40,7 @@ import com.bsd.pjcloud.thirdparty.dto.ks.KsRecognitionCallbackDto; import com.bsd.pjcloud.wallet.api.vo.MchManageVo; import jakarta.validation.Valid; +import java.time.LocalDateTime; import java.util.List; public interface HardOrderService extends IService { @@ -139,6 +140,15 @@ public interface HardOrderService extends IService { */ HardAggregateOrderCodePayStatusVo getAggregateOrderCodePayStatus(Long orderId, String aggregateNo); + /** + * 在订单锁内取消超过服务端自动结算时限的聚合支付订单。 + * + * @param orderId Hardware 业务订单 ID + * @param cancelBefore 允许自动结算的订单创建时间截止点 + * @return 当前订单仍为聚合支付创建状态且已超时时返回 {@code true} + */ + boolean cancelExpiredAggregateOrder(Long orderId, LocalDateTime cancelBefore); + /** * 用户确认设备二维码扫一扫订单 */ diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/order/HardOrderServiceImpl.java b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/order/HardOrderServiceImpl.java index 6207100e6..8879bb66f 100644 --- a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/order/HardOrderServiceImpl.java +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/order/HardOrderServiceImpl.java @@ -206,6 +206,9 @@ public class HardOrderServiceImpl extends ServiceImpl延迟到提交后可避免调度线程提前查询不到刚创建或刚绑定的订单；没有事务同步上下文时立即唤醒， + * 保证非代理内部调用也不会丢失服务端补偿信号。

+ */ + private void wakeAggregatePayPollingAfterCommit() { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + aggregatePayPollingCoordinator.wakePolling(); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + /** + * 事务提交完成后解除当前实例的空闲退避，使新订单最迟在下一个五秒周期被发现。 + */ + @Override + public void afterCommit() { + aggregatePayPollingCoordinator.wakePolling(); + } + }); } /** * 查询同一终端本地订单号尚未结束的聚合支付订单，供创建响应丢失后的重试复用。 * * @param dto 当前收银端创建请求 - * @return 同设备、同商户的待支付或已支付聚合订单；不存在时返回 {@code null} + * @return 同设备、同商户的一小时内待支付订单或既有已支付订单；不存在时返回 {@code null} */ private HardOrderEntity getReusableAggregateOrder(HardCreateOrderDto dto) { + LocalDateTime cancelBefore = LocalDateTime.now() + .minusMinutes(HardAggregatePayPollingCoordinator.MAX_AUTO_SETTLEMENT_MINUTES); return getOne(new LambdaQueryWrapper() .eq(HardOrderEntity::getAndroidOrderCode, dto.getAndroidOrderCode()) .eq(HardOrderEntity::getHardId, dto.getHardwareId()) .eq(HardOrderEntity::getMerchantId, dto.getMerchantId()) .eq(HardOrderEntity::getPayType, HardPayTypeEnum.AGGREGATE_ORDER_PAY.getType()) - .in(HardOrderEntity::getOrderStatus, - HardOrderStatusEnum.CREATE.getType(), HardOrderStatusEnum.PAY.getType()) + .and(statusWrapper -> statusWrapper + .eq(HardOrderEntity::getOrderStatus, HardOrderStatusEnum.PAY.getType()) + .or(waitingWrapper -> waitingWrapper + .eq(HardOrderEntity::getOrderStatus, HardOrderStatusEnum.CREATE.getType()) + .ge(HardOrderEntity::getCreateTime, cancelBefore))) .orderByDesc(HardOrderEntity::getCreateTime) .last("limit 1")); } @@ -988,6 +1022,32 @@ public class HardOrderServiceImpl extends ServiceImpl statusResult = thirdAppPlusPayService.getAggregateOrderCodePayStatus(aggregateNo); if (statusResult == null || !statusResult.isOk() || statusResult.getData() == null) { @@ -1042,6 +1102,35 @@ public class HardOrderServiceImpl extends ServiceImpl锁用于防止超时任务与终端支付成功结算交叉执行；数据库更新继续校验支付方式、创建状态和 + * 创建时间，因此已支付、已取消或截止时间内的订单不会被覆盖。

+ * + * @param orderId Hardware 业务订单 ID + * @param cancelBefore 允许自动结算的订单创建时间截止点 + * @return 成功把订单推进为未支付取消状态时返回 {@code true} + */ + @Override + @Lock4j(keys = {"#orderId"}) + @Transactional(rollbackFor = Exception.class) + public boolean cancelExpiredAggregateOrder(Long orderId, LocalDateTime cancelBefore) { + if (orderId == null || cancelBefore == null) { + return false; + } + return update(new LambdaUpdateWrapper() + .eq(HardOrderEntity::getId, orderId) + .eq(HardOrderEntity::getPayType, HardPayTypeEnum.AGGREGATE_ORDER_PAY.getType()) + .eq(HardOrderEntity::getOrderStatus, HardOrderStatusEnum.CREATE.getType()) + .and(timeWrapper -> timeWrapper + .isNull(HardOrderEntity::getCreateTime) + .or() + .le(HardOrderEntity::getCreateTime, cancelBefore)) + .set(HardOrderEntity::getOrderStatus, HardOrderStatusEnum.UNFOLLOW.getType()) + .set(HardOrderEntity::getUnfollowTime, LocalDateTime.now())); + } + /** * 用户确认设备二维码扫一扫订单 */ diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJobTest.java b/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJobTest.java new file mode 100644 index 000000000..0294a7e8e --- /dev/null +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/job/HardAggregateOrderPayJobTest.java @@ -0,0 +1,77 @@ +package com.bsd.pjcloud.hard.job; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.bsd.pjcloud.common.data.tenant.TenantContextHolder; +import com.bsd.pjcloud.hard.api.constant.order.HardOrderStatusEnum; +import com.bsd.pjcloud.hard.api.constant.order.HardPayTypeEnum; +import com.bsd.pjcloud.hard.api.entity.order.HardOrderEntity; +import com.bsd.pjcloud.hard.service.HardAggregatePayPollingCoordinator; +import com.bsd.pjcloud.hard.service.HardOrderService; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Hardware 聚合订单支付补偿任务测试。 + */ +@ExtendWith(MockitoExtension.class) +class HardAggregateOrderPayJobTest { + + /** + * Hardware 订单服务 Mock，用于返回历史订单并验证带锁取消入口。 + */ + @Mock + private HardOrderService hardOrderService; + + /** + * 近期轮询协调器 Mock；历史取消路径不依赖客户端或近期扫描门。 + */ + @Mock + private HardAggregatePayPollingCoordinator pollingCoordinator; + + /** + * 被测 Hardware 聚合订单支付补偿任务。 + */ + @InjectMocks + private HardAggregateOrderPayJob job; + + /** + * 清理测试期间由 TenantBroker 设置的线程租户上下文。 + */ + @AfterEach + void tearDown() { + TenantContextHolder.setTenantId(null); + } + + /** + * 验证超过一小时的订单只进入带锁取消，不再调用 Thirdparty 状态查询流程。 + */ + @Test + void staleJobShouldCancelOrderOlderThanOneHourWithoutQueryingPaymentStatus() { + HardOrderEntity order = new HardOrderEntity(); + order.setId(101L); + order.setTenantId(66L); + order.setOrderNo("aggregate-no"); + order.setPayType(HardPayTypeEnum.AGGREGATE_ORDER_PAY.getType()); + order.setOrderStatus(HardOrderStatusEnum.CREATE.getType()); + order.setCreateTime(LocalDateTime.now().minusHours(2)); + when(hardOrderService.list(any(Wrapper.class))).thenReturn(List.of(order), List.of()); + + job.settleStaleAggregateOrders(); + + verify(hardOrderService).cancelExpiredAggregateOrder(eq(order.getId()), any(LocalDateTime.class)); + verify(hardOrderService, never()).getAggregateOrderCodePayStatus(any(), any()); + } +} diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinatorTest.java b/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinatorTest.java new file mode 100644 index 000000000..1e3c5cd32 --- /dev/null +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/test/java/com/bsd/pjcloud/hard/service/HardAggregatePayPollingCoordinatorTest.java @@ -0,0 +1,42 @@ +package com.bsd.pjcloud.hard.service; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** + * Hardware 聚合订单支付自适应轮询协调器测试。 + */ +class HardAggregatePayPollingCoordinatorTest { + + /** + * 验证空扫描会进入退避，而创建订单唤醒能够立即恢复扫描资格。 + */ + @Test + void emptyScanShouldBackoffUntilOrderCreationWakesPolling() { + HardAggregatePayPollingCoordinator coordinator = new HardAggregatePayPollingCoordinator(); + + Assertions.assertTrue(coordinator.tryStartPolling()); + coordinator.finishPolling(false); + Assertions.assertFalse(coordinator.tryStartPolling()); + + coordinator.wakePolling(); + + Assertions.assertTrue(coordinator.tryStartPolling()); + coordinator.finishPolling(true); + } + + /** + * 验证扫描执行期间到达的唤醒信号不会被本轮空查询结果覆盖。 + */ + @Test + void wakeDuringScanShouldRemainEffectiveAfterEmptyResult() { + HardAggregatePayPollingCoordinator coordinator = new HardAggregatePayPollingCoordinator(); + + Assertions.assertTrue(coordinator.tryStartPolling()); + coordinator.wakePolling(); + coordinator.finishPolling(false); + + Assertions.assertTrue(coordinator.tryStartPolling()); + coordinator.finishPolling(true); + } +} 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..56c82b9ab 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 @@ -2,11 +2,16 @@ package com.bsd.pjcloud.thirdparty.job; import cn.hutool.core.util.ObjectUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.bsd.pjcloud.common.data.tenant.TenantBroker; import com.bsd.pjcloud.thirdparty.constant.pay.ThirdAggregatePaySessionStatusEnum; +import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; import com.bsd.pjcloud.thirdparty.entity.pay.ThirdAggregatePaySessionEntity; +import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; +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; @@ -28,16 +33,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,67 +49,151 @@ public class ThirdAggregateOrderPayJob { private final ThirdAggregateOrderPayService aggregateOrderPayService; /** - * 每五秒分批处理等待扫码、平台创建中和支付中的聚合会话。 + * 第三方支付子订单服务，用于在一小时硬截止后仅做本地关闭标记。 + */ + private final ThirdPayOrderService thirdPayOrderService; + + /** + * 服务端轮询协调器，用于在无活动会话时退避数据库扫描，并在新会话创建后恢复短间隔。 + */ + 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 { + LocalDateTime historyCutoff = LocalDateTime.now() + .minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES); + List sessions = aggregatePaySessionService.list( + new LambdaQueryWrapper() + .in(ThirdAggregatePaySessionEntity::getStatus, + ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), + ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), + ThirdAggregatePaySessionStatusEnum.PAYING.getType()) + .ge(ThirdAggregatePaySessionEntity::getCreateTime, historyCutoff) + .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 (!pollingCoordinator.tryStartAggregateCompensationPolling()) { + return; + } + boolean hasExpiredSessions = true; + try { + LocalDateTime now = LocalDateTime.now(); + List historicalSessions = aggregatePaySessionService.list( + new LambdaQueryWrapper() + .in(ThirdAggregatePaySessionEntity::getStatus, + ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), + ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), + ThirdAggregatePaySessionStatusEnum.PAYING.getType(), + ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()) + .and(timeWrapper -> timeWrapper + .isNull(ThirdAggregatePaySessionEntity::getCreateTime) + .or() + .lt(ThirdAggregatePaySessionEntity::getCreateTime, + now.minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES))) + .orderByAsc(ThirdAggregatePaySessionEntity::getCreateTime) + .last("limit " + BATCH_SIZE)); + closeHistoricalSessions(historicalSessions); + 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(historicalSessions) || ObjectUtil.isNotEmpty(sessions); + 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); + } + } + + /** + * 将超过一小时的过期聚合会话及其未支付子订单标记为交易关闭。 + * + *

该分支不调用任何支付平台接口，也不再根据子订单恢复会话。会话先进入关闭终态，随后仅把仍为 + * 未支付的子订单标记为关闭；已经落库的真实支付结果保留给人工对账，但不会重新激活聚合会话。

+ * + * @param sessions 超过一小时且仍处于活动或过期状态的聚合会话 + */ + private void closeHistoricalSessions(List sessions) { if (ObjectUtil.isEmpty(sessions)) { return; } for (ThirdAggregatePaySessionEntity session : sessions) { if (session.getTenantId() == null) { - log.warn("到期聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); + log.warn("历史聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); continue; } try { - TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); + TenantBroker.runAs(session.getTenantId(), tenantId -> { + aggregatePaySessionService.markTerminal( + session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + ThirdPayOrderEntity payOrder = session.getPayNo() == null + ? null : thirdPayOrderService.getByPayNo(session.getPayNo()); + if (payOrder != null) { + thirdPayOrderService.update(new LambdaUpdateWrapper() + .eq(ThirdPayOrderEntity::getId, payOrder.getId()) + .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) + .set(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_CLOSED.getType())); + } + }); } catch (Exception exception) { - log.error("补偿关闭聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); + log.warn("关闭历史聚合支付会话失败,sessionId:{},tenantId:{},exceptionType:{}", + session.getId(), session.getTenantId(), exception.getClass().getSimpleName()); } } } 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..3224b664c 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 @@ -3,6 +3,7 @@ package com.bsd.pjcloud.thirdparty.job; import cn.hutool.core.util.ObjectUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.bsd.pjcloud.common.core.constant.enums.YesNoEnum; import com.bsd.pjcloud.common.data.tenant.TenantBroker; import com.bsd.pjcloud.thirdparty.constant.miniplatform.AppPlusPlatformTypeEnum; @@ -11,6 +12,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,16 +34,21 @@ public class ThirdAppPlusPayJob { */ private static final int ORDER_CODE_BATCH_SIZE = 100; - /** - * 原生订单码的服务端查单间隔，单位毫秒。 - */ - private static final long ORDER_CODE_POLL_INTERVAL_MILLIS = 5000L; - /** * 原生订单码到期后的关单补偿窗口，单位分钟；超过窗口后停止后台查询。 */ private static final long ORDER_CODE_CLOSE_COMPENSATION_MINUTES = 10L; + /** + * 超过历史恢复时间的原生订单码本地关闭标记周期，单位毫秒。 + */ + private static final long ORDER_CODE_HISTORY_CLOSE_INTERVAL_MILLIS = 300000L; + + /** + * 服务启动后首次标记历史订单码的等待时间，单位毫秒。 + */ + private static final long ORDER_CODE_HISTORY_CLOSE_INITIAL_DELAY_MILLIS = 60000L; + /** * 支付订单服务，用于分批读取仍未支付的订单码记录。 */ @@ -52,6 +59,11 @@ public class ThirdAppPlusPayJob { */ private final ThirdAppPlusService thirdAppPlusService; + /** + * 服务端轮询协调器，用于在无待处理订单时退避数据库扫描，并在创建订单后恢复短间隔。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + /** * 处理,未勾起业务回调的订单 */ @@ -76,42 +88,100 @@ 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); + if (!pollingCoordinator.tryStartOrderCodePolling()) { + return; + } + 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; + } + 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); + } + } + + /** + * 将超过一小时仍未支付的原生支付宝、微信订单码标记为交易关闭。 + * + *

该任务只执行带未支付状态条件的数据库更新，不再向支付宝或微信发起查单、撤销或关单请求； + * 平台若在硬截止后出现延迟支付，由人工渠道核对处理。

+ */ + @Scheduled(fixedDelay = ORDER_CODE_HISTORY_CLOSE_INTERVAL_MILLIS, + initialDelay = ORDER_CODE_HISTORY_CLOSE_INITIAL_DELAY_MILLIS) + public void closeHistoricalOrderCodeOrders() { + LocalDateTime cutoff = LocalDateTime.now() + .minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_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) + .and(timeWrapper -> timeWrapper + .isNull(ThirdPayOrderEntity::getCreateTime) + .or() + .lt(ThirdPayOrderEntity::getCreateTime, cutoff)) .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)) { - return; - } for (ThirdPayOrderEntity order : orders) { if (order.getTenantId() == null) { - log.warn("订单码缺少租户ID,orderId:{},platform:{}", order.getId(), order.getPayPlatformType()); + 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); - } - }); + TenantBroker.runAs(order.getTenantId(), tenantId -> thirdPayOrderService.update( + new LambdaUpdateWrapper() + .eq(ThirdPayOrderEntity::getId, order.getId()) + .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) + .and(timeWrapper -> timeWrapper + .isNull(ThirdPayOrderEntity::getCreateTime) + .or() + .lt(ThirdPayOrderEntity::getCreateTime, cutoff)) + .set(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_CLOSED.getType()))); } catch (Exception exception) { - log.error("处理订单码状态异常,orderId:{},tenantId:{},platform:{}", - order.getId(), order.getTenantId(), order.getPayPlatformType(), exception); + log.warn("关闭历史订单码失败,orderId:{},tenantId:{},platform:{},exceptionType:{}", + order.getId(), order.getTenantId(), order.getPayPlatformType(), + exception.getClass().getSimpleName()); } } } diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdAggregateOrderPayService.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdAggregateOrderPayService.java index f031b4bd5..92ae37881 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdAggregateOrderPayService.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdAggregateOrderPayService.java @@ -75,14 +75,14 @@ public interface ThirdAggregateOrderPayService { ThirdAggregateOrderCodePayStatusVo getStatusByTicket(String tenantCode, String ticket); /** - * 查询、关闭并同步一条活动会话或处于有限关单补偿窗口内的过期会话。 + * 在一小时自动处理期限内查询、关闭并同步活动或过期会话，期限外仅执行本地关闭标记。 * * @param sessionId 聚合支付会话主键 */ void processPendingSession(Long sessionId); /** - * 判断支付回调订单是否属于聚合会话；属于时同步会话终态并阻止旧业务回调链继续执行。 + * 判断支付回调订单是否属于聚合会话；期限内同步终态，期限外关闭会话并阻止旧业务回调链执行。 * * @param payOrder 已通过平台验签和关键字段校验的最新第三方支付订单 * @return 属于聚合 JSAPI 子订单时返回 {@code true} 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..9de0f7f07 --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/ThirdPayPollingCoordinator.java @@ -0,0 +1,250 @@ +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; + + /** + * 订单码允许服务端自动查单、关单或恢复会话状态的最长时间，单位分钟。 + * + *

超过该硬截止后只允许本地标记关闭，客户端查询、定时任务和创建重试均不得再次访问支付平台。

+ */ + public static final long MAX_AUTO_PROCESSING_MINUTES = 60L; + + /** + * 原生订单码和聚合活动会话连续为空时的最大数据库扫描间隔，单位毫秒。 + */ + 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..bd1cacfff 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 @@ -27,11 +27,7 @@ import com.bsd.pjcloud.thirdparty.model.pay.ThirdAggregateAlipayOauthIdentity; import com.bsd.pjcloud.thirdparty.model.pay.ThirdAggregateCashierLaunch; import com.bsd.pjcloud.thirdparty.model.pay.ThirdAggregateCashierPageData; import com.bsd.pjcloud.thirdparty.model.pay.ThirdAggregatePayCreateContext; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregateAliPayService; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; -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.*; import com.bsd.pjcloud.thirdparty.vo.appplus.pay.ThirdAggregateOrderCodePayStatusVo; import com.bsd.pjcloud.thirdparty.vo.appplus.pay.ThirdAggregateOrderCodePayVo; import com.bsd.pjcloud.wallet.api.entity.mch.MchPaySettingEntity; @@ -218,6 +214,11 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay */ private final StringEncryptor stringEncryptor; + /** + * 服务端轮询协调器，在聚合会话创建或复用成功后唤醒当前实例的自适应扫描任务。 + */ + private final ThirdPayPollingCoordinator pollingCoordinator; + /** * 进程级强随机数生成器，用于不可预测的聚合访问令牌。 */ @@ -239,14 +240,18 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay throw new RuntimeException("聚合订单码正在处理中，请勿重复提交"); } try { - return createOrReuse(dto); + ThirdAggregateOrderCodePayVo result = createOrReuse(dto); + pollingCoordinator.wakeAggregatePolling(); + return result; } finally { releaseLock(lockKey, lockToken); } } /** - * 查询当前租户聚合会话，并先同步已到达终态的第三方子订单。 + * 查询当前租户聚合会话，并在自动处理期限内同步第三方子订单终态。 + * + *

创建已满一小时的会话只执行本地关闭，不因客户端持续查询再次访问支付平台或恢复历史会话。

* * @param aggregateNo 聚合支付会话编号 * @return 最小聚合支付状态 @@ -260,7 +265,9 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay if (session == null) { throw new RuntimeException("聚合支付会话不存在"); } - if (isActive(session) && isExpired(session)) { + if (isBeyondAutoProcessingWindow(session)) { + processPendingSession(session.getId()); + } else if (isActive(session) && isExpired(session)) { processPendingSession(session.getId()); } else { syncSessionFromChildOrder(session); @@ -407,7 +414,8 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay * 处理一条待确认聚合会话或处于有限补偿窗口内的过期会话。 * *

活动会话到期后即使第三方查询或关单异常，也会在本轮结束时推进为过期状态，退出五秒高频队列； - * 后续仅由服务端低频补偿任务在限定窗口内重试，不依赖客户端继续查询。

+ * 后续仅由服务端低频补偿任务在限定窗口内重试。创建已满一小时后只执行本地关闭，不再访问平台， + * 因此客户端查询和创建重试均不能形成无限历史恢复。

* * @param sessionId 聚合支付会话主键 */ @@ -419,6 +427,10 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay if (session == null || (!isActive(session) && !expiredCompensation)) { return; } + if (isBeyondAutoProcessingWindow(session)) { + closeHistoricalSessionLocally(session); + return; + } if (ObjectUtil.equals(session.getStatus(), ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType())) { if (isExpired(session)) { aggregatePaySessionService.markTerminal(session.getId(), ThirdAggregatePaySessionStatusEnum.EXPIRED); @@ -457,7 +469,9 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay } /** - * 识别共享回调中的聚合 JSAPI 子订单并同步聚合会话，旧支付订单返回 {@code false}。 + * 识别共享回调中的聚合 JSAPI 子订单并在自动处理期限内同步聚合会话。 + * + *

超过一小时到达的延迟回调仍保留子订单真实支付事实，但聚合会话只标记关闭，不恢复为成功。

* * @param payOrder 已通过平台验签和关键字段校验的最新第三方支付订单 * @return 找到 pay_no 绑定聚合会话时返回 {@code true} @@ -471,6 +485,10 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay if (session == null) { return false; } + if (isBeyondAutoProcessingWindow(session)) { + closeHistoricalSessionLocally(session); + return true; + } aggregatePaySessionService.syncFromPayOrder(payOrder); return true; } @@ -664,6 +682,10 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay if (session == null || StrUtil.isBlank(session.getPayNo())) { return; } + if (isBeyondAutoProcessingWindow(session)) { + closeHistoricalSessionLocally(session); + return; + } aggregatePaySessionService.syncFromPayOrder(thirdPayOrderService.getByPayNo(session.getPayNo())); } @@ -671,7 +693,8 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay * 新建支付订单前确认已过期会话绑定的第三方子订单不再处于可支付状态。 * *

后台补偿超过十分钟后会停止，但创建新订单时仍会同步执行一次查单和关单；只有旧子订单已经成功、 - * 关闭、失败或不存在时才允许继续，避免旧单与新单同时可支付。

+ * 关闭、失败或不存在时才允许继续，避免旧单与新单同时可支付。创建已满一小时的会话只做本地关闭， + * 不再通过该入口访问平台或恢复会话状态。

* * @param session 同业务最新聚合会话，允许为空或非过期状态 */ @@ -688,7 +711,7 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay ThirdAggregatePaySessionEntity latestSession = aggregatePaySessionService.getById(session.getId()); ThirdPayOrderEntity payOrder = StrUtil.isBlank(session.getPayNo()) ? null : thirdPayOrderService.getByPayNo(session.getPayNo()); - if (payOrder != null) { + if (payOrder != null && !isBeyondAutoProcessingWindow(session)) { aggregatePaySessionService.syncFromPayOrder(payOrder); } if ((latestSession != null && ObjectUtil.equals(latestSession.getStatus(), @@ -954,6 +977,43 @@ public class ThirdAggregateOrderPayServiceImpl implements ThirdAggregateOrderPay return session == null || session.getExpireTime() == null || !session.getExpireTime().isAfter(LocalDateTime.now()); } + /** + * 判断聚合会话是否已经越过服务端自动处理硬截止。 + * + *

创建时间缺失的异常记录同样按越过截止处理，避免任何入口对无法确定年龄的订单持续查单。

+ * + * @param session 聚合支付会话 + * @return 创建时间为空或创建已满一小时时返回 {@code true} + */ + private boolean isBeyondAutoProcessingWindow(ThirdAggregatePaySessionEntity session) { + LocalDateTime cutoff = LocalDateTime.now() + .minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES); + return session == null || session.getCreateTime() == null || !session.getCreateTime().isAfter(cutoff); + } + + /** + * 在自动处理硬截止后仅将聚合会话及其未支付子订单标记为关闭。 + * + *

先关闭会话可阻止随后到达的子订单同步重新激活会话；子订单更新带未支付条件，不覆盖已经落库的 + * 真实成功状态。该方法不调用支付宝或微信接口。

+ * + * @param session 已越过自动处理硬截止的活动或过期会话 + */ + private void closeHistoricalSessionLocally(ThirdAggregatePaySessionEntity session) { + aggregatePaySessionService.markTerminal(session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + if (StrUtil.isBlank(session.getPayNo())) { + return; + } + ThirdPayOrderEntity payOrder = thirdPayOrderService.getByPayNo(session.getPayNo()); + if (payOrder == null) { + return; + } + thirdPayOrderService.update(new LambdaUpdateWrapper() + .eq(ThirdPayOrderEntity::getId, payOrder.getId()) + .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) + .set(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_CLOSED.getType())); + } + /** * 阻止同业务存在未归属于聚合会话的旧原生订单码或既有成功订单。 * 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/ThirdAggregateOrderPayJobTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java new file mode 100644 index 000000000..2de19b7c5 --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java @@ -0,0 +1,120 @@ +package com.bsd.pjcloud.thirdparty.job; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.bsd.pjcloud.common.data.tenant.TenantContextHolder; +import com.bsd.pjcloud.thirdparty.constant.pay.ThirdAggregatePaySessionStatusEnum; +import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; +import com.bsd.pjcloud.thirdparty.entity.pay.ThirdAggregatePaySessionEntity; +import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; +import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; +import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; +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.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +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.never; +import static org.mockito.Mockito.when; + +/** + * Thirdparty 聚合订单码历史终态和补偿窗口测试。 + */ +@ExtendWith(MockitoExtension.class) +class ThirdAggregateOrderPayJobTest { + + /** + * 聚合会话服务 Mock，用于返回超过一小时的历史会话并验证终态更新。 + */ + @Mock + private ThirdAggregatePaySessionService aggregatePaySessionService; + + /** + * 聚合支付编排服务 Mock，用于确认历史关闭路径不再调用支付平台处理流程。 + */ + @Mock + private ThirdAggregateOrderPayService aggregateOrderPayService; + + /** + * 第三方子订单服务 Mock；无子订单会话关闭时不应访问该服务。 + */ + @Mock + private ThirdPayOrderService thirdPayOrderService; + + /** + * 自适应轮询协调器 Mock，用于允许本轮补偿扫描。 + */ + @Mock + private ThirdPayPollingCoordinator pollingCoordinator; + + /** + * 被测 Thirdparty 聚合订单码任务。 + */ + @InjectMocks + private ThirdAggregateOrderPayJob job; + + /** + * 清理测试期间由 TenantBroker 设置的线程租户上下文。 + */ + @AfterEach + void tearDown() { + TenantContextHolder.setTenantId(null); + } + + /** + * 验证超过一小时且没有子订单的会话直接标记关闭，不再进入第三方平台查单流程。 + */ + @Test + void compensationShouldCloseHistoricalSessionWithoutCallingPaymentPlatform() { + ThirdAggregatePaySessionEntity session = new ThirdAggregatePaySessionEntity(); + session.setId(201L); + session.setTenantId(66L); + session.setStatus(ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()); + session.setCreateTime(LocalDateTime.now().minusHours(2)); + when(pollingCoordinator.tryStartAggregateCompensationPolling()).thenReturn(true); + when(aggregatePaySessionService.list(any(Wrapper.class))).thenReturn(List.of(session), List.of()); + + job.handleExpiredSessionCompensation(); + + verify(aggregatePaySessionService).markTerminal( + session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + verify(pollingCoordinator).finishAggregateCompensationPolling(true); + verifyNoInteractions(aggregateOrderPayService, thirdPayOrderService); + } + + /** + * 验证历史子订单即使已保存支付成功事实，也只关闭聚合会话而不执行历史状态恢复。 + */ + @Test + void compensationShouldNotRecoverHistoricalSessionFromSuccessfulChildOrder() { + ThirdAggregatePaySessionEntity session = new ThirdAggregatePaySessionEntity(); + session.setId(202L); + session.setTenantId(66L); + session.setPayNo("historical-child-pay-no"); + session.setStatus(ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()); + session.setCreateTime(LocalDateTime.now().minusHours(2)); + ThirdPayOrderEntity payOrder = new ThirdPayOrderEntity(); + payOrder.setId(302L); + payOrder.setPayNo(session.getPayNo()); + payOrder.setStatus(ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()); + when(pollingCoordinator.tryStartAggregateCompensationPolling()).thenReturn(true); + when(aggregatePaySessionService.list(any(Wrapper.class))).thenReturn(List.of(session), List.of()); + when(thirdPayOrderService.getByPayNo(session.getPayNo())).thenReturn(payOrder); + + job.handleExpiredSessionCompensation(); + + verify(aggregatePaySessionService).markTerminal( + session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + verify(aggregatePaySessionService, never()).syncFromPayOrder(any()); + verifyNoInteractions(aggregateOrderPayService); + } +} 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..d846fc439 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,38 @@ 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); + } + + /** + * 验证超过一小时的原生订单码只做本地关闭标记，不再调用支付平台查单流程。 + */ + @Test + void closeHistoricalOrderCodeOrdersShouldNotQueryPaymentPlatform() { + ThirdPayOrderEntity order = new ThirdPayOrderEntity(); + order.setId(100L); + order.setTenantId(66L); + order.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); + order.setCreateTime(LocalDateTime.now().minusHours(2)); + order.setExpireTime(LocalDateTime.now().minusHours(2)); + when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); + + job.closeHistoricalOrderCodeOrders(); + + verify(thirdPayOrderService).update(any(Wrapper.class)); + verifyNoInteractions(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..28074eaa9 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; @@ -44,6 +45,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; /** @@ -106,6 +108,12 @@ class ThirdAggregateOrderPayServiceImplTest { @Mock private StringEncryptor stringEncryptor; + /** + * 服务端轮询协调器 Mock，用于验证聚合会话创建后主动唤醒扫描任务。 + */ + @Mock + private ThirdPayPollingCoordinator pollingCoordinator; + /** * 被测聚合支付编排服务。 */ @@ -129,7 +137,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 +183,7 @@ class ThirdAggregateOrderPayServiceImplTest { Assertions.assertEquals(ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), savedSession.getStatus()); Assertions.assertNull(savedSession.getSelectedPlatform()); Assertions.assertNull(savedSession.getPayNo()); + verify(pollingCoordinator).wakeAggregatePolling(); } /** @@ -233,6 +242,7 @@ class ThirdAggregateOrderPayServiceImplTest { session.setStatus(ThirdAggregatePaySessionStatusEnum.PAYING.getType()); session.setSelectedPlatform(AppPlusPlatformTypeEnum.ALI_PAY.getType()); session.setPayNo("aggregate-child-pay-no"); + session.setCreateTime(LocalDateTime.now().minusMinutes(1)); session.setExpireTime(LocalDateTime.now().minusSeconds(1)); ThirdPayOrderEntity payOrder = new ThirdPayOrderEntity(); payOrder.setId(101L); @@ -249,6 +259,57 @@ class ThirdAggregateOrderPayServiceImplTest { verify(aggregatePaySessionService).markTerminal(session.getId(), ThirdAggregatePaySessionStatusEnum.EXPIRED); } + /** + * 验证超过一小时的会话从任何处理入口进入时都只做本地关闭，不访问支付宝或微信。 + */ + @Test + void historicalSessionShouldCloseLocallyWithoutQueryingPaymentPlatform() { + ThirdAggregatePaySessionEntity session = reusableSession(); + session.setStatus(ThirdAggregatePaySessionStatusEnum.PAYING.getType()); + session.setSelectedPlatform(AppPlusPlatformTypeEnum.ALI_PAY.getType()); + session.setPayNo("historical-child-pay-no"); + session.setCreateTime(LocalDateTime.now().minusHours(2)); + session.setExpireTime(LocalDateTime.now().minusHours(2)); + ThirdPayOrderEntity payOrder = new ThirdPayOrderEntity(); + payOrder.setId(102L); + payOrder.setPayNo(session.getPayNo()); + payOrder.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); + when(aggregatePaySessionService.getById(session.getId())).thenReturn(session); + when(thirdPayOrderService.getByPayNo(session.getPayNo())).thenReturn(payOrder); + + service.processPendingSession(session.getId()); + + verify(aggregatePaySessionService).markTerminal( + session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + verify(thirdPayOrderService).update(any(Wrapper.class)); + verifyNoInteractions(aggregateAliPayService, aggregateWechatPayService); + } + + /** + * 验证一小时后到达的成功回调只保留子订单事实，不把历史聚合会话恢复为支付成功。 + */ + @Test + void lateSuccessfulCallbackShouldNotRecoverHistoricalSession() { + ThirdAggregatePaySessionEntity session = reusableSession(); + session.setStatus(ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()); + session.setPayNo("late-success-pay-no"); + session.setCreateTime(LocalDateTime.now().minusHours(2)); + session.setExpireTime(LocalDateTime.now().minusHours(2)); + ThirdPayOrderEntity payOrder = new ThirdPayOrderEntity(); + payOrder.setId(103L); + payOrder.setPayNo(session.getPayNo()); + payOrder.setStatus(ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()); + when(aggregatePaySessionService.getByPayNo(payOrder.getPayNo())).thenReturn(session); + when(thirdPayOrderService.getByPayNo(payOrder.getPayNo())).thenReturn(payOrder); + + Assertions.assertTrue(service.syncAggregateChildOrder(payOrder)); + + verify(aggregatePaySessionService).markTerminal( + session.getId(), ThirdAggregatePaySessionStatusEnum.PAY_CLOSED); + verify(aggregatePaySessionService, never()).syncFromPayOrder(any()); + verifyNoInteractions(aggregateAliPayService, aggregateWechatPayService); + } + /** * 构建支付宝和微信全局应用配置，公网地址固定使用 HTTPS。 * 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