2

我正在使用 akka-camel 来处理文件。我最初的测试运行良好，但是当我开始传入实际的 xml 文件时，它正在因类型转换而呕吐。

这是我的消费者（非常简单，但在 msg.bodyAs[String]

class FileConsumer extends Consumer {
  def endpointUri = "file:/data/input/actor"

  val processor = context.actorOf(Props[Processor], "processor")

  def receive = {
    case msg: CamelMessage => {
      println("Parent...received %s" format msg)
      processor ! msg.bodyAs[String]
    }
  }
}

错误：

[ERROR] [04/27/2015 12:10:48.617] [ArdisSystem-akka.actor.default-dispatcher-5] [akka://ArdisSystem/user/$a] Error during type conversion from type: org.apache.camel.converter.stream.FileInputStreamCache to the required type: java.lang.String with value org.apache.camel.converter.stream.FileInputStreamCache@4611b35a due java.io.FileNotFoundException: /var/folders/dh/zfqvn9gn7cl6h63d3400y4zxp3xtzf/T/camel-tmp-807558/cos2920459202139947606.tmp (No such file or directory)
org.apache.camel.TypeConversionException: Error during type conversion from type: org.apache.camel.converter.stream.FileInputStreamCache to the required type: java.lang.String with value org.apache.camel.converter.stream.FileInputStreamCache@4611b35a due java.io.FileNotFoundException: /var/folders/dh/zfqvn9gn7cl6h63d3400y4zxp3xtzf/T/camel-tmp-807558/cos2920459202139947606.tmp (No such file or directory)

我想知道它是否与xml的实际内容有关。它们一点也不大（大约 70kb）。我怀疑我能否提供 XML 本身的实际示例。只是对为什么这么小并被转换为字符串的东西有问题感到困惑。其他虚拟示例 xml 文件运行良好。

编辑： 我的建议之一是启用 StreamCache，我这样做了。但是，它似乎仍然无法正常工作。正如 Ankush 评论的那样，这个错误令人困惑。我不确定它实际上是流问题还是转换问题。

http://camel.apache.org/stream-caching.html

添加了以下内容

  camel.context.setStreamCaching(true)
4

2 回答 2

6

我终于能够找出问题所在。问题不是坏数据，而是文件的大小。为了解决这个问题，您需要向骆驼上下文添加其他设置。

http://camel.apache.org/stream-caching.html

我使用的设置如下。如果我应该关闭流缓存，我将需要进一步研究，但这是一个开始。

camel.context.getProperties.put(CachedOutputStream.THRESHOLD, "750000");

或关闭流缓存

 camel.context.setStreamCaching(false)

希望这对其他人有帮助。

于 2015-04-30T04:06:06.443 回答
0

我们在评论 streamCaching() 帮助时遇到了同样的问题

from(IEricssonConstant.ROUTE_USAGE_DATA_INDIVIDUAL_PROCSESS) 
        //.streamCaching() 
         .split(new ZipSplitter()) .stopOnException()
         .streaming()
         .unmarshal().csv()
         .process(new UsageDataCSVRequestProcessor())
于 2019-01-30T16:05:47.683 回答