0

我有一个基于actor的系统，在该系统中，我正在读取一个位于S3存储桶中的外部文件，并移动获取每个文件行并将其发送给另一个处理该特定行的actor。我很难理解的是在读取文件时引发异常时会发生什么。

我的代码如下：

import akka.actor._
import akka.actor.ActorSystem

class FileWorker(processorWorker: ActorRef) extends Actor with ActorLogging {

  val fileUtils = new S3Utils()

  private def processFile(fileLocation: String): Unit = {
    try{
         fileUtils.getLinesFromLocation(fileLocation).foreach {
         r =>
        {
           //Some processing happens for the line
            }
          }
        }
    }
    }catch{
      case e:Exception => log.error("Issue processing files from the following location %s".format(fileLocation))
    }
  }

  def receive() = {
    case fileLocation: String => {
      processFile(fileLocation)
    }
  }
}

在我的S3Utils课堂上，我定义了getLinesFromLocation如下方法：

 def getLinesFromLocation(fileLocation: String): Iterator[String] = {
    try{
       for {
             fileEntry <- getFileInfo(root,fileLocation)
          } yield fileEntry
    }catch{
      case e:Exception => logger.error("Issue with file location %s:         %s".format(fileLocation,e.getStackTraceString));throw e
    }
  }

我实际读取文件的方法是在私有方法中定义的getFileInfo

 private def getFileInfo(rootBucket: String,fileLocation: String): Iterator[String] = {
    implicit val codec = Codec(Codec.UTF8)
    codec.onMalformedInput(CodingErrorAction.IGNORE)
    codec.onUnmappableCharacter(CodingErrorAction.IGNORE)
    Source.fromInputStream(s3Client.
                       getObject(rootBucket,fileLocation).
                       getObjectContent()).getLines
  }

我写了上面的文章，假设位于 S3 上的基础文件不会被缓存在任何地方，我将简单地遍历恒定空间中的各个行并处理它们。如果在读取特定行时出现问题，迭代器将继续前进，而不会影响 Actor。

我的第一个问题是，我对迭代器的理解是否正确？实际上，我是否真的从底层文件系统（在本例中为 S3 存储桶）读取行，而不会对内存施加任何压力/或引入任何内存泄漏。

下一个问题是，如果迭代器在读取单个条目时遇到错误，是终止整个迭代过程还是继续下一个条目。

我的最后一个问题是，我的文件处理逻辑是否正确编写？

对此有一些见解会很棒。

谢谢

4

1 回答 1

1

看起来亚马逊 s3 没有异步实现，我们被固定的演员卡住了。所以你的实现是正确的，只要你为每个连接分配一个线程并且不会阻塞输入，也不会使用太多的连接。

采取的重要步骤：

1) processFile 不应阻塞当前线程。最好它应该将其输入委托给另一个参与者：

 private def processFile(fileLocation: String): Unit = {
     ...
         fileUtils.getLinesFromLocation(fileLocation).foreach {  r =>
            lineWorker ! FileLine(fileLocation, r)
         }

    ...
 }

2）制作FileWorker一个固定的演员：

## in application.config:
my-pinned-dispatcher {
    executor = "thread-pool-executor"
    type = PinnedDispatcher
}

// in the code:  
val fileWorker = context.actorOf(Props(classOf[FileWorker], lineWorker).withDispatcher("my-pinned-dispatcher"), "FileWorker")

如果迭代器在读取单个条目时遇到错误，是否会终止整个迭代过程？

是的，您的整个过程将被终止，并且演员将从其邮箱中获取下一个工作。

于 2013-06-10T09:32:34.220 回答