数据库、消息与并发测试：把故障窗口变成可重复实验

数据库、消息与并发测试高清原图

计数器当前为 10。两个事务各读到 10，各自计算 11，再分别执行成功的 UPDATE，最后仍可能只留下 11。单线程连续调用两次不会产生这个交错，重复跑很多次也未必碰到它。测试需要控制的是“两个参与者都读完、还没有开始写”的时刻。

为状态变化准备真实参与者

数据库测试需要哪些对象

验证并发写入，至少要区分工作连接、事务与观察连接：

JUnit 测试线程
├── Worker A → JDBC Connection A → Transaction A
├── Worker B → JDBC Connection B → Transaction B
├── CyclicBarrier(2)：两个工作线程读完后共同继续
└── 两个 Future 完成后 → 新 Connection C → 查询已提交结果

同一 Connection 上的两个 Java 方法不构成两个独立数据库事务；在调用测试的事务里查询，也可能读到尚未提交的本地结果。使用真实连接并明确提交点，才能检查锁等待、唯一约束、隔离级别和其他事务的可见性。

数据库的方言、扩展、排序、JSON 操作和约束会影响结果。以目标 PostgreSQL 为例，使用内存 Map 或另一种内存数据库来替代，就无法执行同样的并发控制规则。ORM 测试还要留意一级缓存与 flush，相关实验见 Spring 集成测试。

测试隔离可以按库、schema、表内命名空间或单条业务 ID 组织。共享库时删除整张表会影响并行用例；使用独立 ID 并按 ID 清理，更容易控制对象归属。数据库容器按类隔离则付出更多启动成本。

运行数据库与 RabbitMQ 实验

下载 数据库、消息与并发工程。Java 编译目标 17，可在 Java 17/25 上执行；依赖 JUnit 6.0.3、Testcontainers 2.0.5、PostgreSQL 18.6、RabbitMQ 4.3.5-management。JDBC 使用 42.7.10，RabbitMQ Java Client 使用 5.28.0。

src/main/java/example/testing/
├── CounterStore.java        三种实际 SQL 更新路径
└── InboxProcessor.java      Inbox 与业务更新的本地事务

src/test/
├── resources/schema.sql     counter、inbox 表与约束
└── java/example/testing/
    ├── TestSupport             连接、屏障、Future 与观察查询
    ├── DatabaseConcurrencyTest 四个数据库场景
    ├── MessageRedeliveryTest   三个提交、重投及版本场景
    └── LostUpdateAssumptionCase 单独运行的错误预期

Linux 原生 Docker、默认 Unix socket 下，用有权访问开发 daemon 的普通宿主用户执行：

unzip testing-state-lab.zip
cd testing-state
mkdir -p .m2
LAB_DIR="$(pwd -P)"
MAVEN_IMAGE='maven:3.9.12-eclipse-temurin-25'
docker version
test -S /var/run/docker.sock
SOCKET_GID="$(stat -c '%g' /var/run/docker.sock)"
docker pull "$MAVEN_IMAGE"
docker pull postgres:18.6
docker pull rabbitmq:4.3.5-management

docker run --rm --user "$(id -u):$(id -g)" --group-add "$SOCKET_GID" \
  -e HOME=/tmp -e MAVEN_CONFIG=/tmp/.m2 \
  --mount "type=bind,src=$LAB_DIR,dst=/work" \
  --mount "type=bind,src=$LAB_DIR/.m2,dst=/m2" \
  --mount type=bind,src=/var/run/docker.sock,dst=/var/run/docker.sock \
  --workdir /work "$MAVEN_IMAGE" \
  mvn -B -ntp -Duser.home=/tmp -Dmaven.repo.local=/m2 clean verify

Maven 使用宿主 UID/GID 写入当前用户创建的源码和缓存目录，socket 组权限另外提供。Docker socket 授予的是管理 daemon 的能力，应只在获准的测试环境使用。Docker Linux 权限说明

Docker Desktop 的容器化 Maven 还要增加 -e TESTCONTAINERS_HOST_OVERRIDE=host.docker.internal，让测试 JVM 能访问兄弟容器的映射端口。rootless 和远程 daemon 应按实际 socket/地址配置，不能照抄 Desktop 主机名。容器内运行 Testcontainers

已有本机 Maven/JDK 时直接执行相同 clean verify，测试进程仍须能访问 daemon。正常结果为 7 次执行，零失败、零错误、零跳过。数据库类运行 4 次，消息与状态类运行 3 次；报告位于 target/surefire-reports。首次运行需要拉取 Testcontainers 辅助镜像，受限网络需先配置获准镜像源与 Maven 依赖缓存。

