重试可以帮助恢复因为并发而被拒绝的事务，但不能保证成功，也不会自动让外部副作用变成幂等。重试单位应包含决定写入内容的所有数据库读取与应用判断。如果新事务仍使用第一次快照算出的金额或目标值，即使执行了新的开始事务命令，也没有真正重新做出适合当前数据的决策。
收到明确并发拒绝后，在统一预算内重做读取、判断与写入。永久错误或预算耗尽应停止，提交结果未知时先核对；数据库回滚不能撤销外部副作用。 坏掉的样子 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;') 程序使用真实的两条数据库连接。每次尝试都开启可重复读事务，先读计数值，再由应用算出要写入的绝对值。竞争连接可以在读取之后、更新之前提交一次增量，迫使原快照下的更新失败。这种调度刻意检查“重试有没有重新读取和计算”，而不只是看程序有没有再次发送更新语句。
...
