feat：outbox表app消息推送优化

2026-10-09 17:52:19 +08:00
1 parent 9f289b31a3
commit 1df9c07db3
@@ -0,0 +1,76 @@
+-- 20261009 管理端订单分页索引 + 钱包流水业务号索引（v2 修正版，可重复执行）
+--
+-- 背景：/hard-order/page-join 分页（scopePageIds：tenant_id+del_flag 等值过滤，默认 order by create_time desc, user_id desc, id desc）
+--       线上 hard_order 没有任何以 create_time 为前导的组合索引，EXPLAIN 实测走 idx_hard_order_expired_refund_scan 的
+--       tenant_id 前缀扫约 3.9 万行 + Using filesort，且每次翻页还要再跑一遍同款 COUNT。
+--       另：订单/售后按 biz_id 查钱包流水目前无索引可用，需全表扫约 192 万行。
+-- v2 修正（重要）：v1 版本把 merchant_dept_id 放在 user_id 之后、隐式主键之前，拆断了
+--       (create_time, user_id, id) 连续排序元组，FORCE INDEX 实测仍然 Using filesort。
+--       v2 显式纳入 id 并把 merchant_dept_id 挪到 id 之后：排序列 (create_time, user_id, id) 连续且同向，
+--       等值前缀后反向扫描即可免排序；id 唯一，其后的 merchant_dept_id 不影响顺序，仅作覆盖列。
+--       本脚本会先删除 v1 索引（若存在）再创建 v2。
+-- 执行：均为 Online DDL（INPLACE + LOCK=NONE，不锁表）；hard_order 7.8 万行秒级；
+--       wallet_transaction 192 万行预计 10~60 秒，建议低峰执行。可重复执行。
+-- 回滚：文末 DROP INDEX 语句。
+
+-- ① hard_order 管理端分页索引 v2（含 v1 清理）
+USE pjcloudx_hardware;
+-- 注意：GROUP_CONCAT 是聚合函数，只能出现在子查询 SELECT 列表（标量子查询）中做索引列序检测，
+-- 写进 WHERE 会报 1111 Invalid use of group function。
+SET @ddl = IF(
+    (SELECT GROUP_CONCAT(column_name ORDER BY seq_in_index)
+     FROM information_schema.statistics
+     WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_order'
+       AND index_name = 'idx_hard_order_page_create') = 'tenant_id,del_flag,create_time,user_id,id,merchant_dept_id',
+    'SELECT ''skip: idx_hard_order_page_create 已是 v2 版本'' AS msg',
+    IF(
+        (SELECT COUNT(*) FROM information_schema.statistics
+         WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_order'
+           AND index_name = 'idx_hard_order_page_create') = 0,
+        'ALTER TABLE hard_order ADD INDEX idx_hard_order_page_create (tenant_id, del_flag, create_time, user_id, id, merchant_dept_id), ALGORITHM=INPLACE, LOCK=NONE',
+        'ALTER TABLE hard_order DROP INDEX idx_hard_order_page_create, ADD INDEX idx_hard_order_page_create (tenant_id, del_flag, create_time, user_id, id, merchant_dept_id), ALGORITHM=INPLACE, LOCK=NONE'));
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;
+-- 重建后刷新统计信息，确保优化器立即采用新索引
+ANALYZE TABLE hard_order;
+
+-- ② wallet_transaction 业务号点查索引
+USE pjcloudx_wallet;
+SET @ddl = IF(
+    (SELECT COUNT(*) FROM information_schema.statistics
+     WHERE table_schema = 'pjcloudx_wallet' AND table_name = 'wallet_transaction'
+       AND index_name = 'idx_wallet_tx_tenant_biz_id') = 0,
+    'ALTER TABLE wallet_transaction ADD INDEX idx_wallet_tx_tenant_biz_id (tenant_id, biz_id), ALGORITHM=INPLACE, LOCK=NONE',
+    'SELECT ''skip: idx_wallet_tx_tenant_biz_id 已存在'' AS msg');
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;
+
+-- ③ hard_order App端默认排序分页索引：App 订单列表（appVisibilityPageIds）默认
+--    order by appt_use_date desc, pay_time desc, id desc，此前无任何索引可用，
+--    线上实测主查询与 COUNT 各约 0.58 秒（全范围扫描 + filesort）。
+--    方向匹配索引让主查询沿索引倒序取前 10 行即停；COUNT 建议由 App 端传 searchCount=false 免除。
+USE pjcloudx_hardware;
+SET @ddl = IF(
+    (SELECT COUNT(*) FROM information_schema.statistics
+     WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_order'
+       AND index_name = 'idx_hard_order_app_page') = 0,
+    'ALTER TABLE hard_order ADD INDEX idx_hard_order_app_page (tenant_id, del_flag, appt_use_date DESC, pay_time DESC, id DESC), ALGORITHM=INPLACE, LOCK=NONE',
+    'SELECT ''skip: idx_hard_order_app_page 已存在'' AS msg');
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;
+ANALYZE TABLE hard_order;
+
+-- ============ 执行后验证（期望 key=idx_hard_order_page_create 且 Extra 无 Using filesort） ============
+-- EXPLAIN SELECT id, merchant_dept_id FROM pjcloudx_hardware.hard_order
+--  WHERE tenant_id = 1 AND del_flag = '0'
+--  ORDER BY create_time DESC, user_id DESC, id DESC LIMIT 10;
+--
+-- EXPLAIN SELECT id FROM pjcloudx_wallet.wallet_transaction
+--  WHERE tenant_id = 1 AND biz_id = '2108427023102214146';
+
+-- ============ 回滚（如需） ============
+-- ALTER TABLE pjcloudx_hardware.hard_order DROP INDEX idx_hard_order_page_create;
+-- ALTER TABLE pjcloudx_wallet.wallet_transaction DROP INDEX idx_wallet_tx_tenant_biz_id;
@@ -0,0 +1,33 @@
+-- 20261009 钱包账户管理分页索引（可重复执行）
+--
+-- 背景：/wallet/wallet-manage/page-join 分页查询（scopePage：tenant_id+del_flag 等值过滤，
+--       默认 order by create_time desc, user_id desc, wallet_type_id asc, id desc）在
+--       wallet_manage 上没有任何以 tenant_id 为前导的索引（现有：idx_dept_id(user_id)、
+--       idx_wallet_manage_card_tenant_del(card_number,tenant_id,del_flag)、
+--       idx_wallet_manage_id_tenant(id,tenant_id)、idx_wallet_manage_wallet_type_dept(wallet_type_id,dept_id)），
+--       EXPLAIN 实测 type=ALL 全表扫描 + Using filesort；线上 21.9 万行单次翻页约 28 秒（含 COUNT 复扫）。
+-- 方案：新增与 ORDER BY 列方向完全一致的分页索引。MySQL 8.0 起支持降序索引，
+--       等值前缀 (tenant_id, del_flag) 之后 (create_time DESC, user_id DESC, wallet_type_id ASC, id DESC)
+--       与查询排序逐列同向，优化器可直接沿索引取前 10 行，filesort 消失；COUNT 走同一索引前缀区间。
+-- 执行：Online DDL（INPLACE + LOCK=NONE，不锁表）；线上 21.9 万行预计 5~20 秒，建议低峰执行。
+--       可重复执行（information_schema 判幂等，已存在则跳过）。
+-- 回滚：文末 DROP INDEX。
+
+USE pjcloudx_wallet;
+SET @ddl = IF(
+    (SELECT COUNT(*) FROM information_schema.statistics
+     WHERE table_schema = 'pjcloudx_wallet' AND table_name = 'wallet_manage'
+       AND index_name = 'idx_wallet_manage_page_list') = 0,
+    'ALTER TABLE wallet_manage ADD INDEX idx_wallet_manage_page_list (tenant_id, del_flag, create_time DESC, user_id DESC, wallet_type_id ASC, id DESC), ALGORITHM=INPLACE, LOCK=NONE',
+    'SELECT ''skip: idx_wallet_manage_page_list 已存在'' AS msg');
+PREPARE stmt FROM @ddl;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;
+
+-- ============ 执行后验证（期望 key=idx_wallet_manage_page_list，type=ref，Extra 无 Using filesort） ============
+-- EXPLAIN SELECT id FROM pjcloudx_wallet.wallet_manage
+--  WHERE tenant_id = 1 AND del_flag = '0'
+--  ORDER BY create_time DESC, user_id DESC, wallet_type_id ASC, id DESC LIMIT 10;
+
+-- ============ 回滚（如需） ============
+-- ALTER TABLE pjcloudx_wallet.wallet_manage DROP INDEX idx_wallet_manage_page_list;
@@ -1,6 +1,6 @@
 # 线上
