微服务实例之间的 JPA 同步答案

【问题标题】：JPA synchronization between micro-service instances微服务实例之间的 JPA 同步
【发布时间】：2023-03-22 03:06:02
【问题描述】：

我有一个基于计时器的工作

@Component
public class Worker {

    @Scheduled(fixedDelay = 100)
    public void processEnvironmentActions() {
        Job job = pickJob();
    }

    public Job pickJob() {
        Job job = jobRepository.findFirstByStatus(Status.NOT_PROCESSED);
        job.setStatus(Status.PROCESSING);
        jobRepository.save(job);
        return job;
    }
}

现在，在大多数情况下，这应该会给我正确的结果。但是如果有两个微服务实例同时执行这段代码会怎样呢？

我如何确保即使有多个服务实例，存储库也应始终将一项作业只分配给一个实例而不是其他实例。

编辑：我认为人们对@Transactional 感到困惑/集中注意力所以将其删除。问题还是一样的。

【问题讨论】：

你知道@Transactional 在this 上调用方法的情况下不起作用吗？
如果方法是public应该可以吧？例如，我将其编写为private 方法，但如果我将其编写为public，如何实现多个实例之间的同步？
我不知道如何回答你的问题。我只是注意到您的代码中存在问题。
我认为您的问题是：如何跨多个服务实例实现分布式工作队列？
@talex 是对的。 Transactional 不适用于本地调用，因为不涉及代理。要解决您的问题，您需要查看整个工作表。我稍后会添加答案。

标签： java hibernate spring-boot jpa spring-data-jpa

【解决方案1】：

@Jens Schauder 的回答为我指明了正确的方向。让我分享代码，以便它指导其他人。这就是我解决问题的方法，我改变了我的工作类别如下

@Entity 
public class Job {
   @Version
   private Long version = null; 
   // other fields omitted for bervity
}

现在，让我们跟踪以下代码

@Transactional
public Job pickJob() {
    Job job = jobRepository.findFirstByStatus(Status.NOT_PROCESSED);
    job.setStatus(Status.PROCESSING);
    Job saved jobRepository.save(job);
    return saved;
}

注意：确保返回saved 对象而不是job 对象。如果您返回作业对象，它将失败第二次save 操作作为 job 的版本计数将落后于 saved.

。

Service 1                                 Service 2 

1. Read Object (version = 1)              1. Read Object (version = 1)
2. Change the object and save 
      (changes the version)
3. Continues to process                   2. Change the object and save 
                                               (this operation fails as 
                                                the version that was read 
                                                was 1 but in the DB version is 2)
                                          3. Skip the job processing

这样，作业将仅由一个进程接手。

【讨论】：

【解决方案2】：

但是如果有两个微服务实例同时执行这段代码会发生什么？

答案通常是：视情况而定。

所有这些都假设您的代码在事务中运行

乐观锁定。

如果您的Job 实体具有版本属性，即带有@Version 注释的属性。启用乐观锁定。如果进程访问相同的作业，则会注意到版本属性在尝试保持更改的作业实体时发生更改，并以OptimisticLockingException 失败。您所要做的就是处理该异常，这样您的进程就不会死掉，而是再次尝试获取下一个Job。
无（JPA 级别）锁定。

如果Job 实体没有版本属性，JPA 将默认不应用任何锁定。访问Job 的第二个进程将发出更新，这本质上是一个 NOOP，因为第一个进程已经更新了它。两者都不会注意到问题。您可能希望避免这种情况。
悲观锁定

一个 pessimistic_write 锁会阻止任何人在你完成读写之前读取实体（至少这是我对 JPA 规范的理解）。因此，这应该避免第二个进程能够在第一个进程完成写入之前选择行。这可能会阻止整个第二个过程。所以要确保持有这样一个锁的事务是短的。

为了获得这样的锁，使用@Lock(LockModeType.PESSIMISTIC_WRITE)注释存储库方法findFirstByStatus。

当然，可能有一些库或框架可以为您处理这些细节。

【讨论】：

【解决方案3】：

我同意@Conffusion 的回答，但在不了解刷新策略的情况下，您应该使用jobRepository.saveAndFlush(job) 方法，以便确保将 sql 语句推送到数据库。

另见Difference between save and saveAndFlush in Spring data jpa

【讨论】：

【解决方案4】：

我对 spring-batch 不熟悉，但显然 spring-batch 实现了乐观锁定，因此如果另一个线程已经选择了相同的作业，则保存操作将失败。

见spring batch horizontal scaling

【讨论】：