跳到主要内容

关系型数据、SQL 与事务

订单系统要同时回答两类问题：某位客户一共买了多少，以及两个请求同时扣库存时还能剩多少。前者需要把表里的记录正确关联起来，后者需要让更新在并发下仍然成立。表结构决定每条记录代表什么，约束守住合法状态，索引帮助查找，事务把相关修改一起提交。

PostgreSQL 教程可作为 SQL 的起点。下面用一个虚构的小商店，把这些概念放进同一组数据。

行、键、关系与表结构​

按 PostgreSQL 的关系型数据概念，表由行和有名称、类型的列组成。这里 customers 的一行代表一位客户，orders 的一行代表一笔订单；表结构（schema）还说明列能否为空、如何标识记录，以及哪些关系必须成立。

主键唯一标识一行，值不能为 NULL。名字会重复、会改变，所以用 customer_id 标识客户。订单中的 customer_id 通过外键引用客户：本例要求每笔订单恰好属于一位已存在的客户，一位客户可以有零笔或多笔订单。这是客户到订单的一对多关系。

金额用整数分记录：5000 分等于 50.00 个货币单位。先把下面的代码存入 relational_demo.py，后续代码按页面顺序接在同一文件里，用 python3 relational_demo.py 运行。Python 的 sqlite3让这组 SQL 不需要单独的数据库服务器；isolation_level=None 在这里用于显式控制事务。SQLite 的外键检查需要在连接上启用，代码也检查了启用结果。示例执行在内存数据库中。

SQLite 普通表中的 INTEGER 声明只有类型亲和性，仍可能存入小数或非数字文本。因此本例在金额、库存和预订数量的 CHECK 中用 SQLite 的 typeof 要求存储值为整数，再检查正数或非负数条件。

import sqlite3

# Explicit transaction control; the database lives only in memory.
db = sqlite3.connect(":memory:", isolation_level=None)
db.execute("PRAGMA foreign_keys = ON")
assert db.execute("PRAGMA foreign_keys").fetchone() == (1,)
db.executescript("""
CREATE TABLE customers (
    customer_id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE
);
CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY,
    customer_id INTEGER NOT NULL REFERENCES customers(customer_id),
    amount_cents INTEGER NOT NULL
        CHECK (typeof(amount_cents) = 'integer' AND amount_cents > 0)
);
INSERT INTO customers VALUES
    (1, 'Ada', 'ada@example.com'),
    (2, 'Lin', NULL),
    (3, 'Sam', 'sam@example.com');
INSERT INTO orders VALUES
    (101, 1, 5000), (102, 1, 3000),
    (103, 2, 2000), (104, 2, 1000);
""")
print("customers:", db.execute("SELECT COUNT(*) FROM customers").fetchone()[0])
print("orders:", db.execute("SELECT COUNT(*) FROM orders").fetchone()[0])
customers: 3
orders: 4

Ada 的两笔订单分别是 5000、3000 分；Lin 的两笔是 2000、1000 分；Sam 没有订单。Lin 的邮箱尚未提供，用 NULL 表示。这样既能计算总额，也能检查没有匹配记录时查询会怎样。

查询、连接与聚合​

SELECT 与 WHERE分别选择返回的列和满足条件的行。连接（join）按条件配对两张表的行：内连接只保留匹配，左连接还保留左表中未匹配的行，右侧列补为 NULL。

要列出每位客户的订单数和消费总额，从客户表做左连接，才能保留 Sam。GROUP BY把同一客户的记录归为一组；SUM 加总金额。按聚合函数的规则，COUNT(*) 数行，COUNT(o.order_id) 只数非空订单 ID。Sam 在左连接后有一行占位记录，因此前者是 1，后者是 0。没有非空金额可加时，SUM 得到 NULL；本查询用 COALESCE 将它显示为 0。

