TiDB 7.5 VIP 切换中的门诊收费事务：断线重连与重复请求处理

上一篇《TiDB 7.5 HAProxy 双机与 Keepalived VIP 切换测试记录》把数据库入口统一到了 192.168.56.115:3390。代理节点异常后，应用可以继续用这个地址建立新连接。如果收费员确认入账时报了连接错误，这张单到底有没有入账，还能不能再点一次确认？

这次沿用前面的 VIP 入口，模拟一次付清的门诊现金入账。数据库里不调用银行卡、医保或第三方支付接口，也不处理拆分支付和退费。地址和 TiDB v7.5.7 的示例版本沿用上一篇，业务编号、金额和收费项目均为模拟测试内容，金额也不代表任何医院的收费标准。

固定收费编号

使用上一篇的两台 TiDB Server 和两台 HAProxy：

这次单独建一个模拟库，方便反复检查收费单和流水。原来 his_source 里的门诊、收费和医嘱表继续保留。配置和建表使用有相应权限的测试管理账号，后面的客户端使用只获准访问模拟库的业务账号。

连接后先记录版本和会话状态：

SELECT VERSION() AS server_version,
       @@hostname AS tidb_server,
       CONNECTION_ID() AS connection_id,
       @@autocommit AS autocommit,
       @@tidb_txn_mode AS txn_mode;

一笔收费保留三个不同的编号。bill_no 对应待收费单，request_no 对应这次确认入账的请求，receipt_no 是本例中的入账流水号，不代表财政电子票据号码。同一笔请求重送时，这三个编号以及金额都保持不变。客户端发送前就要保存这些信息，不能等连接断了再临时生成一组新编号。

准备待收费单和费用明细

下面三张表只保留这次验证需要的字段。收费单上的 UNPAID、PAID 是示例程序约定的状态，分别表示未入账和已入账。

CREATE DATABASE his_charge_lab
  DEFAULT CHARACTER SET utf8mb4
  COLLATE utf8mb4_bin;

USE his_charge_lab;

CREATE TABLE outpatient_bill (
    bill_no       VARCHAR(32) NOT NULL COMMENT '待收费单号',
    visit_no      VARCHAR(32) NOT NULL COMMENT '模拟门诊号',
    dept_code     VARCHAR(16) NOT NULL COMMENT '开单科室',
    bill_amount   DECIMAL(12,2) NOT NULL COMMENT '应收金额',
    bill_status   VARCHAR(16) NOT NULL COMMENT 'UNPAID或PAID',
    update_time   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
    PRIMARY KEY (bill_no),
    KEY idx_visit_no (visit_no)
);

CREATE TABLE outpatient_bill_item (
    bill_no       VARCHAR(32) NOT NULL,
    item_no       SMALLINT NOT NULL COMMENT '单内明细序号',
    item_code     VARCHAR(24) NOT NULL COMMENT '模拟收费项目编码',
    item_name     VARCHAR(64) NOT NULL,
    qty           DECIMAL(10,2) NOT NULL,
    unit_price    DECIMAL(12,2) NOT NULL,
    amount        DECIMAL(12,2) NOT NULL COMMENT '本条明细金额',
    PRIMARY KEY (bill_no, item_no)
);

CREATE TABLE outpatient_receipt (
    receipt_no    VARCHAR(32) NOT NULL COMMENT '入账流水号',
    request_no    VARCHAR(40) NOT NULL COMMENT '入账请求号',
    bill_no       VARCHAR(32) NOT NULL,
    paid_amount   DECIMAL(12,2) NOT NULL,
    pay_method    VARCHAR(8) NOT NULL COMMENT '本例只使用CASH',
    cashier_code  VARCHAR(16) NOT NULL COMMENT '模拟收费员工号',
    paid_time     DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
    PRIMARY KEY (receipt_no),
    UNIQUE KEY uk_request_no (request_no),
    UNIQUE KEY uk_bill_no (bill_no)
);

