Kafka消息消费转发时内存持续飙升问题排查求助

我有一条从Kafka Topic消费JSON字符串并转发至direct:streamerRoute的Camel路由，实现代码如下：

from("direct:streamerRoute").routeId("streamerRoute")
                .process(new Processor() {
                    @Override
                    public void process(Exchange exchange) throws Exception {

                        DocumentContext jsonContext = JsonPath.using(configuration).parse(exchange.getIn().getBody(String.class));
                        String tableName = jsonContext.read(eventTableNameInputExpr);
                        String eventId = jsonContext.read(eventIdInputExpr);
                        log.info("eventId: {}, tableName={}", eventId, tableName);

                        exchange.setProperty("staticDataYes", isStaticData(tableName));
                        exchange.setProperty("transactionDataYes", isTransactionData(tableName));
                    }
                })
                .choice()
                    .when(exchangeProperty("staticDataYes").isEqualTo(true))
                        .to("direct:StaticData")
                    .when(exchangeProperty("transactionDataYes").isEqualTo(true))
                        .to("direct:TransactionData")
                    .otherwise()
                        .log("The event is not either static data of transaction data")
                .end();

direct:StaticData和direct:TransactionData最终会将消息发布至各自对应的Topic。但运行中发现POD内存持续增长，处理15万条消息后内存可达4.5GB。本地用JProfiler观测到大量byte[]对象堆积，内存随消息处理不断增长，且消息处理完成后内存未回落，疑似内存泄漏。已在主消费路由中添加.noMessageHistory()和.streamCaching("false")配置，问题仍未解决。

排查思路与解决方案 #

一、聚焦byte[]来源分析 #

• 定位字节数组归属：在JProfiler中追踪byte[]的引用链，确认是Kafka原始消息体、JsonPath解析缓存还是后续路由中间产物。重点检查Kafka消费者配置，比如allowSerializedHeaders、autoCommitEnable等参数是否导致消息资源未释放。
• 清理JsonPath解析资源：代码中创建的DocumentContext可能存在内部缓存未释放，尝试在process方法末尾显式调用jsonContext.close()（若支持），或改用Jackson的ObjectMapper直接解析指定字段，避免JsonPath的内存缓存问题：

  ObjectMapper mapper = new ObjectMapper();
  JsonNode root = mapper.readTree(exchange.getIn().getBody(String.class));
  String tableName = root.get(eventTableNameInputExpr.replaceFirst("\\$\\.", "")).asText();
  String eventId = root.get(eventIdInputExpr.replaceFirst("\\$\\.", "")).asText();
  

二、Camel路由内存管理优化 #

• 主动清理Exchange资源：在路由末尾添加处理器，清空消息体和自定义属性，确保资源及时释放：

  .process(exchange -> {
      exchange.getIn().setBody(null);
      exchange.getProperties().clear();
  })
  

• 检查下游路由泄漏：排查direct:StaticData和direct:TransactionData路由，确认是否持有Exchange引用、存在未关闭的IO流或连接。
• 调整Kafka消费者参数：修改fetch.min.bytes、fetch.max.bytes控制批量拉取大小，避免一次性加载过多消息；合理设置pollTimeout，防止未处理消息堆积内存。

三、JVM与日志层面优化 #

• GC日志分析：启用GC日志（-XX:+PrintGCDetails -XX:+PrintGCTimeStamps），确认是内存泄漏还是GC回收不及时；调整JVM堆内存参数（-Xmx、-Xms），避免堆内存过大导致GC延迟。
• 日志配置检查：若使用异步日志，确认日志框架（如Logback）的队列大小配置合理，避免日志消息堆积内存；降低不必要的日志级别，减少大对象日志输出。

四、全局缓存检查 #

• 排查isStaticData(tableName)和isTransactionData(tableName)方法，确认是否使用静态Map等全局缓存且未清理过期数据，导致缓存持续膨胀占用内存。

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