query = """
SELECT c.customer_id, c.name,
       COUNT(*) AS joined_rows,
       COUNT(o.order_id) AS order_count,
       COALESCE(SUM(o.amount_cents), 0) AS total_cents
FROM customers AS c
LEFT JOIN orders AS o ON o.customer_id = c.customer_id
GROUP BY c.customer_id, c.name
ORDER BY c.customer_id;
"""
for row in db.execute(query):
    print(row)
print("orders >= 3000:", db.execute(
    "SELECT order_id FROM orders WHERE amount_cents >= 3000 ORDER BY order_id"
).fetchall())
(1, 'Ada', 2, 2, 8000)
(2, 'Lin', 2, 2, 3000)
(3, 'Sam', 1, 0, 0)
orders >= 3000: [(101,), (102,)]

输出列依次是客户 ID、名字、连接后的行数、订单数、总额（分）。Ada 的总额为 5000 + 3000 = 8000，Lin 为 2000 + 1000 = 3000；全店共 11000 分，即 110.00 个货币单位。最后一条查询筛出金额至少 3000 分的订单 101 和 102。WHERE 在聚合前筛选行，HAVING 在聚合后筛选组；例如只列总额达到 4000 分的客户，就对总额使用 HAVING，本例只会留下 Ada。需要稳定顺序时要写 ORDER BY。

NULL 与连接基数​

NULL 的比较规则引入第三种逻辑结果“未知”：普通比较只要有一端为 NULL，结果就为未知；WHERE 只留下结果为真的行。因此 email = NULL 找不到缺失邮箱，要用 email IS NULL。同样，等值连接中的两个空键不会相互匹配。

连接基数指连接后会产生多少行。客户 ID 在客户表中唯一，所以每笔订单最多匹配一位客户。若拿订单去连接一张每位客户有多条记录的表，同一笔订单就可能出现多次。下面临时构造 Ada 的两条促销标签：两笔订单各匹配两条标签，共四行，订单总额被重复加到了 16000 分。

print("= NULL:", db.execute(
    "SELECT COUNT(*) FROM customers WHERE email = NULL"
).fetchone()[0])
print("IS NULL:", db.execute(
    "SELECT COUNT(*) FROM customers WHERE email IS NULL"
).fetchone()[0])
print("empty SUM:", db.execute(
    "SELECT SUM(amount_cents) FROM orders WHERE customer_id = 3"
).fetchone()[0])
print("multiplied:", db.execute("""
WITH promotions(customer_id, label) AS (
    VALUES (1, 'spring'), (1, 'member')
)
SELECT COUNT(*), SUM(o.amount_cents)
FROM orders AS o
JOIN promotions AS p ON p.customer_id = o.customer_id;
""").fetchone())
= NULL: 0
IS NULL: 1
empty SUM: None
multiplied: (4, 16000)

None 是 Python 打印 SQL NULL 时的表示。正确的 Ada 订单总额仍是 8000 分；16000 分回答的是“按促销标签展开后的金额和”，不能直接当作消费总额。先明确查询应当一行代表客户、订单还是标签，再选择连接或先做聚合。SUM(DISTINCT amount_cents) 也不能通用地修正重复，因为两笔真正不同的订单可能金额相同。DataFrame 中如何声明连接关系、查出未匹配键，见 Pandas 合并；Pandas 的空键匹配行为需要与 SQL 区分。

约束：让数据库拒绝非法状态​

约束把“必须一直成立的规则”写进表定义。这里的订单主键禁止重复 ID，外键禁止不存在的客户，NOT NULL 要求提供金额，金额的 CHECK 要求存储值为正整数。它们对所有正常写入路径生效，避免每个调用者各自实现一遍检查。

for label, values in [
    ("foreign key", (105, 99, 1000)),
    ("positive amount", (106, 1, -1000)),
    ("required amount", (107, 1, None)),
    ("unique order", (101, 1, 1000)),
]:
    try:
        db.execute("INSERT INTO orders VALUES (?, ?, ?)", values)
    except sqlite3.IntegrityError:
        print(label, "rejected")