uk_request_no 限制同一个请求只保存一份流水。uk_bill_no 限制一张收费单只产生一份入账流水，这种逻辑只适合这种测试环境“一次付清、不含退费”的范围。现实中的组合支付、补缴和冲正有不同的数据关系，不能照搬这个唯一索引。唯一键检查也可能影响提交结果，程序必须同时检查写入和提交阶段的错误，不能把 INSERT 没报错当成事务已经完成。

给每轮操作准备一张独立的收费单，避免上一轮已入账的数据影响下一轮。这些数据不关联真实患者，也不使用正式收费字典。

BEGIN;

INSERT INTO outpatient_bill
    (bill_no, visit_no, dept_code, bill_amount, bill_status)
VALUES
    ('TESTSFD2609071038', 'TESTMZ260907501', 'CARD', 35.00, 'UNPAID'),
    ('TESTSFD2609071062', 'TESTMZ260907524', 'RESP', 28.00, 'UNPAID'),
    ('TESTSFD2609071085', 'TESTMZ260907539', 'SURG', 46.00, 'UNPAID'),
    ('TESTSFD2609071127', 'TESTMZ260907576', 'ENT',  50.00, 'UNPAID');

INSERT INTO outpatient_bill_item
    (bill_no, item_no, item_code, item_name, qty, unit_price, amount)
VALUES
    ('TESTSFD2609071038', 1, 'SIM-REG', '普通门诊诊查费（模拟）', 1, 15.00, 15.00),
    ('TESTSFD2609071038', 2, 'SIM-ECG', '常规心电图（模拟）',     1, 20.00, 20.00),
    ('TESTSFD2609071062', 1, 'SIM-CBC', '血常规检查（模拟）',     1, 20.00, 20.00),
    ('TESTSFD2609071062', 2, 'SIM-BLD', '采血材料费（模拟）',     1,  8.00,  8.00),
    ('TESTSFD2609071085', 1, 'SIM-DRS', '换药处置（模拟）',       1, 30.00, 30.00),
    ('TESTSFD2609071085', 2, 'SIM-MAT', '换药材料费（模拟）',     2,  8.00, 16.00),
    ('TESTSFD2609071127', 1, 'SIM-ENT', '耳鼻喉检查（模拟）',     1, 40.00, 40.00),
    ('TESTSFD2609071127', 2, 'SIM-SUP', '检查材料费（模拟）',     1, 10.00, 10.00);

两条 INSERT 都没有报错时，再执行：

COMMIT;

任何一条报错都先 ROLLBACK，不要接着提交另一部分。TiDB 支持语句级回滚，单条 SQL 失败后，当前事务可能仍然打开，前面成功的修改也可能仍在事务里。

准备数据只执行一次。重复请求测试直接复用现有收费单，不重新插入，也不把已入账的状态改回 UNPAID。录入后先看单头与明细金额能否对上：

SELECT b.bill_no,
       b.visit_no,
       b.bill_amount,
       b.bill_status,
       d.item_count,
       d.item_amount,
       b.bill_amount - COALESCE(d.item_amount, 0) AS amount_diff
FROM outpatient_bill b
LEFT JOIN (
    SELECT bill_no,
           COUNT(*) AS item_count,
           SUM(amount) AS item_amount
    FROM outpatient_bill_item
    GROUP BY bill_no
) d ON d.bill_no = b.bill_no
WHERE b.bill_no IN ('TESTSFD2609071038',
                    'TESTSFD2609071062',
                    'TESTSFD2609071085',
                    'TESTSFD2609071127')
ORDER BY b.bill_no;

按上述示例数据，每张单都应有两条明细，金额差为零。

入账流水与收费单状态一起提交

处理一笔请求时，先锁定已经存在的收费单，读取应收金额和当前状态。未入账时，检查明细合计，再插入流水并把收费单改成 PAID；已经入账时，要核对是不是同一个请求，不能只看到 PAID 就返回成功。

这里显式使用 BEGIN PESSIMISTIC。读取收费单和流水的 SQL 都带 FOR UPDATE：第二个请求可能先等第一个请求释放锁，等到锁后需要读取最新已提交的数据。如果改成普通 SELECT，在默认 RR 隔离级别下可能读到事务开始时的快照。

