我正在研究一个主题“使用Apache Camel和如何处理大文件”.
目的是使用camel处理大文件而不将文件加载到内存中,因为它是一个超过5个GO的大文件.
我们找到了几个轨道,第一个轨道是使用分离器组件,允许我们逐行或逐块读取文件,但是如果我们使用分离器,我们无法再从头开始读取文件,功能需求是即使分割完成也能够读取文件的某些部分.
因此,我们必须使用缓存系统,将块放入缓存中以重用它们.
所以我们认为必须使用类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>
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方法可以使用标头中存在的每个缓存
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; }
thxs
解决方法
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。