diff --git a/db/20261009_add_order_page_and_wallet_biz_index.sql b/db/20261009_add_order_page_and_wallet_biz_index.sql new file mode 100644 index 00000000..6b653367 --- /dev/null +++ b/db/20261009_add_order_page_and_wallet_biz_index.sql @@ -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; diff --git a/db/20261009_add_third_push_delivery_page_index.sql b/db/20261009_add_third_push_delivery_page_index.sql new file mode 100644 index 00000000..7a3b52ac --- /dev/null +++ b/db/20261009_add_third_push_delivery_page_index.sql @@ -0,0 +1,31 @@ +-- 20261009 商户投递记录分页索引（可重复执行） +-- +-- 背景：/app-message/pc/deliveries 页面查询（source_mch_id + recipient_user_id 过滤， +-- create_time desc, id desc 排序）在 third_push_delivery 上仅有 +-- (tenant_id, source_mch_id, create_time, del_flag) 索引，recipient_user_id +-- 不在索引中，需沿商户范围逐行回表过滤；表增长到约 21 万行后实测 COUNT 约 2.2 秒。 +-- 方案：新增与页面查询形态精确匹配的复合索引，等值(mch, recipient)后按 create_time 有序， +-- 分页与 COUNT 均为索引前缀范围扫描；无 recipient 条件时退回既有商户时间索引，不受影响。 +-- 执行：Online DDL（INPLACE + LOCK=NONE，不锁表），21 万行预计 5~15 秒，建议低峰执行。 +-- 回滚：文末 DROP INDEX。 + +USE pjcloudx_thirdparty; +SET @ddl = IF( + (SELECT COUNT(*) FROM information_schema.statistics + WHERE table_schema = 'pjcloudx_thirdparty' AND table_name = 'third_push_delivery' + AND index_name = 'idx_third_push_delivery_mch_recipient') = 0, + 'ALTER TABLE third_push_delivery ADD INDEX idx_third_push_delivery_mch_recipient (tenant_id, source_mch_id, recipient_user_id, create_time, id), ALGORITHM=INPLACE, LOCK=NONE', + 'SELECT ''skip: idx_third_push_delivery_mch_recipient 已存在'' AS msg'); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; +ANALYZE TABLE third_push_delivery; + +-- ============ 执行后验证（期望 key=idx_third_push_delivery_mch_recipient） ============ +-- EXPLAIN SELECT id FROM pjcloudx_thirdparty.third_push_delivery +-- WHERE tenant_id = 1 AND source_mch_id = 2078051759222632449 +-- AND recipient_user_id = 2034824566572941314 +-- ORDER BY create_time DESC, id DESC LIMIT 20; + +-- ============ 回滚（如需） ============ +-- ALTER TABLE pjcloudx_thirdparty.third_push_delivery DROP INDEX idx_third_push_delivery_mch_recipient; diff --git a/db/20261009_add_wallet_manage_page_index.sql b/db/20261009_add_wallet_manage_page_index.sql new file mode 100644 index 00000000..0da34980 --- /dev/null +++ b/db/20261009_add_wallet_manage_page_index.sql @@ -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; diff --git a/db/20261010_add_xfj_log_card_convert.sql b/db/20261010_add_xfj_log_card_convert.sql new file mode 100644 index 00000000..147cfc34 --- /dev/null +++ b/db/20261010_add_xfj_log_card_convert.sql @@ -0,0 +1,56 @@ +-- 20261010 消费机日志记录卡号转换信息（可重复执行） +-- +-- 背景：部分商户（首个 2078051759222632449）的消费机上报 10 位十进制卡号，服务端按 +-- “十进制 → 8 位大写十六进制 → 字节反转”规则换算成库内卡号后查询钱包 +-- （如 2506198930 → 95618F92 → 928F6195）；排查对账需要在日志表直接看到 +-- 转换前后卡号与转换方式，而不是从 req 原文里人工换算。 +-- 方案：hard_xfj_log 新增三列，均可空，仅刷卡综合接口（consumTransactions）日志填充： +-- source_card_no 转换前卡号（设备上报原始卡号） +-- target_card_no 转换后卡号（库内查询卡号，未转换时与转换前一致） +-- card_convert_type 卡号转换方式（NONE=未转换，DECIMAL_UID_REVERSE=十进制转十六进制字节反转） +-- 执行：Online DDL，小字段变更，秒级完成。 +-- 回滚：文末 DROP COLUMN。 + +USE pjcloudx_hardware; + +SET @ddl = IF( + (SELECT COUNT(*) FROM information_schema.columns + WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_xfj_log' + AND column_name = 'source_card_no') = 0, + 'ALTER TABLE hard_xfj_log ADD COLUMN source_card_no varchar(64) DEFAULT NULL COMMENT ''转换前卡号(设备上报原始卡号)'' AFTER req', + 'SELECT ''skip: source_card_no 已存在'' AS msg'); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + (SELECT COUNT(*) FROM information_schema.columns + WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_xfj_log' + AND column_name = 'target_card_no') = 0, + 'ALTER TABLE hard_xfj_log ADD COLUMN target_card_no varchar(64) DEFAULT NULL COMMENT ''转换后卡号(库内查询卡号,未转换时与转换前一致)'' AFTER source_card_no', + 'SELECT ''skip: target_card_no 已存在'' AS msg'); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +SET @ddl = IF( + (SELECT COUNT(*) FROM information_schema.columns + WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_xfj_log' + AND column_name = 'card_convert_type') = 0, + 'ALTER TABLE hard_xfj_log ADD COLUMN card_convert_type varchar(32) DEFAULT NULL COMMENT ''卡号转换方式(NONE=未转换,DECIMAL_UID_REVERSE=十进制转十六进制字节反转)'' AFTER target_card_no', + 'SELECT ''skip: card_convert_type 已存在'' AS msg'); +PREPARE stmt FROM @ddl; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; + +-- ============ 执行后验证（期望三列均为 YES） ============ +-- SELECT column_name, data_type, character_maximum_length +-- FROM information_schema.columns +-- WHERE table_schema = 'pjcloudx_hardware' AND table_name = 'hard_xfj_log' +-- AND column_name IN ('source_card_no', 'target_card_no', 'card_convert_type'); + +-- ============ 回滚（如需） ============ +-- ALTER TABLE hard_xfj_log +-- DROP COLUMN source_card_no, +-- DROP COLUMN target_card_no, +-- DROP COLUMN card_convert_type; diff --git a/local.env b/local.env index 1cb6967d..d909bfb2 100644 --- a/local.env +++ b/local.env @@ -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 diff --git a/pjcloud-hardware/pjcloud-hardware-api/src/main/java/com/bsd/pjcloud/hard/api/entity/xfj/HardXfjLogEntity.java b/pjcloud-hardware/pjcloud-hardware-api/src/main/java/com/bsd/pjcloud/hard/api/entity/xfj/HardXfjLogEntity.java index 0b135b33..133e712a 100755 --- a/pjcloud-hardware/pjcloud-hardware-api/src/main/java/com/bsd/pjcloud/hard/api/entity/xfj/HardXfjLogEntity.java +++ b/pjcloud-hardware/pjcloud-hardware-api/src/main/java/com/bsd/pjcloud/hard/api/entity/xfj/HardXfjLogEntity.java @@ -68,6 +68,24 @@ public class HardXfjLogEntity extends Model { @Schema(description = "error") private String error; + /** + * 转换前卡号(设备上报原始卡号) + */ + @Schema(description = "转换前卡号(设备上报原始卡号)") + private String sourceCardNo; + + /** + * 转换后卡号(库内查询卡号,未转换时与转换前一致) + */ + @Schema(description = "转换后卡号(库内查询卡号,未转换时与转换前一致)") + private String targetCardNo; + + /** + * 卡号转换方式(NONE=未转换,DECIMAL_UID_REVERSE=十进制转十六进制字节反转) + */ + @Schema(description = "卡号转换方式(NONE=未转换,DECIMAL_UID_REVERSE=十进制转十六进制字节反转)") + private String cardConvertType; + /** * 开始时间 */ diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/xfj/HardXfjServiceImpl.java b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/xfj/HardXfjServiceImpl.java index a8c6403b..797c96ab 100644 --- a/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/xfj/HardXfjServiceImpl.java +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/java/com/bsd/pjcloud/hard/service/impl/xfj/HardXfjServiceImpl.java @@ -50,6 +50,7 @@ import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -73,6 +74,42 @@ public class HardXfjServiceImpl implements HardXfjService { private final StringRedisTemplate redisTemplate; + /** + * 十进制 UID 商户:这些商户的消费机读出 10 位十进制卡号,需按 + * 十进制→8 位大写十六进制→字节反转规则换算成库内卡号后查询; + * 如 2506198930 → 95618F92 → 928F6195。后续商户往集合里追加即可。 + */ + private static final Set DECIMAL_UID_MERCHANT_IDS = Set.of(2078051759222632449L); + + /** 卡号转换方式:未转换(原卡号直查)。 */ + private static final String CARD_CONVERT_TYPE_NONE = "NONE"; + + /** 卡号转换方式:十进制转 8 位大写十六进制后按字节反转。 */ + private static final String CARD_CONVERT_TYPE_DECIMAL_UID_REVERSE = "DECIMAL_UID_REVERSE"; + + /** + * 按商户规则换算查询用卡号:命中映射商户且卡号为纯数字时, + * 十进制转 8 位大写十六进制后按 2 字符一组倒序拼接(字节反转); + * 未命中映射、卡号非纯数字或超出 4 字节 UID 范围时原样返回, + * 保持未配置商户的原有查询行为。 + */ + private static String resolveQueryCardNumber(String cardNo, Long merchantId) { + if (ObjectUtil.isEmpty(merchantId) || !DECIMAL_UID_MERCHANT_IDS.contains(merchantId) + || ObjectUtil.isEmpty(cardNo) || !cardNo.matches("\\d+")) { + return cardNo; + } + long value = Long.parseLong(cardNo); + if (value > 0xFFFFFFFFL) { + return cardNo; + } + String hex = String.format("%08X", value); + StringBuilder reversed = new StringBuilder(8); + for (int index = 6; index >= 0; index -= 2) { + reversed.append(hex, index, index + 2); + } + return reversed.toString(); + } + /** * 检查设备并返回配置 */ @@ -260,6 +297,12 @@ public class HardXfjServiceImpl implements HardXfjService { if (!ObjectUtil.isAllNotEmpty(dto.getMode(), dto.getCardNo(), dto.getCardMode(), dto.getOrder())) { throw new RuntimeException("解析参数错误"); } + // 按商户规则换算查询用卡号,并把转换前后卡号与转换方式记录到日志表 + String queryCardNo = resolveQueryCardNumber(dto.getCardNo(), entity.getMerchantId()); + logger.setSourceCardNo(dto.getCardNo()); + logger.setTargetCardNo(queryCardNo); + logger.setCardConvertType(ObjectUtil.equals(queryCardNo, dto.getCardNo()) + ? CARD_CONVERT_TYPE_NONE : CARD_CONVERT_TYPE_DECIMAL_UID_REVERSE); // 刷卡扣费 if (ObjectUtil.equals(dto.getMode(), 0)) { // 取餐 @@ -267,7 +310,7 @@ public class HardXfjServiceImpl implements HardXfjService { return null; } else { - WalletManageEntity wallet = walletService.getWalletByCardNumber(dto.getCardNo()).getData(); + WalletManageEntity wallet = walletService.getWalletByCardNumber(queryCardNo).getData(); if (ObjectUtil.isEmpty(wallet)) { throw new RuntimeException("非本单位饭卡"); } @@ -314,7 +357,7 @@ public class HardXfjServiceImpl implements HardXfjService { } // 余额查询 else if (ObjectUtil.equals(dto.getMode(), 2)) { - WalletManageEntity wallet = walletService.getWalletByCardNumber(dto.getCardNo()).getData(); + WalletManageEntity wallet = walletService.getWalletByCardNumber(queryCardNo).getData(); if (ObjectUtil.isEmpty(wallet)) { throw new RuntimeException("非本单位饭卡"); } diff --git a/pjcloud-hardware/pjcloud-hardware-biz/src/main/resources/mapper/xfj/HardXfjLogMapper.xml b/pjcloud-hardware/pjcloud-hardware-biz/src/main/resources/mapper/xfj/HardXfjLogMapper.xml index 4d261b46..1444cba3 100755 --- a/pjcloud-hardware/pjcloud-hardware-biz/src/main/resources/mapper/xfj/HardXfjLogMapper.xml +++ b/pjcloud-hardware/pjcloud-hardware-biz/src/main/resources/mapper/xfj/HardXfjLogMapper.xml @@ -10,6 +10,9 @@ a.headers, a.res, a.req, + a.source_card_no, + a.target_card_no, + a.card_convert_type, a.start_time, a.end_time, a.cost, diff --git a/pjcloud-wallet/pjcloud-wallet-api/src/main/java/com/bsd/pjcloud/wallet/api/entity/WalletTransactionEntity.java b/pjcloud-wallet/pjcloud-wallet-api/src/main/java/com/bsd/pjcloud/wallet/api/entity/WalletTransactionEntity.java index 78719162..0a7012f4 100644 --- a/pjcloud-wallet/pjcloud-wallet-api/src/main/java/com/bsd/pjcloud/wallet/api/entity/WalletTransactionEntity.java +++ b/pjcloud-wallet/pjcloud-wallet-api/src/main/java/com/bsd/pjcloud/wallet/api/entity/WalletTransactionEntity.java @@ -31,6 +31,7 @@ public class WalletTransactionEntity extends Model { @Schema(description = "编号") private Long id; + /** * 用户 id */ diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/job/WalletMessageOutboxJob.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/job/WalletMessageOutboxJob.java index 1ecfe7f7..996ad0cf 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/job/WalletMessageOutboxJob.java +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/job/WalletMessageOutboxJob.java @@ -33,9 +33,13 @@ public class WalletMessageOutboxJob { } /** - * 每日凌晨分批清理超过保留期的终态Outbox，本轮数量受服务层限制。 + * 每小时第20分钟分批清理超过保留期的终态Outbox，本轮数量受服务层限制。 + * + *

历史为每日一次、服务层每批1000行，删除吞吐低于流水高峰日增量， + * wallet_message_outbox 曾因此增长到百万行级并放大扫描成本；现改为每小时执行、 + * 服务层单批5000行，删除吞吐提升至约24万行/日。

*/ - @Scheduled(cron = "0 20 2 * * ?") + @Scheduled(cron = "0 20 * * * ?") public void cleanupExpiredEvents() { try { int deleted = outboxService.cleanupExpiredEvents(); diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletMessageOutboxMapper.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletMessageOutboxMapper.java index 369f3732..8f59baed 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletMessageOutboxMapper.java +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/WalletMessageOutboxMapper.java @@ -2,7 +2,6 @@ package com.bsd.pjcloud.wallet.mapper; import com.bsd.pjcloud.common.data.datascope.PjcloudBaseMapper; import com.bsd.pjcloud.wallet.api.entity.WalletMessageOutboxEntity; -import com.baomidou.mybatisplus.annotation.InterceptorIgnore; import org.apache.ibatis.annotations.Delete; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; @@ -16,24 +15,32 @@ import java.time.LocalDateTime; public interface WalletMessageOutboxMapper extends PjcloudBaseMapper { /** - * 跨租户物理删除超过30天的成功Outbox，每次限制数量以控制锁范围。 + * 物理删除当前租户超过30天的成功Outbox，每次限制数量以控制锁范围。 * - * @param before 成功处理时间截止值 - * @param limit 单次最大删除数量 - * @return 实际删除数量 - */ - @InterceptorIgnore(tenantLine = "true") - @Delete("DELETE FROM wallet_message_outbox WHERE status = 'SUCCESS' AND processed_time < #{before} LIMIT #{limit}") - int physicalDeleteSuccessBefore(@Param("before") LocalDateTime before, @Param("limit") int limit); - - /** - * 跨租户物理删除超过一年的失败终态Outbox，每次限制数量以保留排查窗口。 + *

必须在租户上下文内调用（租户拦截器自动附加 tenant_id 条件），配合 + * (tenant_id, status, next_retry_time, create_time) 扫描索引：SUCCESS 行的 + * next_retry_time 恒为空，索引内天然按 create_time 有序，DELETE 沿索引范围 + * 从最老记录开始命中 LIMIT 上限即停，避免全表扫描定位待删行。 + * 过滤字段使用 create_time 而非 processed_time：后者不在任何索引中，且成功行 + * 两者相差仅秒级，按保留期粗粒度清理语义不受影响。

+ * + * @param before 创建时间截止值 + * @param limit 单次最大删除数量 + * @return 实际删除数量 + */ + @Delete("DELETE FROM wallet_message_outbox WHERE status = 'SUCCESS' AND create_time < #{before} LIMIT #{limit}") + int physicalDeleteSuccessBefore(@Param("before") LocalDateTime before, @Param("limit") int limit); + + /** + * 物理删除当前租户超过一年的失败终态Outbox，每次限制数量以保留排查窗口。 + * + *

必须在租户上下文内调用（租户拦截器自动附加 tenant_id 条件）， + * 与成功清理同样沿 (tenant_id, status, create_time) 索引范围删除。

* * @param before 创建时间截止值 * @param limit 单次最大删除数量 * @return 实际删除数量 */ - @InterceptorIgnore(tenantLine = "true") @Delete("DELETE FROM wallet_message_outbox WHERE status = 'FAILED' AND create_time < #{before} LIMIT #{limit}") int physicalDeleteFailedBefore(@Param("before") LocalDateTime before, @Param("limit") int limit); } diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImpl.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImpl.java index 2676a1fa..ba1f86b8 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImpl.java +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImpl.java @@ -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 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 cachedTenantIds = List.of(); + + /** 最近一次刷新租户ID缓存的时间戳，单位 System.currentTimeMillis() 毫秒。 */ + private volatile long tenantCacheRefreshedAt = 0L; + /** - * 扫描到期候选，并按每条记录持久化的租户恢复数据库与Feign上下文。 + * 按租户循环扫描到期候选并逐条在租户上下文内处理。 + * + *

历史实现构造不带租户条件的跨租户查询，而扫描索引 (tenant_id, status, ...) 以 + * tenant_id 为前导列，导致每次扫描退化为全表扫描；表增长到百万行后每10秒一次的 + * 全表扫描持续占用IO，拖慢整个数据库实例。现按租户切分查询：租户上下文内的 + * MyBatis-Plus租户拦截器自动补充 tenant_id 等值条件，扫描命中索引前缀，单租户 + * 仍保持 SCAN_LIMIT 上限，整体语义与单租户轮询一致。

* * @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 candidates = outboxMapper.selectList( Wrappers.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++; } @@ -116,16 +156,64 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic } /** - * 分批物理删除到期终态记录，避免逻辑删除继续占用保留空间。 + * 返回待扫描的租户ID列表，带60秒本地缓存；远程查询失败时退回上一次缓存。 * - * @return 本轮实际删除数量 + *

扫描任务每10秒触发一次，逐次远程查询租户列表会产生日均近万次无效调用； + * 租户新增最迟在缓存过期后一轮扫描内被纳入。远程失败且无缓存时返回空列表， + * 本轮扫描跳过，等待下一轮重试，不中断调度线程。

+ * + * @return 当前需要扫描的租户ID列表，可能为空 + */ + List resolveTenantIds() { + long now = System.currentTimeMillis(); + List cached = cachedTenantIds; + if (!cached.isEmpty() && now - tenantCacheRefreshedAt < TENANT_CACHE_TTL_MS) { + return cached; + } + try { + R> result = tenantService.list(); + List tenants = result == null || !result.isOk() ? null : result.getData(); + if (ObjectUtil.isNotEmpty(tenants)) { + List 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; + } + + /** + * 按租户循环分批物理删除到期终态记录，避免逻辑删除继续占用保留空间。 + * + *

历史实现使用跨租户 DELETE（不带 tenant_id 条件），无法命中 + * (tenant_id, status, ...) 扫描索引，清理退化为全表扫描定位待删行； + * 现与扫描任务一致按租户切分，DELETE 沿索引范围从最老记录开始删除。

+ * + * @return 本轮实际删除记录数量 */ @Override public int cleanupExpiredEvents() { - LocalDateTime now = LocalDateTime.now(); - int successCount = outboxMapper.physicalDeleteSuccessBefore(now.minusDays(SUCCESS_RETENTION_DAYS), CLEANUP_BATCH_SIZE); - int failedCount = outboxMapper.physicalDeleteFailedBefore(now.minusDays(FAILED_RETENTION_DAYS), CLEANUP_BATCH_SIZE); - return successCount + failedCount; + LocalDateTime successBefore = LocalDateTime.now().minusDays(SUCCESS_RETENTION_DAYS); + LocalDateTime failedBefore = LocalDateTime.now().minusDays(FAILED_RETENTION_DAYS); + int total = 0; + for (Long tenantId : resolveTenantIds()) { + try { + Integer deleted = TenantBroker.applyAs(tenantId, tenant -> { + int successCount = outboxMapper.physicalDeleteSuccessBefore(successBefore, CLEANUP_BATCH_SIZE); + int failedCount = outboxMapper.physicalDeleteFailedBefore(failedBefore, CLEANUP_BATCH_SIZE); + return successCount + failedCount; + }); + if (deleted != null) { + total += deleted; + } + } catch (Exception exception) { + log.error("钱包消息Outbox租户清理异常,tenantId:{}", tenantId, exception); + } + } + return total; } /** diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/test/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImplTest.java b/pjcloud-wallet/pjcloud-wallet-biz/src/test/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImplTest.java new file mode 100644 index 00000000..44a33700 --- /dev/null +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/test/java/com/bsd/pjcloud/wallet/service/impl/WalletMessageOutboxServiceImplTest.java @@ -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()); + } +}