Apache Camel - 行已被另一个事务更新或删除

Question

我有一批使用骆驼来消费数据库中的一些记录。所以端点是一个带有namedQuery的jpa。

当我启动批处理时，运行结束时出现异常。 在运行期间，我的代码会更新端点正在使用的实体。我的问题是如何避免这个异常？

这是一个例外：

[03.06.2015 10:47:32,277] WARN  org.apache.camel.util.CamelLogger.log:224 :Consumer Consumer[jpa://ch.gma.nova.vaudoise.entity.ImpactEntity?consumer.namedQuery=mutationsContrat&consumer.parameters=%23params&maximumResults=500] failed polling endpoint: Endpoint[jpa://ch.gma.nova.vaudoise.entity.ImpactEntity?consumer.namedQuery=mutationsContrat&consumer.parameters=%23params&maximumResults=500]. Will try again at next poll. Caused by: [javax.persistence.OptimisticLockException - Row was updated or deleted by another transaction (or unsaved-value mapping was incorrect) : [ch.gma.nova.vaudoise.entity.ImpactEntity#101]]
javax.persistence.OptimisticLockException: Row was updated or deleted by another transaction (or unsaved-value mapping was incorrect) : [ch.gma.nova.vaudoise.entity.ImpactEntity#101]
    at org.hibernate.jpa.spi.AbstractEntityManagerImpl.wrapStaleStateException(AbstractEntityManagerImpl.java:1788) ~[hibernate-entitymanager-4.3.6.Final.jar:4.3.6.Final]
    at org.hibernate.jpa.spi.AbstractEntityManagerImpl.convert(AbstractEntityManagerImpl.java:1705) ~[hibernate-entitymanager-4.3.6.Final.jar:4.3.6.Final]
    at org.hibernate.jpa.spi.AbstractEntityManagerImpl.convert(AbstractEntityManagerImpl.java:1677) ~[hibernate-entitymanager-4.3.6.Final.jar:4.3.6.Final]
    at org.hibernate.jpa.internal.QueryImpl.getResultList(QueryImpl.java:458) ~[hibernate-entitymanager-4.3.6.Final.jar:4.3.6.Final]
    at org.apache.camel.component.jpa.JpaConsumer$1.doInTransaction(JpaConsumer.java:90) ~[camel-jpa-2.12.3.jar:2.12.3]
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133) ~[spring-tx-4.1.2.RELEASE.jar:4.1.2.RELEASE]
    at org.apache.camel.component.jpa.JpaConsumer.poll(JpaConsumer.java:80) ~[camel-jpa-2.12.3.jar:2.12.3]
    at org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:187) [camel-core-2.12.3.jar:2.12.3]
    at org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:114) [camel-core-2.12.3.jar:2.12.3]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:482) [na:1.7.0]
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:315) [na:1.7.0]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:193) [na:1.7.0]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:308) [na:1.7.0]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1176) [na:1.7.0]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641) [na:1.7.0]
    at java.lang.Thread.run(Thread.java:795) [na:1.7.0]

还有实体：

@Entity
@Table(name = "IMPACT_FLUX")
@NamedQueries({
    @NamedQuery(name = ImpactEntity.NAMED_QUERY_MUTATION, query = "Select i from ImpactEntity i where i.process = 'MUTATION_CONTRAT' and i.status = :status and i.application = :application"),
    @NamedQuery(name = ImpactEntity.NAMED_QUERY_ANNONCE_SINISTRE, query = "Select i from ImpactEntity i where i.process = 'ANNONCE_SINISTRE' and i.status = :status and i.application = :application"), })
public class ImpactEntity {

  public static final String NAMED_QUERY_MUTATION = "mutationsContrat";

  public static final String NAMED_QUERY_ANNONCE_SINISTRE = "annonceSinistre";

  @Id
  @GeneratedValue(strategy = GenerationType.AUTO, generator = "seqImpactFlux")
  @SequenceGenerator(name = "seqImpactFlux", sequenceName = "SEQ_IMPACT_FLUX")
  @Column(name = "ID_IMPACT")
  private Long id;

  @Column(name = "ID_CONTRAT")
  private Long numContrat;

  @Column(name = "ID_ANCIEN_CONTRAT")
  private Long numAncienContrat;

  @Column(name = "ID_PARTENAIRE")
  private Long numPartenaire;

  @Column(name = "ID_PRESTATION")
  private Long numPrestation;

  @Column(name = "STATUS")
  private String status;

  @Column(name = "PROCESSUS")
  private String process;

  @Column(name = "APPLICATION")
  private String application;

  @Column(name = "CODE")
  private String code;

  @Version
  @Column(name = "OPTLOCK")
  private Integer version;

  @ManyToOne
  @JoinColumn(name = "ID_SUIVI")
  private SuiviMessageEntity suiviMessage;

  @Column(name = "DT_CREATION")
  private Date dateCreation;
...

Answer 1

可能是您的批处理作业实例相互交错，因此它们读取和更新相同的数据。

也许您正在为 JPA 组件使用固定速率（默认），因此另一个作业实例在前一个作业实例完成之前启动。

要解决这个问题，你可能应该使用fixed delay：

consumer.useFixedDelay=true

Answer 2

您遇到此问题是因为两个并发线程更新同一个实体，而optimistic locking mechanism 阻止了losing updates。

有几种方法可以解决此问题：

• 如果两个事务更新 ImpactEntity 的不同部分，您有两个选择：

  1. 您可以使用version-less optimistic locking

  2. 你使用DML UPDATE绕过版本检查

• 如果两个事务更新ImpactEntity的同一个字段，你需要在两个事务中都使用pessimistic locking，所以一旦一个事务获得行级锁，另一个事务将等待锁被释放

  ImpactEntity impactEntity = ...
  session.buildLockRequest(new LockOptions(LockMode.PESSIMISTIC_WRITE))
     .lock(impactEntity);