使用正则拆分大文件时Apache Camel出现内存溢出问题

问题场景 #

处理1GB级大型平面文件，记录以特定字符串分隔。使用普通字符串tokenize并开启流处理的路由运行正常：

from(incomingFilePath+"?delete=true")
                .split()
                .tokenize("ENDRECORD")
                .streaming()
                .bean(bean, "method")
                .bean(bean,"method")
                .end();

但切换为正则表达式tokenize时，流处理失效，抛出内存溢出异常：

from(incomingFilePath+"?delete=true")
                .split()
                .tokenize("\\|END(\\r\\n|\\r|\\n)", true)
                .streaming()
                .bean(bean, "method")
                .bean(bean,"method")
                .end();

异常信息：java.lang.OutOfMemoryError: Java heap space

原因分析 #

Camel中tokenize(regex, true)的实现逻辑和普通字符串tokenize不同。正则匹配模式下，组件会预读更多内容来完成正则匹配，导致大文件的部分甚至全部内容被加载到内存，直接打破了流处理的逐段读取机制，最终引发OOM。

解决方案 #

1. 规避正则tokenize，改用分段处理 #

如果分隔符是固定的|END+换行组合，完全可以不用正则：

• 先按换行符\n进行流分割，然后在bean方法中缓存行内容，直到遇到以|END结尾的行，再拼接成完整记录进行处理。示例调整后的路由：

from(incomingFilePath+"?delete=true&stream=true")
                .split()
                .tokenize("\n")
                .streaming()
                .bean(bean, "processLine")
                .end();

在processLine方法中维护一个缓冲区，每收到一行就检查是否以|END结尾，若是则处理缓冲区的完整记录并清空缓冲区，否则继续追加。

2. 优化正则表达式并调整匹配策略 #

如果必须用正则，要简化表达式并避免不必要的匹配分支：

• 将正则改为\\|END(?:\\r?\\n|\\r)，使用非捕获组(?:...)减少内存开销，同时用\\r?\\n覆盖Windows和Unix换行场景，减少匹配分支。
• 显式在文件URL中添加stream=true，强制开启流读取模式。

3. 自定义Tokenizer实现流匹配 #

如果上述方案都不满足，可自定义Tokenizer接口实现类，基于输入流逐字符读取，手动识别正则分隔符，完全控制内存使用：

public class RegexStreamTokenizer implements Tokenizer {
    private final Pattern pattern;
    private Matcher matcher;
    private InputStream inputStream;
    
    // 构造方法及初始化逻辑
    public RegexStreamTokenizer(String regex) {
        this.pattern = Pattern.compile(regex);
    }

    @Override
    public boolean hasNext() {
        // 逐字符读取输入流，匹配正则分隔符的逻辑实现
    }

    @Override
    public String next() {
        // 返回匹配到的分隔符前的完整记录内容
    }
}

然后在路由中使用自定义Tokenizer：

from(incomingFilePath+"?delete=true&stream=true")
    .split(body().tokenize(new RegexStreamTokenizer("\\|END(\\r\\n|\\r|\\n)")))
    .streaming()
    .bean(bean, "method")
    .end();

4. 临时应急：调整JVM堆内存 #

如果只是临时解决问题，可以适当调大JVM堆内存（如-Xmx4G），但这只是治标不治本，无法从根本上解决流处理失效的问题。

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