DynamoDB事务冲突检测：并发更新场景下的事务行为咨询

我运行的应用需要向DynamoDB多表写入数据，为保证操作全成功或全失败，采用事务机制。

现有3张DynamoDB表：

• EntityA：userId为分区键，requestId为排序键，包含count计数字段
• EntityB和EntityC：userId为分区键，requestId+eventId为排序键

假设每次执行接收的事件eventId唯一，EntityB、EntityC始终为新记录，EntityA则可能被更新。该流程可能存在同一用户同一requestId的并发操作，虽不频繁，但需确保仅一个流程失败以防止幽灵更新。

以下是简化代码：

TransactionWriteRequest transactionWriteRequest = new TransactionWriteRequest();
Optional<EntityA> optionalEntityA = dynamoDbMapper.query(entityA.class, queryExpression).stream().findFirst();
transactionWriteRequest.addPut(createNewEntityB());
transactionWriteRequest.addPut(createNewEntityC());
if (optionalEntityA.isPresent()){
    transactionWriteRequest.addPut(optionalEntityA.get().incrementCount());
} else {
    transactionWriteRequest.addPut(createNewEntityA());
}
dynamoDbMapper.transactionWrite(transactionWriteRequest);

据了解，DynamoDbMapper API无法显式启动事务，也无法将读写操作归为同一事务，事务在执行transactionWrite时启动并结束。

疑问1 #

若EntityA中某(consumerId, requestId)组合的记录不存在，两个并发执行同时创建时，因主键唯一性，其中一个会失败，对吗？

疑问2 #

更新场景下的行为：

1. EntityA已存在(consumerId=1, requestId=1)的记录
2. 收到两个含相同主键(consumerId=1, requestId=1)的并发事件
3. 执行1读取EntityA（代码第2行）
4. 随后执行2成功提交事务

请问执行1提交事务时，DynamoDB会检测到EntityA已被执行2修改并抛出异常，还是会覆盖执行2写入的值？已知DynamoDB支持乐观锁以仅在记录未被修改时更新，但事务操作不支持该特性，故想确认DynamoDB事务是否能检测此类冲突。

回答 #

疑问1解答 #

是的，其中一个并发执行会失败。当两个事务同时尝试创建同一主键的EntityA记录时，DynamoDB会在事务提交阶段检查主键唯一性，第二个提交的事务会触发TransactionCanceledException，整个事务回滚，不会产生重复的EntityA记录。

疑问2解答 #

默认情况下，执行1提交事务时会覆盖执行2写入的EntityA值。原因如下：

1. 代码中读取EntityA的操作（第2行）是在事务启动前完成的，不属于事务的一部分，DynamoDB事务无法感知这次读取与后续写入之间的外部修改。
2. 当前的Put操作没有添加任何条件约束，DynamoDB事务仅保证自身操作的原子性，不会自动检测这种“读取-修改-写入”的并发冲突。

解决方案：手动添加条件表达式实现乐观锁 #

如果要避免覆盖，需要在Put EntityA的操作中添加条件表达式，基于读取到的状态做判断。比如利用count字段的当前值作为条件：

if (optionalEntityA.isPresent()){
    EntityA updatedEntityA = optionalEntityA.get().incrementCount();
    // 创建Put请求并添加条件表达式
    Put putRequest = dynamoDbMapper.createPutRequest(updatedEntityA);
    // 确保count的值和读取时一致，防止并发修改
    putRequest.setConditionExpression("count = :originalCount");
    putRequest.addExpressionAttributeValuesEntry(
        ":originalCount", 
        new AttributeValue().withN(String.valueOf(optionalEntityA.get().getCount()))
    );
    transactionWriteRequest.addPut(putRequest);
} else {
    transactionWriteRequest.addPut(createNewEntityA());
}

这样，当执行1提交事务时，如果EntityA的count已经被执行2修改，条件表达式不满足，事务会抛出TransactionCanceledException并回滚，避免覆盖已更新的值。

内容的提问来源于stack exchange，提问作者user2957378