使用 JpaRepository 进行并发保存

Question

我正在开发一个微服务，它为 REST 端点提供服务，使用 spring 数据在数据库中保存/检索数据。

让我们调用实体类Foo，它的ID 字段有一个简单的Long，还有一些其他数据字段。每个Foo 的 ID 不会在此服务中自动生成，它们是从知道使它们唯一的外部来源提供的。

该服务有一个 POST 端点，它同时服务于 CRUD 模型的创建和更新功能，它在代码的服务层调用相应的函数，我们称这个函数为AddData(Foo foo_incoming)。 POST 消息的正文包含要保存到数据库的数据和要保存到的 Foo 的 ID。 AddData 的逻辑是这样的：

@Service("fooService")
public class FooServiceImpl {

    @Autowired
    FooRepository fooRepository; // Subinterface of JpaRepository

    @Transactional
    public Long AddData(Foo foo_incoming) {
        Optional<Foo> foo_check = fooRepository.findById(incoming.getId());
        Foo foo_exists;

        // Exists already?
        if (foo_check.isEmpty()) {
            // New Foo
            foo_exists = fooRepository.saveAndFlush(foo_incoming);
        } else {
            // Update existing foo
            foo_exists = foo_check.get();
            foo_exists.addToFieldA(foo_incoming.getFieldA());
            foo_exists.addToFieldB(foo_incoming.getFieldB());
        }

        return foo_exists.getId();
    }

}

这个函数负责为Foo 创建初始记录并更新记录。

当 POST 请求进来向一些 ID=1 的 Foo 添加数据时，我们称它为 foo-1，它还不存在，如果它们之间有合理的时间，第一个request 将为foo-1 创建初始记录，所有后续调用只会更新。 IE。 saveAndFlush 有足够的时间实际刷新到数据库，因此对 findById 的后续调用会在数据库中找到 foo-1，然后跳转到 else 块并仅更新其字段。

我遇到的问题是，当相同Foo（相同ID）的N POST 足够快地发送到服务时，似乎所有对AddData 的相应调用同时发生。因此，如果foo-1 尚不存在，则在对AddData 的每个调用中，findById(1) 将返回空。因此，saveAndFlush 在 ID=1 的 Foos 中被调用 N 次，从而引发了 DataIntegrityViolationException。

我已经在网上搜索了几天试图解决这个问题。

• 该方法已注释@Transactional。我试过在方法和整个类上使用@Transactional(isolation = Isolation.SERIALIZABLE)，但没有帮助。
• 我试过用@Lock(LockModeType.PESSIMISTIC_READ) 和@Lock(LockModeType.PESSIMISTIC_WRITE) 分别注释FooRepository 中的findById 和saveAndFlush 方法，但没有成功。
• 我尝试将@Version 字段添加到Foo，没有任何变化。

我不知道如何强制AddData 连续发生，我认为这是@Transactional(isolation = Isolation.SERIALIZABLE) 应该做的。

我正在考虑赋予“创建”和“更新”它们自己的功能 - 为创建创建一个 PUT 端点。但是 PUT 端点也会有类似的问题——如果我想尝试防止代码中的主键冲突，我必须在执行saveAndFlush 之前对findById 进行类似的检查。但是这个服务的实际使用方式，PUT端点可能不是一个选项。

将saveAndFlush 包装在try/catch 块中确实 捕获异常，令我惊讶。当saveAndFlush 失败时，我可以尝试一些时髦的逻辑来尝试再次调用findById，但如果有办法避免抛出异常，我更愿意这样做。

任何建议将不胜感激！

编辑：更多可能有用的上下文。此微服务在 Kubernetes 集群中运行，其中可能有许多此服务实例同时服务请求。我仍在研究处理多个实例的并发性，并弄清楚这不是我必须自己做的事情 - 我的团队正在开发几个这样的微服务，我们可能会开发一个通用库来解决所有这些问题.

编辑 2：我现在忘记了，我在运行服务时使用的是 H2 数据库，而不是真正的数据库。会不会跟这个有关系？

我要重申，这里发生的情况是多次调用检查数据库中是否存在 foo-1，在 foo-1 存在之前；因此，我认为数据库锁定在这里不会对我有帮助，因为没有实体 to 锁定。我认为强制AddData 连续发生可以解决这个问题，我完全不知道为什么将@Transactional(isolation = Isolation.SERIALIZABLE) 添加到AddData 对我没有这样做。

Answer 1

有一些方法可以以有益的方式与 Jpa 一起利用并发性，但总而言之，不可能同时进行 Jpa 调用。

请记住，Jpa 依赖于非线程安全的类对象，例如 EntityManager, Session, Connection, etc.。它们是这样设计的，目的是避免竞争条件、死锁以及多线程可能引起的所有问题。话虽如此，Jpa 需要阻塞数据库调用。

尽管如此，您绝对可以与 JPA 方法同时实现您的业务逻辑以提高性能，您似乎已经知道了。在许多情况下，与数据创建、验证等所需的时间相比，完成 Jpa 操作所需的时间非常短。尽管如此，还是需要做出一些妥协，因为最终需要多线程上下文中的每个线程在 Jpa 事件循环上进行阻塞调用。据我所知，Isolation.SERIALIZABLE 似乎已经是实现您所追求的最重要的一步。

您可能想研究一下 R2dbc，它是 JDBC 的一种反应式实现，可以帮助您完成您在此处尝试做的事情。它已经开发了很长时间，并且接近发布状态。上次我听说应该在 10 月份完成，我的团队已经在单独的分支中开始了转换。

Answer 2

我不确定您如何使用@Transactional 来实现，但是有一种替代方法可以使用同步块来解决您的问题。既然您说过，您的密钥将是唯一的，您将根据该密钥找出 if 对象已存在，我建议将其用作同步插入/更新块的密钥。

当您的键匹配时，我正在使用字符串实习生返回相同的对象。

@Service("fooService")
public class FooServiceImpl {

    @Autowired
    FooRepository fooRepository; // Subinterface of JpaRepository

    @Transactional
    public Long AddData(Foo foo_incoming) {
        Optional<Foo> foo_check = fooRepository.findById(incoming.getId());
        String key = **incoming.getId().intern();**
        Foo foo_exists;
        **synchronized (key)** {
            // Exists already?
            if (foo_check.isEmpty()) {
                // New Foo
                foo_exists = fooRepository.saveAndFlush(foo_incoming);
            } else {
                // Update existing foo
                foo_exists = foo_check.get();
                foo_exists.addToFieldA(foo_incoming.getFieldA());
                foo_exists.addToFieldB(foo_incoming.getFieldB());
            }
        }

        return foo_exists.getId();
    }

}

这篇文章可能会对您有所帮助。 Synchronizing on String objects in Java