Camel拆分迭代中如何跨迭代共享并更新Exchange属性？

你的问题核心是Camel的split组件默认会为每个拆分元素创建独立的子Exchange，子Exchange的属性是父Exchange的副本，修改子Exchange的属性不会同步回父Exchange，导致printed_A状态无法在迭代间共享，无法实现类型A仅打印一次的需求。

下面提供两种可行的解决方案：

方案1：通过父Exchange共享状态 #

所有拆分出的子Exchange都隶属于同一个父Exchange的处理流程，直接读写父Exchange的属性即可实现状态共享。

修改MyMapper代码 #

public void preparePrintRequest(Exchange exchange) {
    // 获取父Exchange，所有拆分迭代共享该实例的属性
    Exchange parentExchange = exchange.getUnitOfWork().getParentExchange();
    if (parentExchange == null) {
        parentExchange = exchange;
    }

    Integer currentId = exchange.getProperty("currentId", Integer.class);

    List<String> printType;
    if (currentId.equals(2)) {
        // 注意：Arrays.asList返回不可变列表，需转为ArrayList支持remove操作
        printType = new ArrayList<>(Arrays.asList("D"));
    } else {
        printType = new ArrayList<>(Arrays.asList("A", "B", "C"));
    }

    Boolean printedA = parentExchange.getProperty("printed_A", Boolean.class);
    if (printedA != null && printedA) {
        printType.remove("A");
    }

    if (printType.contains("A")) {
        parentExchange.setProperty("printed_A", true);
    }

    // 将处理后的打印类型存入当前Exchange，供后续doPrint使用
    exchange.setProperty("printType", printType);
}

原路由代码无需修改，因为初始的printed_A属性是设置在父Exchange上的，所有子Exchange都能通过父Exchange访问到更新后的状态。

方案2：使用ThreadLocal存储状态 #

如果你的split是单线程执行（默认就是单线程，未开启parallelProcessing=true），可以用ThreadLocal存储状态，同一线程内的所有迭代会共享这个状态。

修改MyMapper类 #

public class MyMapper {
    private static final ThreadLocal<Boolean> printedAFlag = new ThreadLocal<>();

    public void preparePrintRequest(Exchange exchange) {
        // 首次迭代时从父Exchange初始化状态
        if (printedAFlag.get() == null) {
            Boolean initialFlag = exchange.getProperty("printed_A", Boolean.class);
            printedAFlag.set(initialFlag != null ? initialFlag : false);
        }

        Integer currentId = exchange.getProperty("currentId", Integer.class);

        List<String> printType;
        if (currentId.equals(2)) {
            printType = new ArrayList<>(Arrays.asList("D"));
        } else {
            printType = new ArrayList<>(Arrays.asList("A", "B", "C"));
        }

        if (printedAFlag.get()) {
            printType.remove("A");
        }

        if (printType.contains("A")) {
            printedAFlag.set(true);
        }

        exchange.setProperty("printType", printType);
    }

    // 路由结束后清理ThreadLocal，避免内存泄漏
    public void cleanupThreadLocal(Exchange exchange) {
        printedAFlag.remove();
    }
}

修改路由代码 #

在split结束后调用清理方法：

from("direct:route1")
    .process(e -> e.setProperty("printed_A", false))
    .split(exchangeProperty("items"))
        .stopOnException()
        .setProperty("currentId", body().method("getId"))
        .bean(MyMapper.class, "preparePrintRequest")
        .to("direct:doPrint")
    .end()
    .bean(MyMapper.class, "cleanupThreadLocal");

关键注意事项 #

• 若split开启了parallelProcessing=true（多线程拆分），ThreadLocal方案会失效，此时只能使用方案1。
• 原代码中Arrays.asList返回不可变列表，直接调用remove会抛出UnsupportedOperationException，必须转为ArrayList。

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