如何实现Camel路由仅单并发执行并直接丢弃额外请求？

需求目标 #

• 通过PUT请求触发一个长耗时的Camel路由
• 路由同一时间仅允许运行1个实例
• 路由处理期间到达的其他请求直接丢弃，不进入队列等待

问题描述 #

我尝试用AtomicBoolean标记路由是否正在处理，实现了需求，但感觉这种方式略显取巧。想确认该实现是否合理，或者Camel是否提供了更原生的最优方案。

最终实现代码（带AtomicBoolean） #

@Component
public class TestRoute2 extends RouteBuilder {

  @Autowired
  private MyService myService;

  @Override
  public void configure() {

    AtomicBoolean isProcessing = new AtomicBoolean(false);

    from("platform-http:/api/myservice?httpMethodRestrict=PUT")
      .routeId("myServiceRoute")
      .doTry()
        .process(exchange -> {
        // Cast the endpoint to SedaEndpoint to access the queue
        SedaEndpoint sedaEndpoint = (SedaEndpoint)
        getContext().getEndpoint("seda:handlePut");
        int currentQueueSize = sedaEndpoint
          .getQueue()
          .size();

        if (currentQueueSize > 0 || isProcessing.get()) {
          throw new org.apache.camel.CamelExchangeException("Service busy", exchange);
        }

        // Mark service as busy
        isProcessing.set(true);
      })
        .wireTap("seda:handlePut")
        .setHeader("CamelHttpResponseCode", constant(204))
      .endDoTry()
      .doCatch(org.apache.camel.CamelExchangeException.class)
      .setHeader("CamelHttpResponseCode", constant(503))
      .setBody(constant("Service busy, please try again later"))
      .end();

    from("seda:handlePut?concurrentConsumers=1")
      .doTry()
        .bean(myService, "slowMethod") // runs for long time
      .endDoTry()
      .doFinally()
      .process(exchange -> isProcessing.set(false));
  }
}

此前有问题的实现（无AtomicBoolean） #

这段代码会导致路由处理时，额外排队1个请求，不符合需求：

@Override
  public void configure() {

    from("platform-http:/api/myservice?httpMethodRestrict=PUT")
      .routeId("myServiceRoute")
      .doTry()
        .process(exchange -> {
        SedaEndpoint sedaEndpoint = (SedaEndpoint)
        getContext().getEndpoint("seda:handlePut");
        int currentQueueSize = sedaEndpoint
          .getQueue()
          .size();
        if (currentQueueSize > 0) {
          throw new org.apache.camel.CamelExchangeException("Service busy", exchange);
        }
      })
        .wireTap("seda:handlePut")
        .setHeader("CamelHttpResponseCode", constant(204))
      .endDoTry()
      .doCatch(org.apache.camel.CamelExchangeException.class)
        .setHeader("CamelHttpResponseCode", constant(503))
        .setBody(constant("Service busy, please try again later"))
      .end();

    from("seda:handlePut?concurrentConsumers=1")
      .bean(myService, "slowMethod");
  }

方案分析与优化建议 #

1. AtomicBoolean实现的优缺点 #

• 可行但不够优雅：该实现逻辑直观，能满足单节点部署的需求，但存在以下潜在问题：
  • 状态存储在RouteBuilder的AtomicBoolean中，若路由重新部署或Camel上下文重启，状态可能出现不一致；
  • 多实例集群部署时，无法跨节点同步状态，会导致每个节点各自允许一个实例运行；
  • 需要手动维护状态重置，虽然用了doFinally保障，但仍属于侵入式的状态管理。

2. Camel原生最优方案：利用Seda端点配置 #

Camel的Seda组件本身支持通过参数配置实现“无缓冲、单消费、满队列直接拒绝”的特性，完全符合需求，无需手动维护状态：

• queueSize=0：设置队列大小为0，不允许缓冲请求；
• concurrentConsumers=1：仅启动1个消费者线程，保证同一时间只有1个路由实例运行；
• blockWhenFull=false：队列满（无空闲线程）时直接拒绝请求，而非阻塞等待。

优化后的代码

@Component
public class TestRoute2 extends RouteBuilder {

  @Autowired
  private MyService myService;

  @Override
  public void configure() {
    // 主路由：处理PUT请求，尝试触发后台路由
    from("platform-http:/api/myservice?httpMethodRestrict=PUT")
      .routeId("myServiceRoute")
      .doTry()
        // 直接wireTap到配置好的seda端点
        .wireTap("seda:handlePut?concurrentConsumers=1&queueSize=0&blockWhenFull=false")
        .setHeader("CamelHttpResponseCode", constant(204))
      .doCatch(java.util.concurrent.RejectedExecutionException.class)
        // 捕获拒绝异常，返回503
        .setHeader("CamelHttpResponseCode", constant(503))
        .setBody(constant("Service busy, please try again later"))
      .end();

    // 后台长耗时路由
    from("seda:handlePut")
      .bean(myService, "slowMethod");
  }
}

方案对比 #

• 原生方案完全基于Camel组件特性，代码更简洁，无需手动维护状态；
• 避免了AtomicBoolean实现的状态一致性问题；
• 异常处理更符合Camel的设计思路，直接利用组件抛出的标准异常。

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