个人中心
gpt4 book ai didi

file - Camel 消耗用base64编码的大文件

转载 作者：行者123 更新时间：2023-12-02 01:41:57 26 4
gpt4 key购买 nike

我正在尝试使用 apache camel 读取一个大文件。文件的内容是 base64 编码的，我想对其进行解码。读取基于 ASCII 的大型文本文件没有问题，但是这个...

在这种情况下我不能在行尾拆分它，因为base64行尾不是我数据的行尾(数据是csv)。稍后我想操作数据。

通过遵循 Camel 路线，它可以工作，但因此文件必须存在且未锁定，当路线开始时，否则我会收到 IllegalArgumentException。但我不知道，该文件何时会存在。

编辑:我忘记了:camel-version: 2.14.1 | java: 1.7

我需要解决我的问题，有人可以帮助我吗？

from("stream:file?fileName=/path/to/file&scanStream=true")
                .routeId(ROUTE_ID_BASE64)
                .unmarshal().base64()
                .setHeader("foo", constant("foo"))
                .aggregate(header("foo"), new StringBodyAggregator())
                .completionSize(1000)
                .completionTimeout(1500) 
                .to("file:" + this.tempPfad + "?fileName=" + this.tempDateiname + "&charset=utf-8&fileExist=Append");

我也试过:

from("file:" + this.eingabePfad + "?filter=#dateinamenFilter&maxDepth=1&readLock=changed&noop=true&charset=utf-8")
                .routeId(ROUTE_ID_BASE64)
                .unmarshal().base64()
                .setHeader("foo", constant("foo"))
                .aggregate(header("foo"), new StringBodyAggregator())
                .completionSize(1000)
                .completionTimeout(1500)
                .to("file:" + this.tempPfad + "?fileName=" + this.tempDateiname + "&charset=utf-8&fileExist=Append");

但随后出现以下错误:

ERROR Failed delivery for (MessageId: ID-smith-52872-1421933301634-0-1 on ExchangeId: ID-smith-52872-1421933301634-0-3). Exhausted after delivery attempt: 2 caught: org.apache.camel.component.file.GenericFileOperationFailedException: Cannot store file: /path/to/temp/tempDatei.csv. Processed by failure processor: FatalFallbackErrorHandler[Channel[DelegateSync[yy.xxxxx.myproject.routes.DateiPolling$1@5734ea]]]

Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
[routeBase64       ] [routeBase64       ] [file:///path/to/?charset=utf-8&filter=%23datein] [      1998]
[routeBase64       ] [unmarshal2        ] [unmarshal[org.apache.camel.model.dataformat.Base64DataFormat@153097e]         ] [         5]
[routeBase64       ] [setHeader1        ] [setHeader[foo]                                                                ] [         0]
[routeBase64       ] [aggregate1        ] [aggregate[header(foo)]                                                        ] [         0]
[routeBase64       ] [to1               ] [file:/path/to/temp/?fileName=tempDatei.csv&char] [        24]
[                  ] [process1          ] [yy.xxxxx.myproject.routes.DateiPolling$1@5734ea                     ] [        15]

Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
    Id                  ID-smith-52872-1421933301634-0-3
    ExchangePattern     InOnly
    Headers             {breadcrumbId=ID-smith-52872-1421933301634-0-1, CamelFileAbsolute=true, CamelFileAbsolutePath=/path/to/base64File.txt, CamelFileLastModified=1417777182000, CamelFileLength=267439971, CamelFileName=base64File.txt, CamelFileNameConsumed=base64File.txt, CamelFileNameOnly=base64File.txt, CamelFileParent=/path/to, CamelFilePath=/path/to/base64File.txt, CamelFileRelativePath=base64File.txt, CamelRedelivered=true, CamelRedeliveryCounter=1, CamelRedeliveryMaxCounter=1, foo=foo}
    BodyType            org.apache.commons.codec.binary.Base64InputStream
    Body                [Body is instance of java.io.InputStream]
]

