事务性发件箱

概念定义

发件箱模式在服务自己的数据库中增加一张表。当请求改变业务状态时，同一事务插入一条或多条描述待发出事件的发件箱行。由于两次写入共享一个事务，它们一起提交或一起回滚，因此绝不会出现状态改变却没有相应排队消息的情况。另有一个中继轮询或流式读取发件箱，发布每条待发消息并将其标记为已投递。

为什么重要

没有它时，服务既要写数据库又要向消息代理发布，两步之间的任何失败都会导致不同步：状态变了却没人被告知，或一条消息宣称了一个已回滚的变更。重试也修不好，因为存储与消息之间没有分布式事务。发件箱把双重写入压缩成一次原子的本地步骤，再用一个会重试的中继处理不可靠的那部分——发布。

工程实践

在与领域变更相同的事务中写入发件箱行，并让中继先发布再标记、绝不反过来，使崩溃导致重发而非丢失。通过对行排序保持按聚合的顺序，并清理已投递的行以免表无界增长。因为投递是至少一次，要为消费者配备幂等键或去重，使重复消息不重复其副作用。许多团队用对发件箱表的变更数据捕获来替代轮询中继。

代码示例

from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any


class TransactionAborted(RuntimeError):
    pass


@dataclass
class Store:
    data: dict[str, Any] = field(default_factory=dict)
    commits: int = 0
    rollbacks: int = 0

    @contextmanager
    def transaction(self) -> Iterator["Transaction"]:
        snapshot = dict(self.data)
        tx = Transaction(self)
        try:
            yield tx
        except BaseException:
            self.data = snapshot
            self.rollbacks += 1
            raise
        else:
            self.commits += 1


@dataclass
class Transaction:
    store: Store

    def set(self, key: str, value: Any) -> None:
        self.store.data[key] = value

    def delete(self, key: str) -> None:
        self.store.data.pop(key, None)


def main() -> None:
    store = Store(data={"balance": 100})
    with store.transaction() as tx:
        tx.set("balance", 80)
        tx.set("currency", "USD")
    print(store.data, store.commits, store.rollbacks)

    try:
        with store.transaction() as tx:
            tx.set("balance", 0)
            raise TransactionAborted("payment declined")
    except TransactionAborted as error:
        print("rolled back:", error)
    print(store.data, store.commits, store.rollbacks)


if __name__ == "__main__":
    main()