修改状态还要带 bill_status = 'UNPAID' 和金额条件，并检查影响行数恰好为 1。示例假定待收费单生成后明细固定。如果还允许改价、撤项或追加项目，相关程序也必须先锁同一张单头，再检查状态并修改明细。

下面是完整的模拟客户端，保存为 charge_client.py。一个进程处理一笔请求，方便保留这笔业务的操作记录。密码从终端输入；金额在 Python 中使用 Decimal，SQL 参数交给驱动绑定。

import argparse
import getpass
import json
import os
import time
from contextlib import suppress
from datetime import datetime
from decimal import Decimal

import pymysql


RECEIPT_SQL = """
    SELECT r.receipt_no, r.request_no, r.bill_no,
           r.paid_amount, r.pay_method, r.cashier_code,
           b.bill_amount, b.bill_status
    FROM outpatient_receipt r
    JOIN outpatient_bill b ON b.bill_no = r.bill_no
    WHERE r.bill_no = %s OR r.request_no = %s OR r.receipt_no = %s
"""


def same_payment(row, req):
    return (
        row["receipt_no"] == req.receipt
        and row["request_no"] == req.request
        and row["bill_no"] == req.bill
        and row["paid_amount"] == req.amount
        and row["bill_amount"] == req.amount
        and row["pay_method"] == "CASH"
        and row["cashier_code"] == req.cashier
        and row["bill_status"] == "PAID"
    )


def check_payment(conn, req):
    # 新连接、autocommit=True；这条查询自己取得一个一致的读快照。
    with conn.cursor() as cur:
        cur.execute(RECEIPT_SQL, (req.bill, req.request, req.receipt))
        rows = cur.fetchall()
    if not rows:
        return "PENDING_CHECK"
    if len(rows) == 1 and same_payment(rows[0], req):
        return "CONFIRMED"
    return "MISMATCH"