Stacktrace
     ---------------------------------------------------------------------------------------------------------------------------------------
    org.apache.camel.component.file.GenericFileOperationFailedException: Cannot store file: /path/to/temp/tempDatei.csv
        at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:276)
        at org.apache.camel.component.file.GenericFileProducer.writeFile(GenericFileProducer.java:277)
        at org.apache.camel.component.file.GenericFileProducer.processExchange(GenericFileProducer.java:165)
        at org.apache.camel.component.file.GenericFileProducer.process(GenericFileProducer.java:79)
        at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
        at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:120)
        at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:416)
        at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
        at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
        at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
        at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
        at org.apache.camel.processor.aggregate.AggregateProcessor$1.run(AggregateProcessor.java:548)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at org.apache.camel.util.concurrent.SynchronousExecutorService.execute(SynchronousExecutorService.java:62)
        at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:110)
        at org.apache.camel.processor.aggregate.AggregateProcessor.onSubmitCompletion(AggregateProcessor.java:540)
        at org.apache.camel.processor.aggregate.AggregateProcessor.access$900(AggregateProcessor.java:82)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.onEviction(AggregateProcessor.java:853)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.onEviction(AggregateProcessor.java:814)
        at org.apache.camel.support.DefaultTimeoutMap.purge(DefaultTimeoutMap.java:212)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.purge(AggregateProcessor.java:826)
        at org.apache.camel.support.DefaultTimeoutMap.run(DefaultTimeoutMap.java:162)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
    Caused by: java.io.IOException: Stream closed
        at java.io.BufferedReader.ensureOpen(BufferedReader.java:115)
        at java.io.BufferedReader.read(BufferedReader.java:172)
        at org.apache.camel.converter.IOConverter$1.read(IOConverter.java:83)
        at java.io.InputStream.read(InputStream.java:170)
        at java.io.InputStream.read(InputStream.java:101)
        at org.apache.commons.codec.binary.BaseNCodecInputStream.read(BaseNCodecInputStream.java:158)
        at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)
        at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)
        at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177)
        at java.io.InputStreamReader.read(InputStreamReader.java:184)
        at java.io.BufferedReader.read1(BufferedReader.java:203)
        at java.io.BufferedReader.read(BufferedReader.java:279)
        at java.io.Reader.read(Reader.java:140)
        at org.apache.camel.util.IOHelper.copy(IOHelper.java:224)
        at org.apache.camel.component.file.FileOperations.writeFileByReaderWithCharset(FileOperations.java:402)
        at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:266)
        ... 29 more
    2015-01-22 14:28:26,113 [eTimeoutChecker] AggregateProcessor                            WARN  Error processing aggregated exchange. Exchange[base64File.txt]. Caused by: [org.apache.camel.component.file.GenericFileOperationFailedException - Cannot store file: /path/to/temp/tempDatei.csv]
    org.apache.camel.component.file.GenericFileOperationFailedException: Cannot store file: /path/to/temp/tempDatei.csv
        at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:276)
        at org.apache.camel.component.file.GenericFileProducer.writeFile(GenericFileProducer.java:277)
        at org.apache.camel.component.file.GenericFileProducer.processExchange(GenericFileProducer.java:165)
        at org.apache.camel.component.file.GenericFileProducer.process(GenericFileProducer.java:79)
        at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
        at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:120)
        at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:416)
        at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
        at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
        at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:105)
        at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
        at org.apache.camel.processor.aggregate.AggregateProcessor$1.run(AggregateProcessor.java:548)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at org.apache.camel.util.concurrent.SynchronousExecutorService.execute(SynchronousExecutorService.java:62)
        at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:110)
        at org.apache.camel.processor.aggregate.AggregateProcessor.onSubmitCompletion(AggregateProcessor.java:540)
        at org.apache.camel.processor.aggregate.AggregateProcessor.access$900(AggregateProcessor.java:82)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.onEviction(AggregateProcessor.java:853)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.onEviction(AggregateProcessor.java:814)
        at org.apache.camel.support.DefaultTimeoutMap.purge(DefaultTimeoutMap.java:212)
        at org.apache.camel.processor.aggregate.AggregateProcessor$AggregationTimeoutMap.purge(AggregateProcessor.java:826)
        at org.apache.camel.support.DefaultTimeoutMap.run(DefaultTimeoutMap.java:162)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:745)
    Caused by: java.io.IOException: Stream closed
        at java.io.BufferedReader.ensureOpen(BufferedReader.java:115)
        at java.io.BufferedReader.read(BufferedReader.java:172)
        at org.apache.camel.converter.IOConverter$1.read(IOConverter.java:83)
        at java.io.InputStream.read(InputStream.java:170)
        at java.io.InputStream.read(InputStream.java:101)
        at org.apache.commons.codec.binary.BaseNCodecInputStream.read(BaseNCodecInputStream.java:158)
        at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)
        at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)
        at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177)
        at java.io.InputStreamReader.read(InputStreamReader.java:184)
        at java.io.BufferedReader.read1(BufferedReader.java:203)
        at java.io.BufferedReader.read(BufferedReader.java:279)
        at java.io.Reader.read(Reader.java:140)
        at org.apache.camel.util.IOHelper.copy(IOHelper.java:224)
        at org.apache.camel.component.file.FileOperations.writeFileByReaderWithCharset(FileOperations.java:402)
        at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:266)
        ... 29 more

最佳答案

添加行

.split().tokenize(LINE_SEPARATOR).streaming()
现在可以用了。重要的是 
.end()
在路由的末尾，而不是在解码之后直接。

我还将 Aggregatorstrategy 从 String 更改为 byte[]，以防止 UTF-8 编码错误。

完整路线:

from("file:" + this.eingabePfad + "?filter=#dateinamenFilter&maxDepth=1&readLock=changed&noop=true")
                .routeId(ROUTE_ID_BASE64)
                .split().tokenize(LINE_SEPARATOR).streaming()
                .unmarshal().base64()
                .setHeader("foo", constant("foo"))
                .aggregate(header("foo"), new ByteArrayBodyAggregator())
                .completionSize(1000)
                .completionTimeout(1500)
                .to("file:" + this.tempPfad + "?fileName=" + this.tempDateiname + "&charset=utf-8&fileExist=Append")
                .end();

关于file - Camel 消耗用base64编码的大文件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/28091165/

26 4 0
文章推荐： google-chrome - 如何使用 netcat 提供 mht 文件？
文章推荐： visual-studio-2013 - NuGet:获取项目的所有可用更新
文章推荐： r - data.table 在函数中使用时的奇怪行为
文章推荐： logging - 如何让 Sails.js 日志记录使用具有不同日志级别的多个 Winston 传输？
行者123
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com