SQL数据库开发实战：用事务保证数据操作的一致性

订单创建时要同时扣减库存、写入订单记录；账户转账时要从一方扣款、向另一方入账。只要这些操作中途失败，数据就可能出现“扣了库存却没有订单”或“钱转出去了却没有到账”的问题。

事务用于把一组数据库操作作为一个整体执行：全部成功则提交，遇到错误则回滚。但事务不是“自动修复一切”的开关。它只能管理事务范围内、受当前数据库支持的操作；并发控制、外部服务调用、超时和重试，都需要额外设计。本文以 PostgreSQL 和 Python 的 psycopg 3 为例，介绍事务的基本用法与关键边界。

一、事务解决什么问题

数据库事务通常用 ACID 描述：

• 原子性（Atomicity）：事务中的操作要么全部生效，要么全部撤销。
• 一致性（Consistency）：事务执行前后都应满足数据库约束和应用定义的业务规则。数据库不会自动知道所有业务规则，需要通过约束、SQL 和应用逻辑共同保证。
• 隔离性（Isolation）：并发事务之间的相互影响由数据库的隔离级别和锁机制控制。
• 持久性（Durability）：提交成功后，数据按数据库的持久化配置保存。具体故障恢复能力还取决于数据库配置、存储和部署方式。

事务适合用于一组必须共同成功或失败的数据库写操作，例如：

• 创建订单并扣减库存；
• 转账时同时更新付款方和收款方余额；
• 写入业务数据及与之配套的审计记录；
• 处理一条消息时更新业务状态并记录消费结果。

如果操作彼此独立，放在一个很大的事务里通常没有好处，反而会延长锁的持有时间、增加冲突和回滚成本。事务边界应尽量贴合业务中“必须一起成立”的那部分数据变化。

二、准备 PostgreSQL 示例环境

以下示例使用 Python 3、PostgreSQL 和 psycopg 3。先安装驱动：

python -m pip install "psycopg[binary]"

准备一个 PostgreSQL 数据库，并将连接信息放入环境变量。以下命令适用于常见的 Linux 或 macOS shell；请按实际环境替换数据库名、用户名和密码。

export DATABASE_URL="postgresql://app_user:your_password@localhost:5432/app_db"

事务示例使用一张账户表。创建表和测试数据的 SQL 如下：

CREATE TABLE IF NOT EXISTS accounts (
    id BIGINT PRIMARY KEY,
    balance NUMERIC(12, 2) NOT NULL CHECK (balance >= 0)
);

INSERT INTO accounts (id, balance)
VALUES (1, 1000.00), (2, 500.00)
ON CONFLICT (id) DO NOTHING;

账户余额使用 NUMERIC，而不是浮点数类型，是为了避免用二进制浮点数表示金额时产生的精度问题。实际系统还应根据货币规则处理小数位、币种和舍入方式。

三、用事务完成一次转账

一次转账至少要完成两项数据库写入：付款方扣款，收款方入账。若扣款成功后入账失败，两条记录必须一起撤销。下面的程序可保存为 transfer.py，设置好 DATABASE_URL 后运行。

import os
from decimal import Decimal

import psycopg


def transfer(conn, from_id, to_id, amount):
    amount = Decimal(str(amount))

    if amount <= 0:
        raise ValueError("转账金额必须大于零")
    if from_id == to_id:
        raise ValueError("付款账户和收款账户不能相同")

    with conn.transaction():
        # 按账户 ID 固定顺序加锁，降低两个相反方向转账时的死锁概率。
        account_ids = sorted((from_id, to_id))
        conn.execute(
            """
            SELECT id
            FROM accounts
            WHERE id = ANY(%s)
            ORDER BY id
            FOR UPDATE
            """,
            (account_ids,),
        )
        locked_ids = {row[0] for row in conn.fetchall()}

        if locked_ids != set(account_ids):
            raise ValueError("账户不存在")

        debit = conn.execute(
            """
            UPDATE accounts
            SET balance = balance - %s
            WHERE id = %s AND balance >= %s
            RETURNING balance
            """,
            (amount, from_id, amount),
        ).fetchone()

        if debit is None:
            raise ValueError("账户不存在或余额不足")

        credit = conn.execute(
            """
            UPDATE accounts
            SET balance = balance + %s
            WHERE id = %s
            RETURNING balance
            """,
            (amount, to_id),
        ).fetchone()

        if credit is None:
            raise ValueError("收款账户不存在")

    # 离开 transaction 块且没有异常时，psycopg 提交事务。
    return debit[0], credit[0]