def post_payment(conn, req, progress):
    with conn.cursor() as cur:
        progress["phase"] = "BEGIN"
        cur.execute("BEGIN PESSIMISTIC")

        progress["phase"] = "CHECK_BILL"
        cur.execute("""
            SELECT bill_amount, bill_status
            FROM outpatient_bill
            WHERE bill_no = %s
            FOR UPDATE
        """, (req.bill,))
        bill = cur.fetchone()
        if bill is None:
            raise ValueError("待收费单不存在")
        if bill["bill_amount"] != req.amount:
            raise ValueError("请求金额与收费单不一致")

        # 等锁以后仍使用当前读，避免把已提交的流水读成不存在。
        cur.execute(RECEIPT_SQL + " FOR UPDATE",
                    (req.bill, req.request, req.receipt))
        rows = cur.fetchall()
        if rows:
            if len(rows) == 1 and same_payment(rows[0], req):
                conn.rollback()  # 本次没有写入，结束事务并释放锁。
                return "ALREADY_POSTED"
            raise ValueError("已有流水与本次请求不一致，停止入账")
        if bill["bill_status"] != "UNPAID":
            raise ValueError("收费单状态不允许入账")

        cur.execute("""
            SELECT item_no, amount
            FROM outpatient_bill_item
            WHERE bill_no = %s
            ORDER BY item_no
            FOR UPDATE
        """, (req.bill,))
        items = cur.fetchall()
        item_amount = sum((row["amount"] for row in items), Decimal("0.00"))
        if not items or item_amount != req.amount:
            raise ValueError("收费明细为空或明细合计不一致")

        progress["phase"] = "WRITE"
        cur.execute("""
            INSERT INTO outpatient_receipt
                (receipt_no, request_no, bill_no,
                 paid_amount, pay_method, cashier_code)
            VALUES (%s, %s, %s, %s, 'CASH', %s)
        """, (req.receipt, req.request, req.bill, req.amount, req.cashier))

        cur.execute("""
            UPDATE outpatient_bill
            SET bill_status = 'PAID', update_time = NOW(3)
            WHERE bill_no = %s
              AND bill_status = 'UNPAID'
              AND bill_amount = %s
        """, (req.bill, req.amount))
        if cur.rowcount != 1:
            raise ValueError("收费单更新行数不是1，取消本次入账")

        if req.pause_before_commit:
            progress["phase"] = "BEFORE_COMMIT"
            input("已执行写入，尚未提交；完成故障操作后按回车继续：")

        progress["phase"] = "COMMIT"
        conn.commit()
        if req.omit_success:
            # 仅模拟：数据库提交已返回成功，但不给上层确认入账成功。
            return "SIMULATED_REPLY_LOSS"
        return "POSTED"


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=("post", "check"))
    parser.add_argument("--bill", required=True)
    parser.add_argument("--request", required=True)
    parser.add_argument("--receipt", required=True)
    parser.add_argument("--amount", required=True, type=Decimal)
    parser.add_argument("--cashier", required=True)
    parser.add_argument("--pause-before-commit", action="store_true")
    parser.add_argument("--omit-success", action="store_true")
    req = parser.parse_args()
    if (not req.amount.is_finite() or req.amount <= 0
            or req.amount >= Decimal("10000000000")
            or req.amount != req.amount.quantize(Decimal("0.01"))):
        parser.error("金额必须为大于0、符合DECIMAL(12,2)范围的两位小数")

    password = getpass.getpass("数据库密码：")
    progress = {"phase": "CONNECT"}
    result = {"request_no": req.request, "bill_no": req.bill}
    conn = None
    started = time.monotonic()
    try:
        conn = pymysql.connect(
            host=os.environ.get("TIDB_HOST", "192.168.56.115"),
            port=int(os.environ.get("TIDB_PORT", "3390")),
            user=os.environ.get("TIDB_USER", "his_app"),
            password=password,
            database="his_charge_lab",
            charset="utf8mb4",
            autocommit=True,
            connect_timeout=5,
            read_timeout=15,
            write_timeout=15,
            cursorclass=pymysql.cursors.DictCursor,
        )
        progress["phase"] = "SESSION"
        with conn.cursor() as cur:
            cur.execute("SET SESSION innodb_lock_wait_timeout = 5")
            cur.execute("SELECT @@hostname AS tidb_server, CONNECTION_ID() AS connection_id")
            result.update(cur.fetchone())

        if req.action == "check":
            progress["phase"] = "QUERY_RESULT"
            result["status"] = check_payment(conn, req)
        else:
            result["status"] = post_payment(conn, req, progress)
    except (pymysql.MySQLError, ValueError, EOFError, KeyboardInterrupt) as exc:
        if conn is not None:
            with suppress(Exception):
                conn.rollback()
        result["status"] = (
            "COMMIT_UNCONFIRMED" if progress["phase"] == "COMMIT" else "STOPPED"
        )
        result["error_code"] = (
            exc.args[0] if isinstance(exc, pymysql.MySQLError) and exc.args else None
        )
        result["error"] = str(exc) or type(exc).__name__
    finally:
        if conn is not None:
            with suppress(Exception):
                conn.close()

    result["phase"] = progress["phase"]
    result["elapsed_ms"] = round((time.monotonic() - started) * 1000, 1)
    result["time"] = datetime.now().astimezone().isoformat(timespec="milliseconds")
    print(json.dumps(result, ensure_ascii=False), flush=True)
    return 0 if result["status"] in ("POSTED", "ALREADY_POSTED", "CONFIRMED") else 2


if __name__ == "__main__":
    raise SystemExit(main())

本例设置了连接和读写超时，避免客户端一直等待；这些数值只用于模拟。read_timeout 到期只表示客户端没有及时读到响应，不能用它判断事务已回滚。

程序没有在异常后循环重发。COMMIT 阶段只要抛出异常，就先标记为 COMMIT_UNCONFIRMED。有些错误码能直接确认提交失败，排查时还要看 error_code。异常处理里的 rollback() 用于尽可能清理当前连接，它不能证明此前发出的提交已经被撤销。

先跑正常请求，再重复提交原请求

在测试客户端安装驱动，并记录实际使用的版本：

