MULTI/EXEC 能阻止别的客户端命令插进队列的执行过程，却不会自动保护应用此前已经读到的值。如果两个购买方先读到相同库存，再各自写回计算结果，即使每次 SET 都放进事务，也仍然可能丢失一次更新。问题出在读取与提交之间的空隙，而不是 SET 执行得是否足够连续。

先监视，再读取依赖

WATCH 让 EXEC 变成有条件的执行：被监视的键从建立监视之后不能发生修改。因此，必须先 WATCH，再读取参与决策的数据。若键在提交前改变，Redis 拒绝整个队列并返回空响应；这只是不执行自己的队列，不会撤销竞争方已经完成的修改。

监视与事务状态都属于同一条连接。redis-py 流水线在 watch 之后保留连接，multi 之前的 GET 会立即发出，应用可以读取结果并做判断；multi 之后才转为缓冲命令，等待 execute。用上下文管理器包住一次尝试，可以在结束时重置流水线状态并释放专用连接。

凡是会影响决策的值或“不存在”条件，都要纳入监视范围。只监视库存，却忽略另一个键保存的额度，并不能保护额度判断。WATCH 也不只是比较最终字节：即便 SET 写回相同内容，也可以使监视失效；原来不存在的被监视键被创建，同样会导致冲突。本次版本中，由过期和内存淘汰引起的变化也计算在内。

重试整个决策，而不是旧计算结果

收到明确的冲突结果后，应建立新监视，重新读取依赖，再检查业务条件和计算写入值。继续使用旧库存，相当于把同一个错误判断再执行一次。EXEC 无论成功还是因监视冲突取消，都会清除监视；如果应用读完后决定放弃，也可以通过 UNWATCH 主动解除。

重试需要次数或时间上限，并让调用方知道何时已耗尽预算。热点键上的大量冲突会放大成本，即使最终没有丢数据也会增加延迟。退避可以缓和竞争，却补不上遗漏的监视键。若原生命令已经能表达所需的原子操作，应优先使用它；实验刻意拆开库存算术，是为了展示旧值覆盖问题。

监视检查通过，也不会让事务获得自动回滚能力。EXEC 内部某一条命令运行失败，其它成功命令仍可能已经生效。应校验相关数据模型并逐项检查响应，不要把数个互不相干的副作用放在一起，就称为业务上的全成全败。

坏掉的样子

先读后 WATCH，会漏掉建立监视前的那段竞争窗口。用一条连接监视、另一条连接提交，不会保护后者的队列。在重试循环中发邮件或发起支付，则可能在 Redis 拒绝本次尝试之前，就已经产生了无法随队列取消的外部效果。

还必须区分服务端明确报告的冲突与传输故障。redis-py 8.1.0 在监视期间遇到连接异常，也可能抛出 WatchError；仅凭异常类型，不能断言 EXEC 从未执行。提交之后应答丢失，需要核对业务结果或依赖幂等协议，不能无条件重新执行不可逆动作。

实验用第二条连接主动制造冲突，并断言固定版本客户端的冲突消息。这个消息比较是测试断言，不是生产错误分类方案。实验关闭网络自动重试，避免隐式重新提交干扰观察。Redis Cluster 中普通多键事务还要求相关键位于同一哈希槽，WATCH 不会额外提供跨分片事务协调器。

快速验证

需要具备 C 编译器、make、curl、tar，以及 Python 3.10 或更新版本的 POSIX 环境。此前记录在 macOS、Python 3.14.4、Redis 8.10.1 与 redis-py 8.1.0 上执行；本轮没有重跑。下面只在临时目录编译使用 libc 的 Redis 核心，并将客户端安装到虚拟环境，不安装系统服务。

redis_lab=$(mktemp -d)
curl --fail --location --output "$redis_lab/redis.tar.gz" https://download.redis.io/releases/redis-8.10.1.tar.gz
tar -xzf "$redis_lab/redis.tar.gz" -C "$redis_lab"
make -C "$redis_lab/redis-8.10.1" build redis -j2 MALLOC=libc
export REDIS_SERVER="$redis_lab/redis-8.10.1/src/redis-server"
python3 -m venv "$redis_lab/venv"
"$redis_lab/venv/bin/python" -m pip install redis==8.1.0

在同一终端使用 "$redis_lab/venv/bin/python" 运行下方代码。它在 /tmp 创建私有 Unix 套接字，关闭 TCP 和持久化；结束时仅停止自己启动的进程并清理运行目录。下载阶段需要联网，实验不会连接已有 Redis 实例。

from contextlib import contextmanager
import os
from pathlib import Path
import subprocess
from tempfile import TemporaryDirectory
import time

import redis
from redis.backoff import NoBackoff
from redis.retry import Retry