print("orders:", db.execute("SELECT COUNT(*) FROM orders").fetchone()[0])
foreign key rejected
positive amount rejected
required amount rejected
unique order rejected
orders: 4

四次插入都被拒绝，原有四笔订单仍在。数据库执行了规则，而不只是把规则留在注释里。

PostgreSQL 的约束文档还说明：CHECK 的结果为真或 NULL 都会通过，所以正数检查仍需配合 NOT NULL。PostgreSQL 默认的 UNIQUE 允许多个 NULL；若业务要求每位客户都有唯一邮箱，要同时要求邮箱非空。跨行规则也有边界：不能用普通 CHECK 持续保证“所有订单金额之和不超过某个额度”，这需要专门设计约束、事务或锁定相关状态。

索引的读写取舍​

索引保存额外的查找结构，并占用独立的存储空间。它可能加快筛选和连接，但写入时要保持索引与表同步。下面为按客户查订单建立索引：

db.execute("CREATE INDEX orders_customer_idx ON orders(customer_id)")
print("index exists:", db.execute(
    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = ?",
    ("orders_customer_idx",),
).fetchone()[0] == 1)
index exists: True

这个输出只确认索引已创建。四笔订单不能说明性能收益；索引使用文档说明，小数据集可能更适合顺序扫描。根据实际过滤条件、连接条件和查询计划决定是否加索引，再衡量写入成本。普通索引不要求键唯一，也不会让两次独立更新自动变成一个事务。

事务、隔离与丢失更新​

事务的原子性让同一事务中的相关行修改一起提交，或在整个事务回滚时一起撤销。BEGIN 开始，COMMIT 提交，ROLLBACK 撤销。这里讨论的是表中行的修改；PostgreSQL 的序列计数器即使在事务失败后也不会退回原值。原子性处理一组修改的成败；隔离则决定并发事务能看到什么、冲突时如何处理。

假设库存起初为 10，请求 A、B 各买一件，都先读取库存，再在应用里减 1。下面是在 PostgreSQL 的 Read Committed 下可能发生的顺序；A、B 各自把读写放在一个事务中：

步骤A 的事务B 的事务已提交库存
1开始并读到 10开始并读到 1010
2写入计算好的 9，提交仍保存旧值 109
3已结束写入计算好的 9，提交9

两次购买后应剩 10 - 1 - 1 = 8，却剩 9：B 覆盖了 A 的结果，称为丢失更新。值仍满足 quantity >= 0，因此行约束发现不了这个业务错误。

下面用同一连接依次执行两个保存了旧值的请求，重现覆盖结果，再与数据库内的相对更新比较。这是确定性的旧值覆盖演示，不是两个并发连接的隔离实验。

db.execute("""
CREATE TABLE stock (
    sku INTEGER PRIMARY KEY,
    quantity INTEGER NOT NULL
        CHECK (typeof(quantity) = 'integer' AND quantity >= 0)
)
""")
db.execute("INSERT INTO stock VALUES (1, 10)")
a_read = db.execute("SELECT quantity FROM stock WHERE sku = 1").fetchone()[0]
b_read = db.execute("SELECT quantity FROM stock WHERE sku = 1").fetchone()[0]
for old_value in (a_read, b_read):
    db.execute("UPDATE stock SET quantity = ? WHERE sku = 1", (old_value - 1,))
print("stale writes:", db.execute("SELECT quantity FROM stock").fetchone()[0])

db.execute("UPDATE stock SET quantity = 10 WHERE sku = 1")
for _ in range(2):
    changed = db.execute("""
        UPDATE stock SET quantity = quantity - 1
        WHERE sku = 1 AND quantity >= 1
    """).rowcount
    assert changed == 1
print("relative updates:", db.execute("SELECT quantity FROM stock").fetchone()[0])
stale writes: 9
relative updates: 8

PostgreSQL 的 Read Committed 规则说明：更新同一行时，后来的更新会等待仍未结束的更新者；对方提交后，会基于新行版本重新检查 WHERE。所以本例应在数据库内计算 quantity - 1，并用 quantity >= 1 作为同一条更新的条件。影响行数为 0 表示没有匹配的可扣库存，调用者必须处理这个结果。

