Scala 中的并行文件处理答案

【问题标题】：Parallel file processing in ScalaScala 中的并行文件处理
【发布时间】：2012-07-19 13:24:07
【问题描述】：

假设我需要并行处理给定文件夹中的文件。在 Java 中，我将创建一个 FolderReader 线程以从文件夹中读取文件名，并创建一个 FileProcessor 线程池。 FolderReader读取文件名并将文件处理函数（Runnable）提交给pool executor。

在 Scala 中，我看到两个选项：

创建一个FileProcessor 演员池并使用Actors.Scheduler 安排一个文件处理函数。
在读取文件名时为每个文件名创建一个actor。

这有意义吗？最好的选择是什么？

【问题讨论】：

并发硬盘访问不好。它会导致额外的磁头运动。不要那样做。
Actors 在这里用错了。
@DanielC.Sobral 谢谢。你能解释一下为什么以及改用什么吗？
@Michael Scala 中的 Actor 不能很好地处理阻塞任务，例如上述的任务。下面的一些答案提供了替代方案。
除非您的文件位于不同的设备上，或者您的处理工作非常受 CPU 限制，否则并行处理文件不太可能有用。另外，您使用“并发”一词，但我认为您的意思是“并行”。 Actor 用于并发，而不是并行。期货和并行数据结构用于并行。

标签： scala akka actor

【解决方案1】：

根据你在做什么，它可能很简单

for(file<-files.par){
   //process the file
}

【讨论】：

这也是我第一时间想到的。很高兴在这里看到引用并行集合 API 的答案。 +1

【解决方案2】：

我建议尽我所能远离线程。幸运的是，我们有更好的抽象来处理下面发生的事情，在你的情况下，在我看来，你不需要使用演员（虽然你可以），但你可以使用更简单的抽象，称为 Futures。它们是 Akka 开源库的一部分，我认为将来也会成为 Scala 标准库的一部分。

Future[T] 只是在将来返回 T 的东西。

你需要运行一个future，就是有一个隐式的ExecutionContext，你可以从一个java executor服务派生。然后，您将能够享受优雅的 API 以及 future 是一个将集合转换为期货集合、收集结果等的单子这一事实。建议你看看http://doc.akka.io/docs/akka/2.0.1/scala/futures.html

object TestingFutures {
  implicit val executorService = Executors.newFixedThreadPool(20)
  implicit val executorContext = ExecutionContext.fromExecutorService(executorService)

  def testFutures(myList:List[String]):List[String]= {

    val listOfFutures : Future[List[String]] = Future.traverse(myList){
      aString => Future{
                        aString.reverse
                       }
     }
    val result:List[String] = Await.result(listOfFutures,1 minute)
    result

  }
}

这里发生了很多事情：

我正在使用Future.traverse，它作为第一个参数接收M[T]<:Traversable[T]，作为第二个参数接收T => Future[T]，或者如果您更喜欢Function1[T,Future[T]]，并返回Future[M[T]]
我正在使用Future.apply 方法创建Future[T] 类型的匿名类

查看 Akka 期货还有很多其他原因。

Futures 可以被映射，因为它们是 monad，即你可以链接 Futures 执行：

Future { 3 }.map { _ * 2 }.map { _.toString }
Future 有回调：future.onComplete、onSuccess、onFailure 和Then 等。
Futures 不仅支持遍历，还支持理解

【讨论】：

【解决方案3】：

理想情况下，您应该使用两个演员。一种用于读取文件列表，另一种用于实际读取文件。

您只需向第一个参与者发送一条“开始”消息即可开始该过程。然后actor可以读取文件列表，并向第二个actor发送消息。然后第二个参与者读取文件并处理内容。

拥有多个参与者，这可能看起来很复杂，实际上是一件好事，因为您有一堆对象相互通信，就像在理论上的 OO 系统中一样。

编辑：你真的不应该同时读取单个文件。

【讨论】：

【解决方案4】：

我打算准确地写出@Edmondo1984 做了什么，但他打败了我。 :) 我非常赞同他的建议。我还建议您阅读Akka 2.0.2 的文档。同样，我会给你一个更具体的例子：

import akka.dispatch.{ExecutionContext, Future, Await}
import akka.util.duration._
import java.util.concurrent.Executors
import java.io.File

val execService = Executors.newCachedThreadPool()
implicit val execContext = ExecutionContext.fromExecutorService(execService)

val tmp = new File("/tmp/")
val files = tmp.listFiles()
val workers = files.map { f =>
  Future {
    f.getAbsolutePath()
  }
}.toSeq
val result = Future.sequence(workers)
result.onSuccess {
  case filenames =>
    filenames.foreach { fn =>
      println(fn)
    }
}

// Artificial just to make things work for the example
Thread.sleep(100)
execContext.shutdown()

这里我使用sequence 而不是traverse，但区别将取决于您的需求。

与未来同行，我的朋友；在这种情况下，Actor 只是一种更痛苦的方法。

【讨论】：

与2.0.1相比有什么相关变化吗？
我不知道，但总是关注最新最好的东西是件好事。
请注意，Futures 现在是 Scala 标准库本身的一部分。如果你用 import scala.concurrent._ 换掉 akka._，上面的 REPL 就可以工作了

【解决方案5】：

但是如果使用演员，那有什么问题呢？

如果我们必须读/写一些属性文件。有我的 Java 示例。但仍然是 Akka Actors。

免得说我们有一个演员ActorFile 代表一个文件。嗯..可能它不能代表一个文件。正确的？（会很好）。那么它代表几个文件，比如PropertyFilesActor then：

为什么不使用这样的东西：

public class PropertyFilesActor extends UntypedActor {

    Map<String, String> filesContent = new LinkedHashMap<String, String>();

    { // here we should use real files of cource
        filesContent.put("file1.xml", "");
        filesContent.put("file2.xml", "");
    }

    @Override
    public void onReceive(Object message) throws Exception {

        if (message instanceof WriteMessage)  {
            WriteMessage writeMessage = (WriteMessage) message;
            String content = filesContent.get(writeMessage.fileName);
            String newContent = content + writeMessage.stringToWrite;
            filesContent.put(writeMessage.fileName, newContent);
        }

        else if (message instanceof ReadMessage) {
            ReadMessage readMessage = (ReadMessage) message;
            String currentContent = filesContent.get(readMessage.fileName);
            // Send the current content back to the sender
            getSender().tell(new ReadMessage(readMessage.fileName, currentContent), getSelf());
        }

        else unhandled(message);

    }

}

...一条消息将带有参数（文件名）

它有自己的in-box，接受如下消息：

WriteLine(文件名，字符串)
ReadLine(文件名，字符串)

这些消息将按顺序存储到in-box，一个接一个。演员将通过接收来自盒子的消息来完成其工作 - 存储/读取，同时发送反馈sender ! message。

因此，假设我们写入属性文件，并发送显示网页上的内容。我们可以开始显示页面（在我们发送消息以将数据存储到文件之后），一旦我们收到反馈，就使用刚刚更新的文件中的数据（通过 ajax）更新页面的一部分。

【讨论】：

【解决方案6】：

好吧，抓住你的文件并将它们放在一个并行结构中

scala> new java.io.File("/tmp").listFiles.par
res0: scala.collection.parallel.mutable.ParArray[java.io.File] = ParArray( ... )

那么……

scala> res0 map (_.length)
res1: scala.collection.parallel.mutable.ParArray[Long] = ParArray(4943, 1960, 4208, 103266, 363 ... )

【讨论】：