scala - 在Scala中逐行并发读取和处理文件

Question

假设我需要应用两个函数 f: String => A和 g: A => B到一个大文本文件中的每一行，最终创建一个 B 的列表.

由于文件很大而且 f和 g很昂贵 我想让处理并发。我可以使用“并行集合”并执行类似 io.Source.fromFile("data.txt").getLines.toList.par.map(l => g(f(l)) 的操作但它不执行读取文件，f , 和 g同时。

在这个例子中实现并发的最佳方式是什么？

Answer 1

首先，重要的一点:不要使用.par在 List因为它需要复制所有数据(因为 List 只能顺序读取)。相反，使用类似 Vector 的东西，其中 .par无需复制即可进行转换。

似乎您以错误的方式考虑并行性。这是会发生的事情:

如果你有这样的文件:

0
1
2
3
4
5
6
7
8
9

和功能 f和 g :

def f(line: String) = {
  println("running f(%s)".format(line))
  line.toInt
}

def g(n: Int) = {
  println("running g(%d)".format(n))
  n + 1
}

然后你可以这样做:

io.Source.fromFile("data.txt").getLines.toIndexedSeq[String].par.map(l => g(f(l)))

并获得输出:

running f(3)
running f(0)
running f(5)
running f(2)
running f(6)
running f(1)
running g(2)
running f(4)
running f(7)
running g(4)
running g(1)
running g(6)
running g(3)
running g(5)
running g(0)
running g(7)
running f(9)
running f(8)
running g(9)
running g(8)

所以即使整个 g(f(l))操作发生在同一个线程上，你可以看到每一行都可以并行处理。因此，许多 f和 g操作可以在单独的线程上同时发生，但 f和 g对于特定的行将按顺序发生。

毕竟，这是您应该期望的方式，因为它实际上无法读取该行，运行 f ，然后运行 ​​ g在平行下。例如，它如何执行 g关于 f 的输出如果该行还没有被读取？