在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

首先,一个重要的注意事项:不要使用.paron,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？