def main():
    database_url = os.environ["DATABASE_URL"]

    with psycopg.connect(database_url) as conn:
        balances = transfer(conn, 1, 2, Decimal("125.50"))
        print("转账完成，转出后余额、转入后余额：", balances)


if __name__ == "__main__":
    main()

在 psycopg 3 中，连接的 transaction() 上下文用于管理事务：代码块正常结束时提交；代码块内抛出异常时回滚，然后异常继续向外传播。外层的 with psycopg.connect(...) 在退出时也会关闭连接，并根据连接上下文规则处理未完成的事务。

这个实现包含几个重要做法：

• 先检查金额。金额必须为正数；同一账户转给自己也应按业务规则处理，示例选择拒绝。
• 在数据库内检查并扣款。WHERE balance >= %s 把余额条件和更新放在同一条 SQL 中，避免先在应用里读取余额、再基于过期结果扣款。
• 用约束作为最后防线。表上的 CHECK (balance >= 0) 能拒绝负余额写入，但不能替代完整的业务验证。
• 锁定将要修改的行。SELECT ... FOR UPDATE 会在事务结束前锁住匹配到的行，避免其他事务同时修改这些账户。锁会占用到提交或回滚，所以事务中不要做无关工作。
• 对锁定顺序保持一致。涉及多行时按固定顺序加锁能降低死锁风险，但不能保证死锁绝不会发生。

程序还要正确处理异常。如果转账代码内部捕获数据库异常后继续执行，却没有回滚或重新开始事务，就可能造成事务状态不符合预期。通常应让异常离开事务上下文，使上下文管理器完成回滚；若要捕获异常，则在清楚事务状态的前提下处理，并避免把失败伪装成成功。

四、事务保证不了什么：并发与隔离

事务原子性保证的是“一组操作一起提交或撤销”，不等于并发事务一定按应用想象的顺序执行。多个事务同时读写数据时，可能出现读取不一致、更新覆盖、余额检查竞争等问题。解决方案取决于业务规则和数据库隔离级别。

PostgreSQL 默认隔离级别是 READ COMMITTED。在这个级别下，每条语句通常看到该语句开始时已经提交的数据；同一个事务中的两条查询，可能看到不同的已提交结果。因此，不能仅凭“所有 SQL 都包在事务里”就推断重复查询结果始终不变。

对简单的余额扣减，可以像示例那样使用带条件的原子更新，并检查是否更新成功。对需要先读取多行、再据此作出决定的业务，可以使用行锁、唯一约束、条件更新，或按需求采用更强隔离级别。选择之前应明确业务不变量，例如“库存不能为负”或“每个用户同一时刻只能有一笔有效预约”。

SELECT ... FOR UPDATE 只锁定实际找到的行，不会自动锁住所有未来可能插入、但尚不存在的行。在 READ COMMITTED 下，单纯查询“某条件下不存在记录”再插入，不能可靠防止两个并发事务同时插入符合该条件的记录。此时应优先考虑唯一约束，并在冲突时处理数据库返回的唯一性错误；其他方案可能涉及更强隔离级别或专门的锁设计。

用约束处理“只能有一条”的规则

例如一个用户只能有一条处于生效状态的订阅，可以用部分唯一索引表达这个规则。PostgreSQL 示例：

CREATE TABLE subscriptions (
    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    user_id BIGINT NOT NULL,
    status TEXT NOT NULL
);

CREATE UNIQUE INDEX one_active_subscription_per_user
ON subscriptions (user_id)
WHERE status = 'active';

