消息语义

消息语义描述的是系统发生重试、断连、进程崩溃后，仍能承诺什么。讨论“恰好一次”之前，必须明确计数对象：网络发送次数、消费者执行次数，还是数据库里最终生效的业务操作次数。这三者可以不同。

本文先用一次确认丢失解释问题，再讨论三种语义与实现机制。Kafka 的日志结构见 Kafka 架构，分区接管与 offset 管理见 消费组与协作。

问题从哪里来：没有收到确认不等于没有成功

假设消费者处理事件 E42，把库存扣减一件，然后向消息系统确认：

消息系统看到的只有“没有收到 ACK”。消费者可能根本没处理，也可能已经完成、只是确认丢了。前者需要重试，后者重试又可能扣减第二次。仅仅调长超时无法消除这个不确定窗口。

同样的问题也存在于生产者到 broker 的链路。生产者确认表示消息在 broker 侧达到某种接受条件；消费者确认表示处理进度可以推进。两种确认不能互相替代，RabbitMQ 将它们明确分为 Publisher Confirms 与 Consumer Acknowledgements。

三种语义的准确含义

“至少一次”依赖消息保留、持久化、恢复能力和重试策略。队列到期删除、重试次数耗尽或所有副本丢失，都不在无条件保证之内。“至多一次”也不是简单地关闭某一处重试就完成了：应用层重新提交同一个业务请求，仍可能产生两条不同消息。

恰好一次不要求处理函数物理上只运行一次。第一次执行可能回滚，第二次执行才提交；外界只能看到一次结果。不能直接由 FLP 推导出“恰好一次效果不可能”：FLP 讨论特定异步故障模型下确定性共识的终止性，不能替代这里对事务边界和可见结果的分析。

确认顺序如何决定故障窗口

把业务效果记为 E，确认或提交进度记为 C，两者独立时只有两种顺序：

所以，单纯交换顺序无法同时消除遗漏和重复。系统必须进一步做到：让 E 和 C 原子发生，或者允许重复 E，但把重复转换为无额外副作用的操作。这就是事务与幂等性的作用。

幂等：重复请求如何得到同一个业务结果

数学上，操作 f 幂等意味着 f(f(x)) = f(x)。把状态设为 CONFIRMED，在只考虑该字段赋值时是幂等的；把数量加一则不是。但赋值旁边若还发送邮件、扣款或写审计事件，整个处理函数未必幂等。

业务去重需要稳定的事件 ID。同一事件的重投必须复用 ID；每次重试生成新 UUID 会让去重失效。常用键为 (consumer_name, event_id)：同一事件可被不同业务消费者各处理一次。

下面是 PostgreSQL 的一个原子消费示例。前提是库存表中 product_id 唯一，处理成功定义为库存增加；消息格式校验在事务前完成。

CREATE TABLE inventory (
    product_id bigint PRIMARY KEY,
    count bigint NOT NULL
);

CREATE TABLE processed_events (
    consumer_name text NOT NULL,
    event_id text NOT NULL,
    PRIMARY KEY (consumer_name, event_id)
);

-- 参数示例：consumer_name='inventory', event_id='E42'。
BEGIN;
WITH first_delivery AS (
    INSERT INTO processed_events (consumer_name, event_id)
    VALUES ('inventory', 'E42')
    ON CONFLICT DO NOTHING
    RETURNING event_id
)
INSERT INTO inventory (product_id, count)
SELECT 42, 1 FROM first_delivery
ON CONFLICT (product_id)
DO UPDATE SET count = inventory.count + EXCLUDED.count;
COMMIT;
-- 仅在 COMMIT 成功后确认消息。

第一次插入去重键成功，库存增加；第二次 first_delivery 为空，不再修改库存。若库存更新失败，整个事务回滚，去重键也不会留下。这里依赖 PostgreSQL 的 ON CONFLICT 与 RETURNING 语义。

这个例子还揭示两个边界：去重记录与库存若位于两个独立数据库，以上原子性就不存在；如果提前清除 E42 的去重记录，旧消息再次回放就可能重新生效。去重保留期必须覆盖业务允许的重放范围。

版本条件更新也能防止覆盖，但 version mismatch 只说明状态已变化，不足以证明“这条事件已处理”。事件身份与并发版本解决的是两个不同问题。

动手观察：事务提交前后的失败，留下什么状态

用一个本地 SQLite 文件模拟库存消费者，初始库存为 100，E42 要扣减 3。这里把事件内容简化为一个整数 delta；去重记录同时保存 delta，避免同一个 ID 意外对应不同操作时被默默吞掉。真实事件还应比较商品、租户和业务参数，或比较规范化后的内容摘要。

import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory

class Interrupted(Exception):
    pass

def consume(path, event_id, delta, fail_at=None):
    db = sqlite3.connect(path, isolation_level=None)
    try:
        db.execute("BEGIN IMMEDIATE")
        inserted = db.execute(
            "INSERT INTO done VALUES (?, ?) ON CONFLICT(id) DO NOTHING",
            (event_id, delta),
        ).rowcount
        if not inserted:
            old_delta = db.execute(
                "SELECT delta FROM done WHERE id = ?", (event_id,)
            ).fetchone()[0]
            if old_delta != delta:
                raise ValueError("同一个事件 ID 对应不同内容")
        else:
            if fail_at == "after_claim":
                raise Interrupted()
            changed = db.execute(
                "UPDATE stock SET n = n + ? WHERE n + ? >= 0",
                (delta, delta),
            ).rowcount
            if changed != 1:
                raise ValueError("库存不足")
            if fail_at == "after_effect":
                raise Interrupted()
        db.execute("COMMIT")
        if fail_at == "after_commit":
            raise Interrupted()  # 结果已提交，但调用者没得到成功响应
    except Exception:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise
    finally:
        db.close()

