我正在研究一个主题“在 Apache Camel 中使用缓存以及如何处理大文件”。

目的是用 camel 处理大文件而不将文件加载到内存中，因为它是一个超过 5 GO 的大文件。

我们找到了几个轨道，第一个轨道是使用拆分器组件，允许我们逐行或逐 block 读取文件，但是如果我们使用拆分器，我们将无法再次读取文件从一开始，功能需求就是即使在拆分完成后也能够读取文件的某些部分。

所以我们必须使用缓存系统，将 block 放入缓存中以重用它们。

所以我们认为必须使用类 CachedOutputStream 将拆分器后的文件的某些部分写入磁盘，该类还提供了对磁盘数据进行加密的能力。

例子如下:

<camelContext xmlns="http://camel.apache.org/schema/spring" trace="false" streamCache="true">

    <streamCaching id="myCacheConfig"  spoolDirectory="target/cachedir" spoolThreshold="16"/>

    <route id="SPLIT-FLOW" streamCache="true">
        <from uri="file:src/data/forSplitCaching\SimpleRecord?noop=true"/>
        <split streaming="true">
            <tokenize token="\n"/>
            <to uri="direct:PROCESS-BUSINESS"/>
        </split>
    </route>

    <route id="PROCESS-BUSINESS" streamCache="true">
        <from uri="direct:PROCESS-BUSINESS"/>
        <bean ref="ProcessBusiness" method="dealRecord"/>
        <choice>
            <when>
                <simple>${in.header.CamelSplitComplete} == "true"</simple>
                <to uri="direct:STREAM-CACHING"/>
            </when>
        </choice>
    </route>

    <route id="STREAM-CACHING">
        <from uri="direct:STREAM-CACHING"/>
        <bean ref="ProcessStreamCaching" method="usingStream"/>
        <setHeader headerName="CamelFileName">
            <simple>${header.CamelFileName}.${header.CamelSplitIndex}</simple>
        </setHeader>
        <to uri="file:src/out"/>
    </route>

</camelContext>

dealRecord 方法将拆分后的每一行放入缓存中:

public void dealRecord(Exchange exchange) throws Exception { 

   String body; 
   File file; 
   String[] files; 
   boolean isSplitComplete; 

   body = (String) exchange.getIn().getBody(); 
   isSplitComplete = (boolean) exchange.getProperties().get("CamelSplitComplete"); 

   CachedOutputStream cos = new CachedOutputStream(exchange, false); 
   cos.write(body.getBytes("UTF-8")); 

   file = new File("target/cachedir"); 
   files = file.list(); 
   for (String nameTmpfile : files) { 
      LOG.info("Genered File [" + nameTmpfile + "]"); 
   } 

   lstCache.add(cos); 

   if(isSplitComplete){ 
      exchange.getIn().setHeader("Cached",lstCache); 
   } 
} 

usingStream方法，可以使用header中存在的各个缓存

public byte[] usingStream(Exchange exchange) throws InputStreamException { 

   final ArrayList<CachedOutputStream> lstcache; 
   byte[] bytesMessage; 
   StringBuilder messageCompleteOut = new StringBuilder(); 
   InputStream is = null; 

   lstcache = (ArrayList<CachedOutputStream>) exchange.getIn().getHeader("Cached"); 
   for (CachedOutputStream oneCache : lstcache) { 
      try { 
         is = oneCache.getWrappedInputStream(); 
         String messageInputstream = toString(is); 
         LOG.info("Message of Cache ["+ messageInputstream +"]"); 
         messageCompleteOut.append(messageInputstream); 
         messageCompleteOut.append(System.lineSeparator()); 
      } catch (IOException e) { 
         LOG.error(InputStreamException.ERROR_MANIPULATING_INPUT_STREAM_CHANNEL); 
         throw new InputStreamException(InputStreamException.ERROR_MANIPULATING_INPUT_STREAM_CHANNEL,e); 
      } 
      // On ferme le flux 
      IOHelper.close(is); 
   } 
   bytesMessage = messageCompleteOut.toString().getBytes(Charset.forName("UTF-8")); 
   return bytesMessage; 
} 

这个解决方案看起来不错吗？或者也许有更好的方法？

谢谢

最佳答案

GenericFileMessage(文件组件使用的消息实现)除非需要，否则不会将文件内容加载到内存中。所以实际上你只需要确保你不会以强制转换它的方式访问正文。您还可以编写自己的消息(继承自 GenericFileMessage)并阻止此类转换，或者返回不同的内容(某种“摘要”)。

一路上的处理器可以获取文件在文件系统中的位置(从消息头)并直接打开它，可能用其他消息替换文件消息。

关于java - 缓存流处理大文件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/36542766/