这样即使两个事务同时尝试插入同一用户的有效订阅，数据库也会拒绝其中一个。应用仍需捕获并解释唯一性冲突。对数据完整性要求高的规则，数据库约束通常比“先查一下有没有，再插入”更可靠。

隔离级别与序列化重试

当业务需要更强的隔离时，可以在事务开始时设置隔离级别。以 psycopg 3 为例，可通过连接事务选项配置；具体写法应以所用驱动版本的文档为准。PostgreSQL 的 SERIALIZABLE 会检测某些无法等价于串行执行的并发结果，并可能中止其中一个事务。被中止的事务需要由应用重新执行。

以下示例展示事务失败后进行有限次数的重试。它针对 PostgreSQL 的序列化失败和死锁错误使用 SQLSTATE 判断；重试次数应有上限，并可加入随机退避，避免多个请求持续同步重试。

import random
import time

import psycopg
from psycopg import errors


RETRYABLE = (
    errors.SerializationFailure,
    errors.DeadlockDetected,
)


def run_with_retry(operation, attempts=4):
    for attempt in range(attempts):
        try:
            return operation()
        except RETRYABLE:
            if attempt == attempts - 1:
                raise
            time.sleep((0.02 * (2 ** attempt)) + random.random() * 0.02)

这里的 operation 必须在每次调用时重新开始一个完整事务，不能在失败后继续使用已经中止的事务状态。只有确认事务失败且没有提交时，才适合按这种方式重试。不要对所有数据库异常一概重试：连接断开、约束错误、权限错误等问题未必能靠重试解决。

还有一种更棘手的情况：客户端发送提交请求后，连接突然断开。此时客户端可能不知道数据库是否已经提交。盲目重做可能造成重复扣款或重复下单。对此要使用业务幂等键、唯一约束或操作记录，使相同请求重复到达时不会产生重复效果。

五、参数化查询与 SQL 安全边界

示例中的 %s 是 psycopg 的参数占位符。参数值由驱动单独传递和编码，不应使用字符串拼接把用户输入拼进 SQL。下面这种写法有 SQL 注入风险：

# 错误：不要把不可信输入拼进 SQL
sql = "SELECT id FROM accounts WHERE id = " + user_input
conn.execute(sql)

应使用参数化查询：

# 正确：值作为参数传递
conn.execute(
    "SELECT id FROM accounts WHERE id = %s",
    (account_id,),
)

参数化的安全用途是把数据值与 SQL 语法分开，降低 SQL 注入风险。它不是对业务输入的完整验证，也不能自动防止越权、错误的账户归属判断、资源滥用或不正确的业务流程。

参数占位符通常用于数据值，不能直接替代表名、列名、排序方向等 SQL 标识符。例如，不能把用户输入作为 ORDER BY %s 的参数，期待它变成列名。对动态标识符，应使用驱动提供的标识符组合功能，或从严格的允许列表中选择固定 SQL 片段；不要把任意用户字符串直接拼接到查询语句中。

六、嵌套事务与保存点

数据库事务通常不能简单地在一个事务里再次开启独立事务。需要撤销事务的一部分、同时保留前面操作时，可以使用保存点。PostgreSQL 支持保存点；psycopg 的嵌套事务上下文可用于创建相应的保存点。

with conn.transaction():
    conn.execute(
        "INSERT INTO audit_log (event) VALUES (%s)",
        ("request_started",),
    )

    try:
        with conn.transaction():
            conn.execute(
                "INSERT INTO optional_items (name) VALUES (%s)",
                ("extra-item",),
            )
            raise ValueError("模拟可选步骤失败")
    except ValueError:
        # 内层事务回滚到保存点，外层事务仍可继续。
        pass

    conn.execute(
        "INSERT INTO audit_log (event) VALUES (%s)",
        ("request_finished",),
    )