让丢失更新稳定出现，再比较修复

屏障放在读完之后

CounterStore.increment 为每个调用打开自己的连接，关闭 autocommit，查询 value/version，然后等待屏障：

try (Connection c = connections.open()) {
    c.setAutoCommit(false);
    // 执行 SELECT value, version FROM counter WHERE id=?
    bothRead.await(5, TimeUnit.SECONDS);
    // 按指定模式执行 UPDATE，检查影响行数，再 commit。
}

工程里 SELECT、UPDATE、commit 和异常 rollback 都是完整 JDBC 调用。两个线程都通过屏障后才允许写入，因此它们的本地 readValue 都是 10。等待上限、SQL lock timeout、statement timeout 和 Future.get 上限分别限制不同的阻塞位置。

读改写路径执行的是 UPDATE counter SET value=? WHERE id=?，参数值由 Java 已经算成 11。实际交错如下：

时间       连接 A                     连接 B
1          SELECT → 10                SELECT → 10
2          等待屏障                    等待屏障
3          屏障释放                    屏障释放
4          UPDATE value=11             UPDATE value=11，可能等待 A 的行锁
5          COMMIT                      继续执行自己的 value=11
6                                      COMMIT
7          观察连接读取最终 value=11

在 PostgreSQL READ COMMITTED 下，后来的 UPDATE 可以等待前一个更新提交，再处理最新行版本，但传入的常量仍是 11。测试断言两个 UPDATE 都影响 1 行、两个线程最初都读到 10、最终值为 11。这是对错误算法实际行为的观察。PostgreSQL 事务隔离

让业务预期变成一次失败

如果业务要求两次成功增加后应为 12，选择 LostUpdateAssumptionCase 会出现真实断言失败。它执行相同双连接交错，仅把最终预期设为 12：

set +e
docker run --rm --user "$(id -u):$(id -g)" --group-add "$SOCKET_GID" \
  -e HOME=/tmp -e MAVEN_CONFIG=/tmp/.m2 \
  --mount "type=bind,src=$LAB_DIR,dst=/work" \
  --mount "type=bind,src=$LAB_DIR/.m2,dst=/m2" \
  --mount type=bind,src=/var/run/docker.sock,dst=/var/run/docker.sock \
  --workdir /work "$MAVEN_IMAGE" \
  mvn -B -ntp -Duser.home=/tmp -Dmaven.repo.local=/m2 \
  -Dtest=LostUpdateAssumptionCase test > lost-update.log 2>&1
status=$?
set -e
test "$status" -ne 0
grep -F 'expected: <12> but was: <11>' lost-update.log

Docker Desktop 同样需要在这条容器命令中增加前述 host override。若日志是连接拒绝或镜像失败，说明实验尚未到达预期断言，不能将任意非零退出算作丢失更新。

原子 SQL 与乐观条件各解决什么

原子增加把计算放回数据库：

UPDATE counter SET value = value + 1 WHERE id = ?;

两个 UPDATE 仍会竞争行锁，但后续更新基于它实际获得的当前行值进行计算。工程检查总影响行数为 2，最终值为 12。对于余额扣减，还可把余额充足条件写入 WHERE，并检查影响行数；涉及多表规则时要把整个事务一起考虑。

乐观路径保留读改写，同时添加版本条件：

UPDATE counter
SET value = ?, version = version + 1
WHERE id = ? AND version = ?;

两个参与者最初都读取 version 0，其中一个更新并提交 version 1；另一个的旧版本条件匹配失败，影响行数为 0。工程先断言只有一次更新成功、值为 11，再重新读取当前值和版本，执行一次新的条件更新，最终得到 12。

影响 0 行需要由业务处理为冲突、重读或有限重试。若原操作还包含网络调用，重试整个函数可能重复产生外部效果；应区分可重新计算的数据库状态与已经发出的操作。

悲观锁也可以通过 SELECT ... FOR UPDATE 串行化读改写。此时第二个读取要等第一个释放锁，不能再要求“双方都持锁读完后一起过屏障”，否则测试自己就制造了无法满足的等待。锁类型、冲突和死锁规则见 PostgreSQL 显式锁。

隔离级别和失败类别要一起记录