@contextmanager
def instance():
    with TemporaryDirectory(prefix="redis-lab-", dir="/tmp") as directory:
        endpoint = str(Path(directory) / "redis.sock")
        process = subprocess.Popen([
            os.environ["REDIS_SERVER"], "--port", "0",
            "--unixsocket", endpoint, "--unixsocketperm", "700",
            "--save", "", "--appendonly", "no", "--dir", directory,
            "--maxmemory-policy", "noeviction",
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        client = redis.Redis(unix_socket_path=endpoint, decode_responses=False,
                             socket_timeout=2, retry=Retry(NoBackoff(), 0))
        try:
            deadline = time.monotonic() + 5
            while True:
                assert process.poll() is None, "Redis exited during startup"
                try:
                    if client.ping():
                        break
                except redis.ConnectionError:
                    if time.monotonic() >= deadline:
                        raise
                    time.sleep(0.02)
            yield client
        finally:
            client.close()
            if process.poll() is None:
                process.terminate()
            try:
                _, errors = process.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.communicate()
                raise
            assert process.returncode == 0, errors.decode()


with instance() as client, client.client() as rival:
    client.set("stock", 2)
    stale = int(client.get("stock"))
    assert rival.decr("stock") == 1
    with client.pipeline(transaction=True) as transaction:
        transaction.set("stock", stale - 1)
        assert transaction.execute() == [True]
    assert client.get("stock") == b"1"
    print("a stale value written inside MULTI/EXEC still loses another decrement")

    client.set("stock", 2)
    conflicts = 0
    for attempt in range(2):
        with client.pipeline(transaction=True) as transaction:
            transaction.watch("stock")
            current = int(transaction.get("stock"))
            assert current > 0
            if attempt == 0:
                assert rival.decr("stock") == 1
            transaction.multi()
            transaction.set("stock", current - 1)
            transaction.set("reservation", "accepted")
            try:
                responses = transaction.execute()
            except redis.WatchError as error:
                assert str(error) == "Watched variable changed."
                conflicts += 1
                assert client.get("stock") == b"1"
                assert client.get("reservation") is None
                continue
            assert responses == [True, True]
            break
    else:
        raise AssertionError("both attempts conflicted")
    assert conflicts == 1 and attempt == 1
    assert client.get("stock") == b"0"
    assert client.get("reservation") == b"accepted"
    print("WATCH rejects the stale queue; the second attempt rereads 1 and commits 0")

    client.set("watched", "unchanged")
    with client.pipeline(transaction=True) as transaction:
        transaction.watch("watched")
        assert transaction.get("watched") == b"unchanged"
        rival.set("watched", "unchanged")
        transaction.multi()
        transaction.set("same-value-marker", "yes")
        try:
            transaction.execute()
        except redis.WatchError as error:
            assert str(error) == "Watched variable changed."
        else:
            raise AssertionError("same-value SET did not invalidate WATCH")
    assert client.get("same-value-marker") is None
    print("WATCH detects a SET even when the resulting bytes are unchanged")

    with client.pipeline(transaction=True) as transaction:
        transaction.watch("not-yet-created")
        assert transaction.get("not-yet-created") is None
        rival.set("not-yet-created", "now-present")
        transaction.multi()
        transaction.set("absent-marker", "yes")
        try:
            transaction.execute()
        except redis.WatchError as error:
            assert str(error) == "Watched variable changed."
        else:
            raise AssertionError("creating a watched key did not invalidate WATCH")
    assert client.get("absent-marker") is None
    print("WATCH also protects a read of an absent key against later creation")

第一组先读到库存二，让竞争连接扣成一，再把根据旧值计算出的一写回事务。两个调用方都想扣一次，最后却只从二变成一。这段明确安排的执行顺序说明：不被插队的提交，修复不了事务之前已经过期的读取结果。

第二组先监视再读。故意插入的竞争扣减使第一次队列被拒绝，预订标记仍不存在，而对手把库存改成一的结果保留。第二次尝试重新监视并读到一，再一起写入零和预订标记。循环最多两次，每一次是否冲突以及最终值都由断言检查，不依赖碰运气产生竞态。

最后两组验证容易漏想的条件：写回相同字节，以及创建原本不存在的监视键。两种情况都会阻止队列里的标记落地。代码打印的四条摘要都来自响应和最终存储状态验证，而不是只看有没有抛异常。

这次验证覆盖隔离实例上的乐观冲突处理，不证明调度公平、预订恰好执行一次、网络故障恢复、持久化或集群切换。实际预订仍需要自己的业务标识，以及对提交结果未知的明确处理方式；示例中的标记键只是便于观察，不能直接当作完整订单协议。

来源