保存点适合处理可选步骤或局部可恢复错误，不应成为吞掉关键失败的手段。示例中的异常是应用异常，用来展示控制流程；数据库错误应特别注意事务状态，按驱动和数据库的行为正确回滚到保存点或回滚整个事务。

七、外部服务调用不属于数据库事务

数据库事务只能原子地管理相应数据库中的事务操作，不能自动撤回已经发送的 HTTP 请求、邮件、短信或消息队列发布。例如，在事务中扣款后调用支付服务，如果服务成功而数据库随后回滚，外部系统和数据库就会不一致；反过来，若先调用外部服务再写数据库，写库失败也可能留下外部副作用。

常见做法是避免在持有数据库锁期间等待外部网络请求，并根据业务选择可靠的协调方式：

• 事务性发件箱（Outbox）。在同一个数据库事务中写入业务数据和一条待发送事件；事务提交后由后台任务读取事件并投递。投递可能重复，因此消费者需要幂等处理。
• 幂等键。为请求分配稳定的业务标识，服务端记录已处理结果。客户端重试时返回原结果或避免重复执行。幂等键需要有明确的作用范围、保留期限和冲突处理规则。
• 补偿流程。对于无法放进同一事务的多步骤业务，设计可执行的反向操作或人工处理流程。补偿不等于真正的回滚，因为外部操作可能已经被观察或使用。

不要为了追求“整个系统只有一个事务”而在数据库事务中长时间等待第三方服务。长事务会延长锁的持有时间，导致其他请求排队，甚至引发连接耗尽。

八、常见错误及排查方向

• 每条 SQL 都单独提交。若业务需要多条写操作共同成立，逐条提交就失去了整体原子性。应把相关写操作放进同一个事务。
• 先查询、后无条件更新。并发请求可能在查询后改变数据。将校验条件放进更新语句，或使用适当的行锁，并检查受影响行数。
• 事务范围过大。在事务内执行网络请求、等待用户输入或处理大量无关数据，会增加锁竞争和失败成本。只把必要的数据库操作放进去。
• 忽略提交失败和异常。代码执行到事务块末尾不代表任何环境下都能成功提交。捕获异常时要区分业务拒绝、约束冲突、死锁和连接问题。
• 以为加锁就不会出错。锁只能控制特定数据上的并发访问。死锁依然可能发生，锁也不能替代约束和业务校验。
• 重试所有失败请求。约束错误通常不是暂时故障；提交结果未知时重复执行可能产生副作用。只重试明确可重试的事务错误，并设计幂等机制。
• 认为参数化查询能保护整个业务。参数化主要保护 SQL 值的传递方式，不会自动验证用户是否有权操作该账户，也不会替代权限控制和业务规则。
• 忽略数据库约束。应用校验能改善错误提示，但其他程序、脚本或并发请求也可能写入数据。重要的不变量应尽可能由数据库约束兜底。

九、事务设计检查清单

在实现事务前，可以逐项确认：

1. 哪些数据库变更必须共同成功或失败？事务是否只包含这些必要步骤？
2. 关键不变量能否用非空、唯一、外键、检查约束表达？
3. 并发请求同时执行时，是否可能超卖、重复插入或覆盖更新？需要条件更新、行锁还是更强隔离级别？
4. 事务发生死锁或序列化失败时，是否能够安全地有限重试？
5. 请求重复到达或提交结果未知时，是否有幂等键或唯一业务记录可用于判重？
6. 事务中是否包含网络调用、长时间计算或人工等待？能否移到提交之后，并使用发件箱或补偿流程？
7. 测试是否覆盖余额不足、账户不存在、并发更新、重复请求和中途异常？

结语

事务是维护数据库一致性的基础工具，但正确性来自整体设计，而不只是调用一次 BEGIN。要明确哪些操作必须原子地完成，用约束表达重要数据规则，用条件更新或锁处理并发，并为死锁、序列化失败和重复请求设计有边界的恢复策略。涉及邮件、支付和消息队列等外部副作用时，还要采用幂等、发件箱或补偿流程。这样设计出的事务才既能保护数据，也能在真实并发和故障中可靠运行。