python3 -m venv .venv-charge
source .venv-charge/bin/activate
python3 -m pip install PyMySQL
python3 -m pip show PyMySQL

35 元的收费单用于检查正常入账。请求号和流水号在首次发送前确定，后面的查询、重送都使用这一组：

python3 charge_client.py post \
  --bill TESTSFD2609071038 \
  --request TESTREQ260907SF0182 \
  --receipt TESTRCPT2609070816 \
  --amount 35.00 \
  --cashier SIM018

代码只有在 commit() 返回成功后才报告 POSTED。原样再运行一次时，程序应读取已有流水，核对请求内容后返回 ALREADY_POSTED，不再执行插入和状态更新。

把第二次请求的金额改成 35.01，检查程序是否拒绝；或者保留收费单号，改一个请求号，检查它是否会把另一笔请求错误地当成已经成功。测试这些反例时不能删掉原流水，否则检验不到重复处理的分支。

并发检查用 46 元的收费单，在两个客户端同时发送相同请求：

python3 charge_client.py post \
  --bill TESTSFD2609071085 \
  --request TESTREQ260907SF0204 \
  --receipt TESTRCPT2609070843 \
  --amount 46.00 \
  --cashier SIM021

一边可能先提交，另一边随后读到已有流水；也可能因为等锁或其他错误需要重新核对。判断条件是这张单最多只有一份入账流水，并且请求号、金额和收费员与原请求一致。不能要求两边都返回“首次入账成功”。

在提交前停止当前代理

断线验证用 28 元的收费单。开始前，两台 HAProxy 和 Keepalived 都要恢复正常，两台代理的实际地址都能连接 TiDB，VIP 只出现在一个节点。每轮故障恢复完成后，再开始下一轮。

在客户端执行：

python3 charge_client.py post \
  --bill TESTSFD2609071062 \
  --request TESTREQ260907SF0191 \
  --receipt TESTRCPT2609070829 \
  --amount 28.00 \
  --cashier SIM018 \
  --pause-before-commit

程序会停在提交之前。这个暂停只用于留出故障操作时间，模拟数据在事务里还没有提交。此时在两台 HAProxy 服务器上分别检查 VIP：

ip -4 addr show dev ens192 | grep 192.168.56.115

在当前持有 VIP 的节点上记录时间并停止 HAProxy：

date '+%F %T.%N'
systemctl stop haproxy
systemctl status haproxy
ss -lntp | grep 3390

这里先确认本机代理已经停止。HAProxy 支持硬停止和优雅停止，两者对已有连接的处理不同，实际 systemctl stop 行为还要结合本机服务单元检查；不能用一次 reload 代替连接中断测试。

按上一篇的方法检查 VIP 是否转到另一台代理，再回到客户端按回车，让它尝试提交。因为故障操作发生在发出 COMMIT 之前，这一轮针对的是未提交事务断线。TiDB 在连接终止后会回滚未提交事务，但客户端刚报错时，不能假定服务端已经立刻完成所有清理。

重新打开一个客户端，使用相同参数查询：

python3 charge_client.py check \
  --bill TESTSFD2609071062 \
  --request TESTREQ260907SF0191 \
  --receipt TESTRCPT2609070829 \
  --amount 28.00 \
  --cashier SIM018

如果没有查到匹配流水，程序返回 PENDING_CHECK。这个状态只表示尚未确认入账，不能当成“收费失败”。先看客户端停在哪一步，再检查收费单状态和数据库是否已经恢复。确认可以重新处理后，去掉 --pause-before-commit，仍用原来的请求号、流水号和金额执行 post。新事务会重新锁定收费单并检查现有记录，不会跳过判断直接补一条流水。

本轮结束后，在刚才停止 HAProxy 的节点恢复服务。默认抢占可能再次引起 VIP 回切，需要把这次变化也记录下来。

systemctl start haproxy
systemctl status haproxy
ss -lntp | grep 3390

提交已经发出，怎样核对结果

上一轮是在 COMMIT 发出前停代理，没有验证提交结果返回途中断线的情况。COMMIT 已经发出时，数据库可能已提交，但客户端没收到响应。这种情况要保留原请求，另开连接核对收费单和流水。查询也报错，仍然是待确认；查询暂时没有记录，也不能据此认定原事务已经回滚。