把隔离级别提高到 REPEATABLE READ 或 SERIALIZABLE，某些交错会转为事务失败，需要重试完整事务。测试既要检查最终结果，也要检查哪一个事务被拒绝以及错误类别。它与应用版本条件返回 0 行的处理路径不同。

工程额外执行真实唯一键冲突，并检查 SQLState 23505。生产诊断还常见序列化失败 40001、死锁 40P01 和锁不可用 55P03。优先使用稳定分类，不要只匹配驱动的本地化文案。PostgreSQL 错误码

事务失败后先 rollback，再开启新的尝试。记录输入、事务隔离、首次读取值、更新行数、提交或拒绝结果，可以直接解释一次并发历史。只留最终值会丢掉“哪一步没有发生”的信息。

在提交与消息确认之间制造重投

发布确认和消费确认属于两段通信

RabbitMQ publisher confirm 告诉发布者消息在 Broker 一侧的处理结果；consumer ack 告诉 Broker 某次 delivery 已处理，可以从待确认集合移除。二者不能合并为一次业务确认。RabbitMQ Acknowledgements 与 Confirms

工程先创建已知队列，发布带 messageId 的消息，并调用 waitForConfirmsOrDie。消费者第一次取到消息后完成数据库事务，却故意关闭 channel，不发送 ack。新 channel 随后取得重投消息：

发布者             RabbitMQ               消费者 / PostgreSQL
publish EVT-1  --> 入队
confirm        <--
                   delivery EVT-1  ------> BEGIN
                                          INSERT inbox(EVT-1)
                                          UPDATE counter + 1
                                          COMMIT
                   channel 关闭   <------ 没有 basicAck
                   重新入队
                   delivery EVT-1  ------> Inbox 已存在，跳过重复业务更新
                   basicAck        <------ 数据库处理完成

发布确认之后，数据库业务可能尚未开始；数据库提交之后，ack 可能还没有到达 Broker。第二个窗口决定了消费者必须能够处理重复投递。

Inbox 与业务更新放在同一事务

InboxProcessor 先插入事件 ID：

INSERT INTO inbox(event_id)
VALUES (?)
ON CONFLICT DO NOTHING;

插入影响 1 行时才执行 counter 增加，然后提交二者；影响 0 行表示同一去重键已经处理，提交当前空操作并返回 false。测试在第一次 delivery 中从实际消息属性和 body 解出事件与计数器 ID，交给这个处理器。

事件记录与业务写入必须共同提交。先提交 Inbox 再执行业务，可能留下“已消费但没执行”的状态；先提交业务再单独记录 Inbox，重复消息又可能执行一次业务。工程的 rollbackDoesNotConsumeInboxKey 在提交前注入异常，检查 counter 仍为 10、Inbox 为 0，再重试同一事件并得到 11 与 1。

实际系统的去重键常需包含消费者职责或业务范围，并对同一键却携带不同内容的情况作出处理。Inbox 的保存期限也必须覆盖可能的重放窗口。删除过早，历史消息可能再次产生效果。

重投测试同时观察消息与数据库

工程的重投测试检查以下实际结果：第一次处理返回 true，数据库值为 11；关闭第一个 channel 后，新 delivery 的 redelivered 标志为 true，messageId 和 body 保持一致；第二次处理返回 false，数据库仍为 11，Inbox 只有 1 条。

最后通过新 channel 的 delivery tag 发送 basicAck，再检查该队列没有剩余待取消息。delivery tag 只在所属 channel 中有效，不能拿第一次 delivery 的 tag 到另一个 channel 确认。连接恢复、channel 和确认 API 说明见 RabbitMQ Java Client Guide。

测试关闭自动连接恢复，以便精确控制“关闭第一个 channel、创建第二个 channel”的动作。队列使用服务端生成名称，并限定当前连接的排他范围；测试结束删除队列，连接关闭提供额外清理。

为使获取动作有明确截止时间，示例用 basicGet 轮询并设置 5 秒 deadline，短暂间隔只调节查询频率。它适合小型可控实验；生产消费者通常使用推送式消费、合理 prefetch 和相应线程模型，不必复制这种轮询写法。

重复与乱序分别处理

Inbox 处理同一事件再次到达的问题；两个不同事件乱序到达时，仍需按照业务版本或状态迁移规则判断。工程用数据库条件更新验证 version 2 写入后，version 1 不能把状态覆盖回去。这个测试观察的是状态保护 SQL，消息的实际跨分区排序还要根据所用 Broker 和消费方式单独验证。

