重试可以帮助恢复因为并发而被拒绝的事务，但不能保证成功，也不会自动让外部副作用变成幂等。重试单位应包含决定写入内容的所有数据库读取与应用判断。如果新事务仍使用第一次快照算出的金额或目标值，即使执行了新的开始事务命令，也没有真正重新做出适合当前数据的决策。

坏掉的样子

PostgreSQL 用 40001 表示序列化失败，用 40P01 表示死锁。收到这些明确拒绝后，可以在回滚并检查剩余预算的前提下再次尝试完整操作。唯一约束错误却可能是永久的：同一个无效标识不会因为多等一会儿就变成合法输入。某些键值确实由此前查询结果决定，需要特定业务的重算策略，不能据此将所有异常都塞进统一重试循环。

错误既可能发生在语句执行期间，也可能出现在提交阶段。服务器明确拒绝事务，与客户端等待提交确认时连接断开，是两种不同情况。后一种可能已经提交，只是确认没有传回来；盲目重放就可能重复操作。在数据库事务之外已经完成的支付、发信等动作，也不会随着数据库回滚自动撤销。循环本身既不能回收这些副作用，也不能凭空识别重复。

快速验证

准备 Linux 或 macOS 上的 Python 3，以及可通过命令路径运行的 PostgreSQL 18 客户端 psql。设置 PGDATABASE、PGHOST 及必要连接参数，明确指向允许创建模式的可丢弃数据库。提前配置好认证，示例禁止交互式密码提示。把下面的程序保存为 retry-demo.py，再执行 python3 retry-demo.py。它创建并删除自己的实验模式；如果同名模式已经存在，初始化会失败，不能拿已有业务模式充当实验数据。

import os
import random
import re
import selectors
import subprocess
import time

PSQL = ['psql', '-X', '-w', '-qAt', '-v', 'ON_ERROR_STOP=1', '-v', 'VERBOSITY=sqlstate']
ENV = dict(os.environ, PGCONNECT_TIMEOUT='2')


def run(sql):
    return subprocess.run(PSQL, input=sql, text=True, capture_output=True,
                          check=True, timeout=5, env=ENV).stdout.strip()


def read_value(worker):
    deadline = time.monotonic() + 5
    data = bytearray()
    with selectors.DefaultSelector() as ready:
        ready.register(worker.stdout, selectors.EVENT_READ)
        while b'\n' not in data:
            remaining = deadline - time.monotonic()
            if remaining <= 0 or not ready.select(remaining):
                raise TimeoutError('Initial read was not confirmed')
            chunk = os.read(worker.stdout.fileno(), 4096)
            if not chunk:
                raise RuntimeError('Connection ended before initial read')
            data.extend(chunk)
    return int(data.strip())


def attempt(mode, number):
    worker = subprocess.Popen(PSQL, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE, text=True, env=ENV)
    try:
        worker.stdin.write("SET statement_timeout='1s';\n"
                           'BEGIN ISOLATION LEVEL REPEATABLE READ;\n'
                           'SELECT n FROM retry_demo.counter WHERE id=1;\n')
        worker.stdin.flush()
        observed = read_value(worker)
        # A real second connection commits after this attempt's snapshot.
        if mode == 'always' or (mode == 'once' and number == 1):
            run('UPDATE retry_demo.counter SET n=n+1 WHERE id=1;')
        statement = ('INSERT INTO retry_demo.counter VALUES (1,99);' if mode == 'permanent'
                     else f'UPDATE retry_demo.counter SET n={observed + 1} WHERE id=1;')
        finish = r'''
\set ON_ERROR_STOP off
__STATEMENT__
\if :ERROR
  \echo ATTEMPT :SQLSTATE
  ROLLBACK;
\else
  COMMIT;
  \if :ERROR
    \echo ATTEMPT :SQLSTATE
  \else
    \echo ATTEMPT 00000
  \endif
\endif
'''.replace('__STATEMENT__', statement)
        output, _ = worker.communicate(finish, timeout=5)
        match = re.fullmatch(r'ATTEMPT ([0-9A-Z]{5})', output.strip())
        if worker.returncode != 0 or match is None:
            raise RuntimeError('No confirmed transaction outcome; do not replay blindly')
        return match[1], observed
    finally:
        if worker.poll() is None:
            worker.kill()
            worker.communicate()