为了单独检查“数据库已提交、上层尚未确认”的处理分支，程序提供了 --omit-success。它在 commit() 已返回成功后，故意不给上层正常的成功状态，返回 SIMULATED_REPLY_LOSS。这是客户端故障注入，不能当成真实网络丢包或真实提交响应丢失的复现。使用 50 元的收费单，与前面的并发检查分开。

python3 charge_client.py post \
  --bill TESTSFD2609071127 \
  --request TESTREQ260907SF0228 \
  --receipt TESTRCPT2609070871 \
  --amount 50.00 \
  --cashier SIM026 \
  --omit-success

随后使用相同参数执行 check，核对已提交的结果；再去掉 --omit-success 重送原 post，检查是否走已有流水分支。可以检查重复请求处理是否正确，但不能拿来计算 VIP 切换耗时。

遇到异常，按下面的方式处理：

如果在应用里增加自动重试，要限制次数并留出等待间隔；本次测试客户端保留人工决定重送的方式。

重新建立连接时，要重新设置会话参数。原连接上的事务、锁和会话设置不会随着 VIP 一起搬过来。本例每次创建连接都设置 innodb_lock_wait_timeout，每次入账都重新执行 BEGIN PESSIMISTIC。

故障结束后按单核对

在停止发送测试请求、处理完待确认请求后再汇总。下面按指定收费单查询，费用明细先按单聚合，再关联入账流水，避免一张单多条明细把已收金额重复累计。

USE his_charge_lab;

SELECT b.bill_no,
       b.visit_no,
       b.bill_status,
       b.bill_amount,
       d.item_count,
       d.item_amount,
       r.receipt_no,
       r.request_no,
       r.paid_amount,
       r.pay_method,
       r.cashier_code,
       r.paid_time
FROM outpatient_bill b
LEFT JOIN (
    SELECT bill_no,
           COUNT(*) AS item_count,
           SUM(amount) AS item_amount
    FROM outpatient_bill_item
    GROUP BY bill_no
) d ON d.bill_no = b.bill_no
LEFT JOIN outpatient_receipt r ON r.bill_no = b.bill_no
WHERE b.bill_no IN ('TESTSFD2609071038',
                    'TESTSFD2609071062',
                    'TESTSFD2609071085',
                    'TESTSFD2609071127')
ORDER BY b.bill_no;

入账流水与单头是一笔事务里的修改。程序正常处理后，PAID 应有对应流水；仍是 UNPAID 时不应留下已入账流水。可以用下面的 SQL 找状态或金额对不上的记录：

SELECT b.bill_no,
       b.bill_status,
       b.bill_amount,
       r.receipt_no,
       r.request_no,
       r.paid_amount
FROM outpatient_bill b
LEFT JOIN outpatient_receipt r ON r.bill_no = b.bill_no
WHERE b.bill_no IN ('TESTSFD2609071038',
                    'TESTSFD2609071062',
                    'TESTSFD2609071085',
                    'TESTSFD2609071127')
  AND (
      (b.bill_status = 'PAID' AND r.receipt_no IS NULL)
      OR (b.bill_status = 'UNPAID' AND r.receipt_no IS NOT NULL)
      OR (r.receipt_no IS NOT NULL AND r.paid_amount <> b.bill_amount)
  );

这条查询只筛选它列出的异常，不返回行也不能代替逐笔核对。请求号、流水号、金额和收费员还要与最初发送的请求对应。程序返回 ALREADY_POSTED 时，表示识别到了已有入账，不能再次累计为一笔新收费。

客户端的 elapsed_ms 包含连接、会话初始化、SQL 执行和等锁；使用人工暂停时，还包含人工等待时间。主要用于核对一笔请求经历了多久，不能当成单条 SQL 耗时或数据库切换时间。真实收费还涉及支付渠道对账、票据和退费流程，测试数据库里的 PAID 只能说明本例的入账事务，不能直接判断认为银行扣款或医保结算已经完成。