-# MYSQL_HOST=123.157.250.162
-# MYSQL_PORT=28815
+# MYSQL_HOST=192.168.0.100
+# MYSQL_PORT=3306
 # 测试
 MYSQL_HOST=123.157.250.162
 MYSQL_PORT=28816
@@ -31,6 +31,7 @@ public class WalletTransactionEntity extends Model<WalletTransactionEntity> {
    @Schema(description = "编号")
    private Long id;

+
    /**
     * 用户 id
     */
@@ -33,9 +33,13 @@ public class WalletMessageOutboxJob {
    }

    /**
-     * 每日凌晨分批清理超过保留期的终态Outbox，本轮数量受服务层限制。
+     * 每小时第20分钟分批清理超过保留期的终态Outbox，本轮数量受服务层限制。
+     *
+     * <p>历史为每日一次、服务层每批1000行，删除吞吐低于流水高峰日增量，
+     * wallet_message_outbox 曾因此增长到百万行级并放大扫描成本；现改为每小时执行、
+     * 服务层单批5000行，删除吞吐提升至约24万行/日。</p>
     */
-    @Scheduled(cron = "0 20 2 * * ?")
+    @Scheduled(cron = "0 20 * * * ?")
    public void cleanupExpiredEvents() {
        try {
            int deleted = outboxService.cleanupExpiredEvents();
@@ -1,7 +1,10 @@
 package com.bsd.pjcloud.wallet.service.impl;

+import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.bsd.pjcloud.admin.api.entity.SysTenant;
+import com.bsd.pjcloud.admin.api.feign.RemoteTenantService;
 import com.bsd.pjcloud.common.core.util.R;
 import com.bsd.pjcloud.common.data.tenant.TenantBroker;
 import com.bsd.pjcloud.thirdparty.api.feign.RemoterThirdAppMessageService;
@@ -18,6 +21,7 @@ import org.springframework.transaction.support.TransactionTemplate;
 import java.time.LocalDateTime;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;

 /**
 * 钱包业务消息Outbox处理实现，以稳定聚合事件ID向Thirdparty进行至少一次投递。
@@ -54,8 +58,8 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
    /** 单轮跨租户最多扫描的候选Outbox行数。 */
    private static final int SCAN_LIMIT = 100;

-    /** 单轮清理每种终态最多物理删除的记录数量。 */
-    private static final int CLEANUP_BATCH_SIZE = 1000;
+    /** 单轮清理每种终态最多物理删除的记录数量；清理任务每小时执行，按 5000/小时 提供约 12 万/日 的删除吞吐，覆盖流水高峰日增量。 */
+    private static final int CLEANUP_BATCH_SIZE = 5000;

    /** 成功Outbox保留天数。 */
    private static final long SUCCESS_RETENTION_DAYS = 30L;
@@ -63,6 +67,9 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
    /** 失败终态Outbox保留天数。 */
    private static final long FAILED_RETENTION_DAYS = 365L;

+    /** 租户ID列表的本地缓存有效期，单位毫秒；扫描任务每10秒触发，缓存避免每次远程查询租户列表。 */
+    private static final long TENANT_CACHE_TTL_MS = 60_000L;
+
    /** 八次尝试后续使用的退避分钟数，索引与已完成尝试次数对应。 */
    private static final List<Long> RETRY_INTERVAL_MINUTES = List.of(1L, 5L, 15L, 60L, 180L, 360L, 720L);

@@ -72,11 +79,26 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
    /** Thirdparty业务消息中心内部Feign客户端。 */
    private final RemoterThirdAppMessageService appMessageService;

+    /** UPMS租户内部查询客户端，扫描前获取全部租户用于按租户切分查询。 */
+    private final RemoteTenantService tenantService;
+
    /** 事务模板，保证条件领取、远程结果和本组状态变更形成清晰的本地事务边界。 */
    private final TransactionTemplate transactionTemplate;

+    /** 最近一次成功获取的租户ID列表缓存；远程查询失败时退回该缓存继续扫描。 */
+    private volatile List<Long> cachedTenantIds = List.of();
+
+    /** 最近一次刷新租户ID缓存的时间戳，单位 System.currentTimeMillis() 毫秒。 */
+    private volatile long tenantCacheRefreshedAt = 0L;
+
    /**
-     * 扫描到期候选，并按每条记录持久化的租户恢复数据库与Feign上下文。
+     * 按租户循环扫描到期候选并逐条在租户上下文内处理。
+     *
+     * <p>历史实现构造不带租户条件的跨租户查询，而扫描索引 (tenant_id, status, ...) 以
+     * tenant_id 为前导列，导致每次扫描退化为全表扫描；表增长到百万行后每10秒一次的
+     * 全表扫描持续占用IO，拖慢整个数据库实例。现按租户切分查询：租户上下文内的
+     * MyBatis-Plus租户拦截器自动补充 tenant_id 等值条件，扫描命中索引前缀，单租户
+     * 仍保持 SCAN_LIMIT 上限，整体语义与单租户轮询一致。</p>
     *
     * @return 本轮成功领取的聚合事件数量
     */
@@ -85,26 +107,44 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime mergeCutoff = now.minusSeconds(MERGE_WINDOW_SECONDS);
        LocalDateTime processingCutoff = now.minusMinutes(PROCESSING_TIMEOUT_MINUTES);
+        int processed = 0;
+        for (Long tenantId : resolveTenantIds()) {
+            try {
+                Integer tenantProcessed = TenantBroker.applyAs(tenantId, tenant ->
+                        processDueEventsInTenant(mergeCutoff, processingCutoff));
+                if (tenantProcessed != null) {
+                    processed += tenantProcessed;
+                }
+            } catch (Exception exception) {
+                log.error("钱包消息Outbox租户扫描异常,tenantId:{}", tenantId, exception);
+            }
+        }
+        return processed;
+    }
+
+    /**
+     * 在当前租户上下文内扫描并处理到期候选；查询由租户拦截器自动附加 tenant_id 条件以命中扫描索引。
+     *
+     * @param mergeCutoff 本轮2秒归并截止时间
+     * @param processingCutoff PROCESSING超时恢复截止时间
+     * @return 本租户成功领取的聚合事件数量
+     */
+    private Integer processDueEventsInTenant(LocalDateTime mergeCutoff, LocalDateTime processingCutoff) {
        List<WalletMessageOutboxEntity> candidates = outboxMapper.selectList(
                Wrappers.<WalletMessageOutboxEntity>lambdaQuery()
                        .and(wrapper -> wrapper
                                .eq(WalletMessageOutboxEntity::getStatus, STATUS_PENDING)
                                .le(WalletMessageOutboxEntity::getCreateTime, mergeCutoff)
                                .or(current -> current.eq(WalletMessageOutboxEntity::getStatus, STATUS_RETRY)
-                                        .le(WalletMessageOutboxEntity::getNextRetryTime, now))
+                                        .le(WalletMessageOutboxEntity::getNextRetryTime, LocalDateTime.now()))
                                .or(current -> current.eq(WalletMessageOutboxEntity::getStatus, STATUS_PROCESSING)
                                        .le(WalletMessageOutboxEntity::getUpdateTime, processingCutoff)))
                        .orderByAsc(WalletMessageOutboxEntity::getCreateTime)
                        .last("limit " + SCAN_LIMIT));
        int processed = 0;
        for (WalletMessageOutboxEntity candidate : candidates) {
-            if (candidate.getTenantId() == null) {
-                log.warn("钱包消息Outbox缺少租户ID,outboxId:{}", candidate.getId());
-                continue;
-            }
            try {
-                Boolean claimed = TenantBroker.applyAs(candidate.getTenantId(), tenantId ->
-                        transactionTemplate.execute(status -> processGroup(candidate.getId(), mergeCutoff, processingCutoff)));
+                Boolean claimed = transactionTemplate.execute(status -> processGroup(candidate.getId(), mergeCutoff, processingCutoff));
                if (Boolean.TRUE.equals(claimed)) {
                    processed++;
                }
@@ -115,6 +155,36 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
        return processed;
    }

+    /**
+     * 返回待扫描的租户ID列表，带60秒本地缓存；远程查询失败时退回上一次缓存。
+     *
+     * <p>扫描任务每10秒触发一次，逐次远程查询租户列表会产生日均近万次无效调用；
+     * 租户新增最迟在缓存过期后一轮扫描内被纳入。远程失败且无缓存时返回空列表，
+     * 本轮扫描跳过，等待下一轮重试，不中断调度线程。</p>
+     *
+     * @return 当前需要扫描的租户ID列表，可能为空
+     */
+    List<Long> resolveTenantIds() {
+        long now = System.currentTimeMillis();
+        List<Long> cached = cachedTenantIds;
+        if (!cached.isEmpty() && now - tenantCacheRefreshedAt < TENANT_CACHE_TTL_MS) {
+            return cached;
+        }
+        try {
+            R<List<SysTenant>> result = tenantService.list();
+            List<SysTenant> tenants = result == null || !result.isOk() ? null : result.getData();
+            if (ObjectUtil.isNotEmpty(tenants)) {
+                List<Long> tenantIds = tenants.stream().map(SysTenant::getId).filter(Objects::nonNull).toList();
+                cachedTenantIds = tenantIds;
+                tenantCacheRefreshedAt = now;
+                return tenantIds;
+            }
+        } catch (Exception exception) {
+            log.warn("钱包消息Outbox查询租户列表失败,退回缓存继续,缓存租户数:{}", cached.size(), exception);
+        }
+        return cached;
+    }
+
    /**
     * 分批物理删除到期终态记录，避免逻辑删除继续占用保留空间。
     *
@@ -0,0 +1,77 @@
+package com.bsd.pjcloud.wallet.service.impl;
+
+import com.bsd.pjcloud.admin.api.entity.SysTenant;
+import com.bsd.pjcloud.admin.api.feign.RemoteTenantService;
+import com.bsd.pjcloud.common.core.util.R;
+import com.bsd.pjcloud.thirdparty.api.feign.RemoterThirdAppMessageService;
+import com.bsd.pjcloud.wallet.mapper.WalletMessageOutboxMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.List;
+
+/**
+ * 钱包消息Outbox租户解析测试：验证扫描按租户切分后的租户列表缓存与远程失败兜底行为。
+ */
+@ExtendWith(MockitoExtension.class)
+class WalletMessageOutboxServiceImplTest {
+
+    /** Outbox Mapper Mock。 */
+    @Mock
+    private WalletMessageOutboxMapper outboxMapper;
+
+    /** Thirdparty业务消息Feign Mock。 */
+    @Mock
+    private RemoterThirdAppMessageService appMessageService;
+
+    /** UPMS租户Feign Mock。 */
+    @Mock
+    private RemoteTenantService tenantService;
+
+    /** 事务模板Mock。 */
+    @Mock
+    private TransactionTemplate transactionTemplate;
+
+    /** 被测服务。 */
+    @InjectMocks
+    private WalletMessageOutboxServiceImpl service;
+
+    /**
+     * 构造指定ID的租户对象。
+     *
+     * @param id 租户ID
+     * @return 租户实体
+     */
+    private SysTenant tenant(Long id) {
+        SysTenant sysTenant = new SysTenant();
+        sysTenant.setId(id);
+        return sysTenant;
+    }
+
+    /**
+     * 验证首次解析发起远程查询并缓存租户ID，缓存有效期内不重复远程调用。
+     */
+    @Test
+    void resolveTenantIdsShouldCacheRemoteResult() {
+        Mockito.when(tenantService.list()).thenReturn(R.ok(List.of(tenant(1L), tenant(2L))));
+        Assertions.assertEquals(List.of(1L, 2L), service.resolveTenantIds());
+        // 第二次在缓存有效期内直接命中本地缓存，不再触发远程调用
+        Assertions.assertEquals(List.of(1L, 2L), service.resolveTenantIds());
+        Mockito.verify(tenantService, Mockito.times(1)).list();
+    }
+
+    /**
+     * 验证无缓存时远程查询异常返回空列表跳过本轮，不向调度线程抛出异常。
+     */
+    @Test
+    void resolveTenantIdsShouldReturnEmptyWhenRemoteFailsWithoutCache() {
+        Mockito.when(tenantService.list()).thenThrow(new IllegalStateException("remote unavailable"));
+        Assertions.assertTrue(service.resolveTenantIds().isEmpty());
+    }
+}