银行核心系统测试：交易幂等性与数据一致性问题测试分享

import redisimport threadingimport timer = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)def build_idempotency_key(user_id, order_id):    # 幂等键：业务维度唯一，由 用户ID + 订单ID 组成    return "idem:" + user_id + ":" + order_iddef pay_once(key, amount):    # SET NX EX —— 不存在才写入，5分钟过期，原子操作    ok = r.set(key, "PROCESSING", nx=True, ex=300)    if not ok:        return "DUPLICATE"   # 已存在 = 重复请求，直接返回，不二次扣款    try:        time.sleep(0.05)     # 模拟核心扣款逻辑        r.set(key, "SUCCESS", xx=True)        return "SUCCESS"    except Exception:        r.delete(key)        # 失败清理，允许客户端重试        return "FAILED"def simulate_timeout_and_retry(key, amount):    # 首次请求超时（未返回结果），客户端用相同幂等键重试    r.set(key, "PROCESSING", nx=True, ex=300)    time.sleep(2)            # 模拟网络超时窗口    return pay_once(key, amount)   # 第二次应识别为重复def concurrent_test(key, amount, threads=10):    results = []    lock = threading.Lock()    def worker():        res = pay_once(key, amount)        with lock:            results.append(res)    ts = [threading.Thread(target=worker) for _ in range(threads)]    for t in ts: t.start()    for t in ts: t.join()    return results   # 应只有 1 个 SUCCESS，其余 DUPLICATEif __name__ == "__main__":    k = build_idempotency_key("u1001", "o20260816")    # 用例1：并发重复提交，断言只成功 1 次    print(concurrent_test(k, 100.0))    # 用例2：超时重试，断言不二次扣款    print(simulate_timeout_and_retry("k2", 50.0))

-- 账务表：每笔会计分录CREATE TABLE t_account (  id          BIGINT PRIMARY KEY AUTO_INCREMENT,  biz_no      VARCHAR(64)  NOT NULL COMMENT '业务流水号(幂等键)',  acct_no     VARCHAR(32)  NOT NULL COMMENT '账户号',  dr_amount   DECIMAL(18,2) DEFAULT 0 COMMENT '借方金额',  cr_amount   DECIMAL(18,2) DEFAULT 0 COMMENT '贷方金额',  create_time DATETIME NOT NULL) COMMENT='账务表';-- 流水表：逐笔交易痕迹CREATE TABLE t_journal (  id          BIGINT PRIMARY KEY AUTO_INCREMENT,  biz_no      VARCHAR(64)  NOT NULL,  acct_no     VARCHAR(32)  NOT NULL,  amount      DECIMAL(18,2) NOT NULL COMMENT '交易金额',  direction   CHAR(1)    NOT NULL COMMENT 'D借方 C贷方',  create_time DATETIME NOT NULL) COMMENT='流水表';-- 余额表：账户当前余额CREATE TABLE t_balance (  acct_no     VARCHAR(32) PRIMARY KEY,  balance     DECIMAL(18,2) NOT NULL DEFAULT 0,  update_time DATETIME NOT NULL) COMMENT='余额表';-- 对账表：记录未平差异CREATE TABLE t_recon (  id          BIGINT PRIMARY KEY AUTO_INCREMENT,  biz_no      VARCHAR(64),  diff_amount DECIMAL(18,2),  status      CHAR(1) COMMENT '0未平 1已平',  create_time DATETIME) COMMENT='对账表';

-- 校验1：账务表借贷是否平衡（每个 biz_no 借贷应相等）SELECT biz_no,       SUM(dr_amount) AS total_dr,       SUM(cr_amount) AS total_crFROM t_accountGROUP BY biz_noHAVING ABS(SUM(dr_amount) - SUM(cr_amount)) > 0.005;-- 返回空结果 = 借贷平衡；有行 = 存在不平衡-- 校验2：余额表的变动是否等于流水的借贷净额SELECT j.acct_no,       b.balance                                          AS cur_balance,       (SELECT SUM(CASE WHEN direction='C' THEN amount ELSE -amount END)          FROM t_journal j2 WHERE j2.acct_no = j.acct_no) AS net_flowFROM t_journal jJOIN t_balance b ON b.acct_no = j.acct_noGROUP BY j.acct_no, b.balanceHAVING ABS(b.balance - (    SELECT SUM(CASE WHEN direction='C' THEN amount ELSE -amount END)    FROM t_journal j3 WHERE j3.acct_no = j.acct_no)) > 0.005;-- 返回空 = 余额与流水一致-- 校验3：对账表中仍存在未平差异的记录SELECT * FROM t_recon WHERE status = '0';-- 返回空 = 无未平差异；有行 = 需人工干预-- 校验4：幂等性对账——同一 biz_no 不应出现多笔"已生效"流水SELECT biz_no, COUNT(*) cntFROM t_journalGROUP BY biz_noHAVING COUNT(*) > 1;-- 返回空 = 无重复生效；有行 = 幂等可能失效，排查

import threading# 用字典模拟账户与冻结状态（生产用数据库+分布式锁）class TccTransfer:    def __init__(self):        self.balance = {"A": 1000.0, "B": 0.0}        self.frozen  = {}          # biz_no -> 冻结金额        self.confirmed = set()     # 已确认的幂等键        self.lock = threading.Lock()    def try_lock(self, biz_no, payer, amount):        with self.lock:            if biz_no in self.frozen:                return "DUPLICATE_TRY"   # 幂等：重复 Try 直接返回            if self.balance[payer] < amount:                return "INSUFFICIENT"            self.balance[payer] -= amount            self.frozen[biz_no] = amount            return "TRY_OK"    def confirm(self, biz_no, payer, payee, amount):        with self.lock:            if biz_no not in self.frozen:                return "REJECT_NO_TRY"  # Try 未成功，拒绝 Confirm            if biz_no in self.confirmed:                return "DUPLICATE_CONFIRM"  # 幂等            self.balance[payee] += amount            self.frozen.pop(biz_no, 0)            self.confirmed.add(biz_no)            return "CONFIRM_OK"    def cancel(self, biz_no, payer, amount):        with self.lock:            if biz_no not in self.frozen:                return "NO_NEED"      # 没冻结就无需释放            self.balance[payer] += self.frozen.pop(biz_no, 0)  # 幂等：多次调用结果一致            return "CANCEL_OK"# ===== 测试开始 =====if __name__ == "__main__":    t = TccTransfer()    bn = "biz_20260816_001"    # 用例1：Try 成功后再 Confirm    print(t.try_lock(bn, "A", 200.0))   # TRY_OK    print(t.confirm(bn, "A", "B", 200.0)) # CONFIRM_OK    print(t.confirm(bn, "A", "B", 200.0)) # DUPLICATE_CONFIRM（幂等）    # 用例2：未 Try 直接 Confirm，应被拒绝    t2 = TccTransfer()    print(t2.confirm("no_try", "A", "B", 50.0))  # REJECT_NO_TRY    # 用例3：Cancel 幂等，调两次结果一致    t3 = TccTransfer()    t3.try_lock("biz_x", "A", 100.0)    print(t3.cancel("biz_x", "A", 100.0))  # CANCEL_OK    print(t3.cancel("biz_x", "A", 100.0))  # NO_NEED（已释放）    print("A余额=", t3.balance["A"])        # 应为 1000.0，钱回来了

-- 正确的余额扣减：条件更新，数据库保证原子性UPDATE t_balance   SET balance = balance - 80.00,       update_time = NOW() WHERE acct_no = 'A'   AND balance >= 80.00;-- 返回影响行数=1 才表示扣款成功；=0 表示余额不足，绝不超扣