基于Apache Camel实现带头部的大XML文件按字段条件拆分

针对你的大XML拆分需求，我整理了一套流式优先、低内存占用的实现方案，完美解决父节点保留、按条件分流的问题，同时避免重复读取文件：

核心思路概述 #

1. 流式提取公共头尾：用StAX仅读取XML开头的父节点（根标签+header内容）和结尾的根闭合标签，全程不加载整个文件到内存；
2. 实时分流追加：流式拆分每个order节点，按ID条件实时追加到对应输出文件；
3. 补全文件结构：拆分完成后给两个输出文件分别追加尾部标签，确保每个文件都是完整的XML结构。

具体路由实现 #

from("file:input?fileName=large.xml&noop=true")
    // 第一步：流式提取XML公共头部和尾部
    .process(exchange -> {
        InputStream in = exchange.getIn().getBody(InputStream.class);
        XMLInputFactory factory = XMLInputFactory.newInstance();
        XMLStreamReader reader = factory.createXMLStreamReader(in);
        
        StringBuilder header = new StringBuilder();
        String rootName = null;
        boolean reachedOrders = false;
        
        // 只读取到第一个order节点之前的内容
        while (reader.hasNext() && !reachedOrders) {
            int event = reader.next();
            switch (event) {
                case XMLStreamConstants.START_ELEMENT:
                    if (rootName == null) {
                        // 记录根节点名称并写入完整起始标签（含属性）
                        rootName = reader.getLocalName();
                        header.append("<").append(rootName);
                        for (int i = 0; i < reader.getAttributeCount(); i++) {
                            header.append(" ").append(reader.getAttributeLocalName(i))
                                  .append("=\"").append(reader.getAttributeValue(i)).append("\"");
                        }
                        header.append(">");
                    } else if ("order".equals(reader.getLocalName())) {
                        // 回退到当前节点，让后续split继续读取
                        reader.previous();
                        reachedOrders = true;
                    } else {
                        // 写入根节点下的其他子元素（如<documentHeader>等）
                        header.append("<").append(reader.getLocalName());
                        for (int i = 0; i < reader.getAttributeCount(); i++) {
                            header.append(" ").append(reader.getAttributeLocalName(i))
                                  .append("=\"").append(reader.getAttributeValue(i)).append("\"");
                        }
                        header.append(">");
                        // 读取元素文本内容（简单元素场景）
                        if (reader.hasNext()) {
                            int nextEvent = reader.next();
                            if (nextEvent == XMLStreamConstants.CHARACTERS) {
                                header.append(reader.getText().trim());
                            }
                            // 补全结束标签
                            if (reader.hasNext() && reader.next() == XMLStreamConstants.END_ELEMENT) {
                                header.append("</").append(reader.getLocalName()).append(">");
                            }
                        }
                    }
                    break;
                case XMLStreamConstants.CHARACTERS:
                    // 保留空白字符以维持原格式（可选）
                    String text = reader.getText().trim();
                    if (!text.isEmpty()) {
                        header.append(text);
                    }
                    break;
            }
        }
        
        // 存储头尾到Exchange属性，后续复用
        exchange.setProperty("XML_HEADER", header.toString());
        exchange.setProperty("XML_FOOTER", "</" + rootName + ">");
        // 传递未读完的流给后续split，避免重复读取文件
        exchange.getIn().setBody(reader);
    })
    // 第二步：初始化两个输出文件，写入公共头部
    .multicast()
        .to("direct:init-file-1", "direct:init-file-2")
    .end()
    // 第三步：流式拆分每个order节点，按条件分流
    .split().tokenizeXML("order", "*", true).streaming()
        .choice()
            // 用XPath判断ID是否以1开头
            .when(xpath("/order/@ID[starts-with(., '1')]").booleanResult())
                .to("direct:append-to-file-1")
            .otherwise()
                .to("direct:append-to-file-2")
        .endChoice()
    .endSplit()
    // 第四步：拆分完成后，给两个文件追加尾部标签
    .multicast()
        .to("direct:append-footer-1", "direct:append-footer-2")
    .end();

// 初始化ID以1开头的输出文件
from("direct:init-file-1")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-start-1.xml"))
    .setBody(simple("${exchangeProperty.XML_HEADER}"))
    .to("file:output?fileExist=Override");

// 初始化其他ID的输出文件
from("direct:init-file-2")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-others.xml"))
    .setBody(simple("${exchangeProperty.XML_HEADER}"))
    .to("file:output?fileExist=Override");

// 追加符合条件的order到第一个文件
from("direct:append-to-file-1")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-start-1.xml"))
    .setBody(simple("${body}"))
    .to("file:output?fileExist=Append");

// 追加符合条件的order到第二个文件
from("direct:append-to-file-2")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-others.xml"))
    .setBody(simple("${body}"))
    .to("file:output?fileExist=Append");

// 给第一个文件追加尾部标签
from("direct:append-footer-1")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-start-1.xml"))
    .setBody(simple("${exchangeProperty.XML_FOOTER}"))
    .to("file:output?fileExist=Append");

// 给第二个文件追加尾部标签
from("direct:append-footer-2")
    .setHeader(Exchange.FILE_NAME, constant("orders-id-others.xml"))
    .setBody(simple("${exchangeProperty.XML_FOOTER}"))
    .to("file:output?fileExist=Append");

关键问题解答 #

1. 是否需要先解析父标签？
   是的，但我们用StAX流式仅读取XML开头部分（到第一个order节点前），不会加载整个文件，性能损耗可以忽略。

2. 是否会导致两次流式读取？
   不会！我们把读取到一半的XMLStreamReader直接传递给后续的split组件，整个文件只被读取一次，完全避免重复IO。

3. 聚合方案是否合理？
   传统聚合会把所有符合条件的节点缓存到内存，不适合大文件。这里采用实时追加文件的方式，每个order节点处理完就写入磁盘，内存占用极低，是大文件场景的最优选择。

注意事项 #

• XPath在流式环境下可以正常使用，因为tokenizeXML拆分后每个order都是完整的XML元素；
• 如果需要保留原XML的缩进格式，可以在提取header时保留空白字符；
• 可添加onException处理XML解析错误、文件写入失败等异常场景。

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