如何在Apache Camel中正确合并列表项并实现字段映射

初始消息体如下：

<itemsList>
  <item>
    <id>1</id>
    <name></name>
    <description></description>
  </item>
  <item>
    <id>2</id>
    <name></name>
    <description></description>
  </item>
</itemsList>

我通过以下Camel路由遍历itemsList：

<setProperty propertyName="itemsListSize">
  <xpath saxon="true" resultType="java.lang.Integer">count(//*[local-name()='itemsList'])</xpath>
</setProperty>
<loop>
  <simple>${exchangeProperty.itemsListSize}</simple>
  <setProperty propertyName="currentId">
    <xpath saxon="true" resultType="java.lang.String">
       //*[local-name()='itemsList'][function:simple('${exchangeProperty.CamelLoopIndex}')+1]/*[local-name()='id']/text()
    </xpath>
  </setProperty>
  <enrich strategyRef="someAggregationStrategy">
    <constant>direct:getItemInfoById</constant>
  </enrich>
</loop>

其中getItemInfoById路由会根据currentId调用服务，返回格式如下：

<customItem>
  <customId>1</customId>
  <customName>John</customName>
  <customDescription>Human</customDescription>
</customItem>

需求是每次循环结束后，将返回的customName映射到当前循环的item的name字段，customDescription映射到description字段，以此丰富初始消息。请问能否通过enrich节点的AggregationStrategy实现？如果可以，具体代码该如何编写？

当前我的AggregationStrategy参考代码：

public Exchange aggregate(Exchange original, Exchange resource) {

        for (Entry<String, Object> e : resource.getProperties().entrySet()) {
            String k = e.getKey();
            Object v = e.getValue();
            original.setProperty(k, v);
        }

        original.getOut().setBody(original.getIn().getBody());
        return original;
}

完全可以通过AggregationStrategy实现这个需求，核心思路是在聚合方法中定位原始消息的当前循环节点，解析返回的补充数据后更新对应字段。

完整实现代码 #

import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;

import javax.xml.parsers.DocumentBuilderFactory;
import java.util.Map;

public class ItemEnrichAggregationStrategy implements AggregationStrategy {

    @Override
    public Exchange aggregate(Exchange original, Exchange resource) {
        // 获取Camel内置的循环索引（从0开始）
        Integer loopIndex = original.getProperty(Exchange.LOOP_INDEX, Integer.class);
        if (loopIndex == null) {
            return original;
        }

        try {
            // 解析原始消息的XML文档
            Document originalDoc = DocumentBuilderFactory.newInstance()
                    .newDocumentBuilder()
                    .parse(original.getIn().getBody(java.io.InputStream.class));
            originalDoc.getDocumentElement().normalize();

            // 定位当前循环对应的item节点
            NodeList itemNodes = originalDoc.getElementsByTagName("item");
            if (loopIndex >= itemNodes.getLength()) {
                return original;
            }
            Element currentItem = (Element) itemNodes.item(loopIndex);

            // 解析补充接口返回的customItem数据
            Document resourceDoc = DocumentBuilderFactory.newInstance()
                    .newDocumentBuilder()
                    .parse(resource.getIn().getBody(java.io.InputStream.class));
            resourceDoc.getDocumentElement().normalize();

            String customName = getTagText(resourceDoc, "customName");
            String customDescription = getTagText(resourceDoc, "customDescription");

            // 更新当前item的name和description字段
            updateTagText(currentItem, "name", customName);
            updateTagText(currentItem, "description", customDescription);

            // 将更新后的XML写回原始消息体
            original.getOut().setBody(originalDoc);
            // 保留原始Exchange的所有属性
            copyProperties(original.getProperties(), original.getOut());
            // 同步补充接口返回的属性
            copyProperties(resource.getProperties(), original.getOut());

        } catch (Exception e) {
            // 可根据业务需求替换为日志记录或自定义异常处理
            e.printStackTrace();
        }

        return original;
    }

    // 工具方法：获取XML标签的文本内容
    private String getTagText(Document doc, String tagName) {
        NodeList nodes = doc.getElementsByTagName(tagName);
        return nodes.getLength() > 0 ? nodes.item(0).getTextContent().trim() : "";
    }

    // 工具方法：更新XML标签的文本内容
    private void updateTagText(Element parent, String tagName, String text) {
        NodeList nodes = parent.getElementsByTagName(tagName);
        if (nodes.getLength() > 0) {
            nodes.item(0).setTextContent(text);
        }
    }

    // 工具方法：复制属性集合到目标Exchange
    private void copyProperties(Map<String, Object> sourceProps, Exchange target) {
        for (Map.Entry<String, Object> entry : sourceProps.entrySet()) {
            target.setProperty(entry.getKey(), entry.getValue());
        }
    }
}

关键注意事项 #

• 无需手动维护循环索引，直接使用Camel内置的Exchange.LOOP_INDEX属性即可定位当前节点
• 若XML存在命名空间，需将getElementsByTagName替换为getElementsByTagNameNS来适配
• 异常处理逻辑可根据业务场景调整，比如添加日志告警或跳过异常节点的更新
• 确保项目包含XML解析依赖（JDK自带的javax.xml包通常足够，无需额外引入）

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