问题

我在一个抽象实体上使用Apache Camel JPA消费者，使用 FOR UPDATE 和 @Consumed 注解的原生查询。消费端是从 application.yml 动态配置的。

消费者URI：

java

dbRoute.getComponentName() + ":" + dbRoute.getEntityClass()
    + "?resultClass=" + dbRoute.getEntityClass()
    + "&nativeQuery=" + dbRoute.getQuery()
    + "&consumeDelete=false"
    + "&skipLockedEntity=true"

application.yml：

yaml

db-routes:
  - componentName: jpa
    entityClass: com.example.FileTransaction
    query: "select * from FILE_TRANSACTION where 
            TRANSACTION_NAME='UPDATE_MARKET_PRICE' 
            and TRANSACTION_STATUS='INITIATE' for update"
    otherParameters: "&consumeDelete=false&skipLockedEntity=true"

实体（库类 — 无法修改）：

java

@Entity
public abstract class AbstractTransaction {

    @Consumed
    public void onConsumed() {
        this.setTransactionStatus("COMPLETED");
    }
}

@Entity
public class FileTransaction extends AbstractTransaction {
    // cannot add @Consumed here as parent handles it
}

仓储：

java

@Modifying()
@Transactional()
@Query("""
    UPDATE FileDetailsEntity fd
       SET fd.fileDetailStatus = :newStatus,
           fd.updateDate = CURRENT_TIMESTAMP
     WHERE fd.fileDetailsId = :id
       AND fd.fileDetailStatus = :expectedStatus
""")
int updateFileDetailsStatus(
    @Param("id") Long id,
    @Param("expectedStatus") FileDetailsStatus expectedStatus,
    @Param("newStatus") FileDetailsStatus newStatus
);

需求

我需要在路由的中段将实体状态更新为 PROCESSING，在最后触发 @Consumed 之前——让其他系统可以看到该行正在被处理。

Expected flow:
INITIATE
   │
   ▼
SELECT FOR UPDATE 🔒
   │
   ▼
UPDATE status = PROCESSING  ← need this visible mid-route
   │
   ▼
Business logic runs
   │
   ▼
@Consumed → COMPLETED
   │
   ▼
COMMIT + UNLOCK

我尝试了

1.使用Spring Repository更新：

java

// Still blocked — FOR UPDATE lock held by Camel TX
int updated = repository.updateFileDetailsStatus(
    entity.getId(),
    FileDetailsStatus.INITIATE,
    FileDetailsStatus.PROCESSING
);

2.切换到 namedQuery + PESSIMISTIC_WRITE：

yaml

otherParameters: "&consumeDelete=false
                  &skipLockedEntity=true
                  &consumeLockEntity=true
                  &lockModeType=PESSIMISTIC_WRITE"

仍然处于阻塞状态——PESSIMISTIC_WRITE 也在数据库层生成 SELECT FOR UPDATE。

3.使用 consumeLockEntity=false： 移除了锁，但在两个并行消费者和 maxMessagesPerPoll=5 的情况下，两个消费者获取相同的行，导致重复处理尝试和浪费的负载。

环境

- Apache Camel 4.x
- Spring Boot 4.x
- JPA / Hibernate
- Oracle DB
- 2 parallel consumers
- maxMessagesPerPoll=5

问题

1. 在Camel JPA消费者持有 FOR UPDATE 锁时，是否有在路由中段更新实体状态的方法？
2. 是否可以在Camel JPA消费者中将 PESSIMISTIC_WRITE 与 nativeQuery 一起使用？
3. 在Camel JPA消费者与并行消费者的场景下，中间状态可见性的正确模式是什么？(INITIATE → PROCESSING → COMPLETED)

解决方案

在你执行SELECT FOR UPDATE的中间阶段，被选中的行会被锁定，直到你提交或回滚事务，其他人都看不到该行的更新。

为了在保持对行的锁定的同时解决你的问题，我将尝试把流程分成两部分。

第一子流程仅使用SELECT FOR UPDATES对行进行轮询，然后将业务处理委派给另一个子流程，接着通过带有 @Consumed注解的方法将行标记为PROCESSING。

第二子流程从第一子流程获取记录，按需处理，然后将记录状态标记为PROCESSED。

from(jpa("com.kireygroup.camel.jpa.model.CamelJpaLock")
  .skipLockedEntity(true)
  .consumeDelete(false)
  .query("select c from CamelJpaLock c where c.fInv = 'N'")
  .consumeLockEntity(true)
  .maxMessagesPerPoll(1)
  .lockModeType(LockModeType.PESSIMISTIC_WRITE))
  .to(seda("processor")); //mark as processing after this with @Consumed

from(seda("processor"))
  .log("simulating work")
  .delay(10000).syncDelayed()
  .process(exchange -> {
    CamelJpaLock entity = exchange.getIn().getBody(CamelJpaLock.class);
    entity.setfInv("S"); //mark as processed
  })
  .to(jpa("com.kireygroup.camel.jpa.model.CamelJpaLock").useExecuteUpdate(true))
  .log("${body} entity processed");