Merge remote-tracking branch 'origin/master'
This commit is contained in:
commit
e3e2726f91
13 files changed
+472
-35
No files matched your search
@@ -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,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;
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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
|
||||
|
||||
+18
@@ -68,6 +68,24 @@ public class HardXfjLogEntity extends Model<HardXfjLogEntity> {
|
||||
@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;
|
||||
|
||||
/**
|
||||
* 开始时间
|
||||
*/
|
||||
|
||||
+45
-2
@@ -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<Long> 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("非本单位饭卡");
|
||||
}
|
||||
|
||||
+3
@@ -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,
|
||||
|
||||
+1
@@ -31,6 +31,7 @@ public class WalletTransactionEntity extends Model<WalletTransactionEntity> {
|
||||
@Schema(description = "编号")
|
||||
private Long id;
|
||||
|
||||
|
||||
/**
|
||||
* 用户 id
|
||||
*/
|
||||
|
||||
+6
-2
@@ -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();
|
||||
|
||||
+20
-13
@@ -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<WalletMessageOutboxEntity> {
|
||||
|
||||
/**
|
||||
* 跨租户物理删除超过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,每次限制数量以保留排查窗口。
|
||||
* <p>必须在租户上下文内调用(租户拦截器自动附加 tenant_id 条件),配合
|
||||
* (tenant_id, status, next_retry_time, create_time) 扫描索引:SUCCESS 行的
|
||||
* next_retry_time 恒为空,索引内天然按 create_time 有序,DELETE 沿索引范围
|
||||
* 从最老记录开始命中 LIMIT 上限即停,避免全表扫描定位待删行。
|
||||
* 过滤字段使用 create_time 而非 processed_time:后者不在任何索引中,且成功行
|
||||
* 两者相差仅秒级,按保留期粗粒度清理语义不受影响。</p>
|
||||
*
|
||||
* @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,每次限制数量以保留排查窗口。
|
||||
*
|
||||
* <p>必须在租户上下文内调用(租户拦截器自动附加 tenant_id 条件),
|
||||
* 与成功清理同样沿 (tenant_id, status, create_time) 索引范围删除。</p>
|
||||
*
|
||||
* @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);
|
||||
}
|
||||
+104
-16
@@ -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++;
|
||||
}
|
||||
@@ -116,16 +156,64 @@ public class WalletMessageOutboxServiceImpl implements WalletMessageOutboxServic
|
||||
}
|
||||
|
||||
/**
|
||||
* 分批物理删除到期终态记录,避免逻辑删除继续占用保留空间。
|
||||
* 返回待扫描的租户ID列表,带60秒本地缓存;远程查询失败时退回上一次缓存。
|
||||
*
|
||||
* @return 本轮实际删除数量
|
||||
* <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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 按租户循环分批物理删除到期终态记录,避免逻辑删除继续占用保留空间。
|
||||
*
|
||||
* <p>历史实现使用跨租户 DELETE(不带 tenant_id 条件),无法命中
|
||||
* (tenant_id, status, ...) 扫描索引,清理退化为全表扫描定位待删行;
|
||||
* 现与扫描任务一致按租户切分,DELETE 沿索引范围从最老记录开始删除。</p>
|
||||
*
|
||||
* @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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+77
@@ -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());
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user