按 PostgreSQL 的隔离级别，还要区分这些保证：

级别可依赖的行为
Read Uncommitted在 PostgreSQL 中等同于 Read Committed
Read Committed（默认）普通查询用语句开始时的已提交快照，也能看到本事务之前的修改
Repeatable Read从首条非事务控制语句开始保持快照，也能看到自身修改；仍可能出现序列化异常
Serializable成功提交的事务效果等价于某种逐个执行的顺序

Repeatable Read 和 Serializable 下都要准备因序列化失败而重试整个事务。需要先读再决定如何更新时，可以在同一事务里锁定相关行，例如 PostgreSQL 的 SELECT FOR UPDATE；涉及多行的业务规则，还要选择能覆盖整个规则的隔离或锁定方案。

数据库提交与外部副作用​

事务发件箱文档指出，数据库修改与发通知分开执行会留下两个失败窗口：先发通知、后提交，数据库回滚后通知却已发出；先提交、后发通知，进程可能在两步之间退出。数据库回滚无法撤回外部服务已接受的请求。

事务发件箱（transactional outbox）把业务修改和待发送事件写入同一数据库事务，再由另一个发送者读取已提交事件并投递。沿用剩余库存 8 的状态，下一段把扣库存和写入事件绑在一起。事件 ID 标识一次逻辑预订，重试时使用相同 ID。

db.execute("""
CREATE TABLE outbox (
    event_id TEXT PRIMARY KEY NOT NULL,
    sku INTEGER NOT NULL REFERENCES stock(sku),
    units INTEGER NOT NULL
        CHECK (typeof(units) = 'integer' AND units > 0)
)
""")

def reserve(event_id, units):
    db.execute("BEGIN")
    try:
        changed = db.execute("""
            UPDATE stock SET quantity = quantity - ?
            WHERE sku = 1 AND quantity >= ?
        """, (units, units)).rowcount
        if changed != 1:
            raise ValueError("insufficient stock")
        db.execute("INSERT INTO outbox VALUES (?, 1, ?)", (event_id, units))
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK")
        raise

def state():
    return (db.execute("SELECT quantity FROM stock WHERE sku = 1").fetchone()[0],
            db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0])

try:
    reserve("reservation-1", 0)
except sqlite3.IntegrityError:
    print("invalid event:", state())
reserve("reservation-1", 1)
print("committed:", state())
try:
    reserve("reservation-1", 1)
except sqlite3.IntegrityError:
    print("duplicate event:", state())
db.close()
invalid event: (8, 0)
committed: (7, 1)
duplicate event: (7, 1)

状态元组是“库存、发件箱事件数”。非法事件触发约束失败并回滚，库存仍是 8、事件数为 0；成功预订后库存为 7、事件数为 1。重复 ID 的插入失败，也撤销了这次扣减，状态仍为 (7, 1)。接收重复请求的应用还需要核对原始参数并返回已存结果，不能仅凭重复 ID 就假定两次请求相同。

提交后的事件仍待投递。发送者若在外部服务接受事件后、记录发送成功前退出，就可能再次发送；发件箱保存了发送意图，未消除重复投递。AWS 的幂等 API 设计说明强调，接收方应把请求 ID 和本地效果原子提交，同一 ID 的参数也要核对。调用外部 API 时，只有服务支持的幂等接口才能按其有效期复用键。响应超时时保留“结果未知”：使用服务提供的状态查询，或在其幂等保证范围内用原键重试。

长时运行 Agent把这条边界放进检查点与恢复流程；Cloudflare Workers介绍具体存储和队列的选择。选定服务后，还要核对它的事务接口与并发保证，再把这里的表结构和更新规则接入应用。

探索关联

相关讨论 (3)

引用了 (3)

被引用 (4)

同主题的其他笔记 (15)

打开关联网络