Apache Camel重试路由事务提交失败问题排查求助

数据老陈

2026-6-30

问题根源分析

你的批量处理挂起卡在事务提交环节，核心原因是**split().streaming()的异步处理模式与全局事务边界的冲突**：

主路由的transacted()会创建一个覆盖整个批量流程的全局事务，但streaming()是异步分批处理消息，主事务会一直等待所有异步子任务完成；而子任务中涉及的数据库操作（比如子路由里的实体保存）会导致事务上下文传递异常，再加上FOR UPDATE SKIP LOCKED持有的数据库锁长时间未释放，最终造成事务提交阻塞。
单条处理时没有异步拆分，事务上下文清晰，锁和资源能及时释放，因此运行正常。

批量处理解决方案

方案1：拆分事务边界（推荐，适配重试场景）

将全局事务改为每个消息的独立事务，避免批量事务长时间持有锁，同时兼容异步处理：

from("quartz://database/message-retry-scheduler?cron=0/5 * * * * ?")
    .autostartup(true)
    .routeId("message-retry-route")
    .choice()
       .when(retryCheckPredicate)
        .log("Message retry scheduler is running and will check for retryable messages.")
        .to("jpa:" + Retryable.class.getCanonicalName()
            + "?nativeQuery=SELECT * FROM my_retryable WHERE is_retryable ORDER BY event_time ASC FOR UPDATE SKIP LOCKED"
            + "&consumeDelete=false"
            + "&maximumResults=10")
        .split(body()).streaming()
        // 每个子消息单独开启事务
        .transacted()
        .setHeader("reprocessedMessage", () -> true)
        .process(myProcessor)
        .log("Started reprocessing message with correlationId=${header.correlationId}")
        .to("direct:process-message")
    .endChoice()
.end();

from("direct:process-message")
    .routeId("message-processing-subroute")
    // 子路由内的数据库操作自动加入当前子事务
    // ... 原有处理逻辑

说明：每个消息处理完成后立即提交事务，释放FOR UPDATE SKIP LOCKED的锁，避免资源长时间占用；即使单条消息处理失败，也只会回滚该消息的事务，不影响其他消息，符合重试机制的容错需求。

方案2：保留批量事务原子性

如果需要批量处理的原子性（要么全部成功，要么全部回滚），调整split参数确保事务上下文正确传递：

from("quartz://database/message-retry-scheduler?cron=0/5 * * * * ?")
    .autostartup(true)
    .routeId("message-retry-route")
    .transacted()
    .choice()
       .when(retryCheckPredicate)
        .log("Message retry scheduler is running and will check for retryable messages.")
        .to("jpa:" + Retryable.class.getCanonicalName()
            + "?nativeQuery=SELECT * FROM my_retryable WHERE is_retryable ORDER BY event_time ASC FOR UPDATE SKIP LOCKED"
            + "&consumeDelete=false"
            + "&maximumResults=10")
        // 关闭异步streaming，开启事务上下文共享
        .split(body()).shareUnitOfWork(true).stopOnException()
        .setHeader("reprocessedMessage", () -> true)
        .process(myProcessor)
        .log("Started reprocessing message with correlationId=${header.correlationId}")
        .to("direct:process-message")
    .endChoice()
.end();

说明：

移除streaming()避免异步脱离主事务上下文，shareUnitOfWork(true)确保所有子处理共享主路由的事务，保证批量原子性。
stopOnException()确保单条消息失败时，整个批量事务回滚，适合对一致性要求高的场景。

方案3：优化JPA事务配置（适配必须保留streaming的场景）

如果一定要用streaming()，可以通过配置事务超时和JPA参数避免阻塞：

// 在JPA endpoint添加事务超时参数，防止无限等待
.to("jpa:" + Retryable.class.getCanonicalName()
    + "?nativeQuery=SELECT * FROM my_retryable WHERE is_retryable ORDER BY event_time ASC FOR UPDATE SKIP LOCKED"
    + "&consumeDelete=false"
    + "&maximumResults=10"
    + "&transactionTimeout=30") // 设置事务超时时间（单位：秒）

同时确保子路由message-processing-subroute中的数据库操作使用同一个EntityManager，避免额外的事务上下文冲突。

关键注意事项

多实例部署下，FOR UPDATE SKIP LOCKED的锁机制在两种方案中依然有效，不会出现重复消费。
方案1更适合重试场景：即使部分消息处理失败，未成功的消息会留在表中等待下一次调度重试，无需回滚整个批量。
建议监控数据库连接池，排查是否存在连接泄漏，这也是事务提交卡住的常见诱因。

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