def state(path):
    with sqlite3.connect(path) as db:
        return (db.execute("SELECT n FROM stock").fetchone()[0],
                db.execute("SELECT count(*) FROM done").fetchone()[0])

with TemporaryDirectory() as directory:
    for cut in ("after_claim", "after_effect", "after_commit"):
        path = Path(directory) / f"{cut}.sqlite"
        with sqlite3.connect(path) as db:
            db.executescript("""
                CREATE TABLE stock (n INTEGER NOT NULL CHECK(n >= 0));
                INSERT INTO stock VALUES (100);
                CREATE TABLE done (id TEXT PRIMARY KEY, delta INTEGER NOT NULL);
            """)
        try:
            consume(path, "E42", -3, cut)
        except Interrupted:
            pass
        assert state(path) == ((97, 1) if cut == "after_commit" else (100, 0))
        consume(path, "E42", -3)  # 调用者不知道上次结果，所以重试
        consume(path, "E42", -3)  # 消息再次投递
        assert state(path) == (97, 1)
        for event_id, delta in (("E42", -4), ("E43", -200)):
            try:
                consume(path, event_id, delta)
            except ValueError:
                pass
            else:
                raise AssertionError("应该拒绝内容冲突或库存不足")
        assert state(path) == (97, 1)

每次调用都重新打开数据库，观察的是提交后可读的状态。实验注入的是 Python 异常并显式回滚，展示事务边界；它没有模拟断电、磁盘故障或消息 broker。SQLite 只允许一个写事务，BEGIN IMMEDIATE 可能因其他写者返回忙；生产消费者还需要处理这类可重试失败。具体事务行为见 SQLite 事务文档。

第三行才是幂等记录真正消解的不确定窗口。若去重键先在独立事务提交，第一行会变成“已处理但库存没改”；若业务先独立提交，第二行会变成“库存已改但没有去重证据”。两张表放在同一个数据库里还不够，必须由同一个事务提交。

库存不足在这里被定义为拒绝执行，因此没有留下成功去重记录；业务系统可以把它记录为一个可查询的拒绝结果，但需要明确这个结果是否允许以后重试。若需求是让重试返回首次生成的订单号、收据或拒绝原因，还要在事务里保存该结果，仅记录一个“出现过”的 ID 不足以重建响应。

响应丢失后，重投会不会再扣一次？

把同一次 E42 的“处理尝试数”与“持久扣减次数”分别计数，才能准确讨论恰好一次效果。可以在提交前崩溃，也可以在提交后、ACK 前崩溃：前者丢弃暂存，后者保留数据库结果却让 broker 仍可重投。切换策略会重置整段历史，比较去重记录是否与效果共享原子边界。

Kafka 的幂等生产与事务

Kafka 的幂等生产把 producer 身份、epoch 与分区内序号带入协议，识别客户端发送重试产生的重复批次。这不等于按业务内容去重：应用把同一个订单重新作为一条新记录发送，仍可能被正常追加。相关约束见 Producer Configs。

事务进一步把多个分区的输出与消费进度纳入一次提交。一个 Kafka 到 Kafka 的变换可以写成：

崩溃时，事务可能尚未提交，也可能已经提交但响应丢失；恢复由事务协议处理。下游用 read_committed 排除未提交或中止事务的记录。应把“结果可见”和“恢复时从哪里继续”看作同一个原子决定。机制与限制见 Kafka Message Delivery Semantics / Transactions。

Kafka 事务不会自动包含外部 SQL 更新、HTTP 请求或邮件发送。数据库侧要使用自己的事务与去重；外部接口需要稳定幂等键，或者可查询的业务状态来消解不确定结果。

Outbox：数据库提交后，事件怎样可靠发出

“先写数据库，再发消息”也有两个独立提交点：数据库成功后进程崩溃，消息可能永远未发。反过来先发消息，又可能发布一个最后没有提交的业务事实。

Outbox 把业务变更与待发布事件写入同一数据库事务，由独立转发器读取事件并投递：

转发器在“已发送、未标记”之间崩溃仍会重发，所以 Outbox 主要解决漏发，消费去重负责重复。两者配合才形成完整链路。更一般的事务取舍见 分布式事务。

快速判断保证边界

检查一条消息链路时，依次确定：谁生成稳定事件 ID；哪一步可能重试；ACK 到底确认了什么；业务效果与进度能否共事务；外部副作用如何去重；去重记录和原消息保留多久。

如果这些问题没有明确答案，仅写“开启 exactly-once”并不能说明系统实际提供什么保证。

参考

• RabbitMQ：确认机制
• Kafka：设计与消息语义
• Kafka：生产者配置
• SQLite：事务与显式回滚
• PostgreSQL：INSERT

关键词：消息语义、确认、重试、幂等、去重、exactly-once、事务边界、Outbox