def increment(mode):
    observations = []
    for number in range(1, 4):
        state, observed = attempt(mode, number)
        observations.append(observed)
        if state not in {'40001', '40P01'} or number == 3:
            return {'attempts': number, 'state': state, 'reads': observations}
        time.sleep(random.uniform(0, 0.01 * 2 ** (number - 1)))


run('CREATE SCHEMA retry_demo; '
    'CREATE TABLE retry_demo.counter (id integer PRIMARY KEY,n integer); '
    'INSERT INTO retry_demo.counter VALUES (1,0);')
try:
    cases = [
        ('once', {'attempts': 2, 'state': '00000', 'reads': [0, 1]}, 2),
        ('permanent', {'attempts': 1, 'state': '23505', 'reads': [0]}, 0),
        ('always', {'attempts': 3, 'state': '40001', 'reads': [0, 1, 2]}, 3),
    ]
    for mode, expected, expected_value in cases:
        run('UPDATE retry_demo.counter SET n=0 WHERE id=1;')
        result = increment(mode)
        value = int(run('SELECT n FROM retry_demo.counter WHERE id=1;'))
        assert result == expected and value == expected_value, (mode, result, value)
        print(mode, result, 'stored', value)
finally:
    run('DROP SCHEMA retry_demo CASCADE;')

程序使用真实的两条数据库连接。每次尝试都开启可重复读事务，先读计数值，再由应用算出要写入的绝对值。竞争连接可以在读取之后、更新之前提交一次增量，迫使原快照下的更新失败。这种调度刻意检查“重试有没有重新读取和计算”，而不只是看程序有没有再次发送更新语句。

第一种情况只制造一次冲突。预期尝试两次，最终状态成功，两次读取依次为零和一，数据库最终存储二。其中一次增量来自竞争连接，另一次来自应用成功的重试。被拒绝的第一次尝试没有贡献任何提交。如果第二次只是复用原先算好的值，就不能保留两次合法增量的结果，因此最终数据也是判断重试边界的重要证据。

第二种情况制造永久错误：向已有主键再次插入相同标识。预期只尝试一次，状态为 23505，读取零，存储仍为零。分类器应立即停止，而不是把三次机会全部浪费在未改变的无效输入上。此时不能报告应用完成了加一，也不能仅因为客户端没有崩溃就把业务状态改成成功。

第三种情况在每次尝试中都安排竞争者先提交。预期尝试三次后退出，最终仍是 40001，读取依次为零、一、二，存储值为三。这里的三次增量全部属于竞争连接，应用自己的操作一次也没有提交。这是达到预算上限的失败结果，不是“重试终于成功”。断言同时检查控制流程报告与持久化数据，因为单看表里的数字无法知道是哪一方完成了操作。

让错误处理覆盖完整决策

示例从客户端取得结构化错误码，再输出专用结果标记，没有根据本地化错误文字猜测类型。预期语句错误后执行回滚，提交阶段则单独检查。若缺少确认、进程异常退出或传输等待超时，异常会离开重试分类器，不能被随意当作一次可安全重放的并发拒绝。首次读取与进程完成分别设置了等待边界，避免实验无期限挂在一个阶段。

真实应用应使用数据库驱动的事务接口和结构化异常接口，而不是直接复制这个进程调度工具。请求总预算还应包含获取连接、全部尝试以及退避时间；调用方取消后要停止继续重放，并记录尝试次数和最终错误。随机退避有助于打散同时重试的请求，却不能修复永久占用、保证公平，或让有限次数必然成功。

对外部可见操作，应使用稳定操作标识和可查询的持久化结果。提交结果不确定时，先核对结果，再决定是否重新发起非幂等动作。不可逆副作用不应直接放进会被多次执行的回调，除非相应的交付协议已经明确处理重复。让一次失败尝试可以正确退出、让重复决策基于新状态、让未知结果得到核实，才构成可靠恢复；循环次数和短暂睡眠只是其中负责安排时间的部分。

来源