永久失败的消息需要明确拒绝、死信或人工处理路径，不能无限 requeue 形成紧循环。Outbox 则处理业务提交后“消息还没发出”的另一个窗口：应分别测试提交前失败、提交后 relay 未发送、发送成功未记录、记录成功等断点。

Kafka 的位点提交、生产者幂等和事务使用另一套机制，不能用一次 RabbitMQ 重投实验推导其行为。相关语义入口见 Kafka Design。JVM 内存重排与对象竞争适合使用专门的并发测试工具，见 OpenJDK jcstress。

收集并发失败并安全结束测试

Future、屏障与 SQL 各有超时

Worker 抛出的异常需要通过 Future.get 回到 JUnit 线程。只调用 executor.submit 而不读取 Future，后台 SQLException 可能被留在任务对象里，测试方法却正常结束。超时后应取消未完成任务，关闭 executor，并等待线程退出。

屏障超时表示某个参与者没有到达约定位置，原因可能是前置 SQL 失败、连接耗尽或错误锁顺序。锁等待超时表示数据库里的竞争尚未解除。Future 超时则是整个工作任务没有完成。先确定是哪一层超时，再看相应线程和数据库会话。

工程在 finally 中取消 Future 并调用 shutdownNow，随后有界等待退出；连接和语句由 try-with-resources 关闭。中断是协作式请求，不保证任意驱动或外部操作立即停止，因此数据库的 statement/lock timeout 仍有独立用途。

常见失败的下一步

现象	优先检查	恢复方式
屏障等待超时	两个 Future 的首个异常、读取是否被锁住	修正屏障位置，重新建立独立测试数据
两次增加变成 11	两次读取值、SQL 是常量赋值还是原子表达式	选择原子更新或版本冲突处理，再跑同一交错
乐观更新全为 0 行	id、初始版本、其他清理线程是否改动对象	独立准备 ID 与版本，保留冲突结果
唯一冲突后所有 SQL 都失败	当前事务是否已经处于 aborted 状态	rollback 后开启新事务
未看到消息重投	消息是否已自动 ack、队列是否随 channel 删除、自动恢复是否介入	检查 ack 模式、队列生命周期和关闭位置
重投导致再次增加	Inbox 与业务是否同事务、去重键是否稳定	修正原子性与键，再跑提交后关闭实验
同组测试互相删除数据	清理 SQL 是否缺少当前用例 ID	缩小清理条件或换独立 schema/container
容器连接失败	daemon 权限、实际映射端口、镜像启动日志	先修复容器路径，再执行业务断言

数据库竞争负例结束后，移除 -Dtest=LostUpdateAssumptionCase，重新执行最初的 clean verify。报告应恢复为 7 次正常执行；其中“观察错误算法最终为 11”与“原子更新最终为 12”是两个有不同目的的正常测试。

临时 PostgreSQL 和 RabbitMQ 由 Testcontainers 按测试类生命周期停止。异常遗留时先用 docker ps -a --filter label=lab=test16-state 列出本实验对象，再对确认过的 ID 清理，不影响其他项目容器。

控制交错之后再做重复与压力

受控双参与者实验给出一个明确交错下的结果。生产还有更高并发、更大数据量、连接池排队、不同锁集合与故障恢复。增加重复和压力测试前，先保存每次失败的输入与参与者历史，避免只得到“偶发失败 1 次”的统计。

共享线程池、固定端口、全局 Locale 和延迟清理也会制造与业务竞争无关的失败。整理这些隐藏输入的方法见 Fixture、隔离与 Flaky Test。

权威资料与规范地址

数据库与并发

• PostgreSQL 事务隔离：https://www.postgresql.org/docs/18/transaction-iso.html
• PostgreSQL 显式锁：https://www.postgresql.org/docs/18/explicit-locking.html
• PostgreSQL 错误码：https://www.postgresql.org/docs/18/errcodes-appendix.html
• OpenJDK jcstress：https://openjdk.org/projects/code-tools/jcstress/

消息与实验环境

• RabbitMQ 确认与重投：https://www.rabbitmq.com/docs/confirms
• RabbitMQ Java Client：https://www.rabbitmq.com/client-libraries/java-api-guide
• Kafka Design：https://kafka.apache.org/documentation/#design
• Docker Linux 权限：https://docs.docker.com/engine/install/linux-postinstall/
• 容器内运行 Testcontainers：https://java.testcontainers.org/supported_docker_environment/continuous_integration/dind_patterns/