Iteracja linii w pliku równolegle (Scala)?

Ujmę to jako osobną odpowiedź, ponieważ zasadniczo różni się od mojej poprzedniej (i faktycznie działa)

Oto zarys rozwiązania wykorzystującego aktorów, który jest w zasadzie tym, co opisuje komentarz Kim Stebel. Istnieją dwie klasy aktorów, jeden aktor FileReader, który odczytuje poszczególne linie z pliku na żądanie, i kilku aktorów pracujących. Wszyscy pracownicy wysyłają wnioski o linie do czytnika, a linie przetwarzania równolegle, gdy są odczytywane z plik.

Używam tutaj Akka actors, ale użycie innej implementacji to w zasadzie ten sam pomysł.

case object LineRequest
case object BeginProcessing

class FileReader extends Actor {

  //reads a single line from the file or returns None if EOF
  def getLine:Option[String] = ...

  def receive = {
    case LineRequest => self.sender.foreach{_ ! getLine} //sender is an Option[ActorRef]
  }
}

class Worker(reader: ActorRef) extends Actor {

  def process(line:String) ...

  def receive = {
    case BeginProcessing => reader ! LineRequest
    case Some(line) => {
      process(line)
      reader ! LineRequest
    }
    case None => self.stop
  }
}

val reader = actorOf[FileReader].start    
val workers = Vector.fill(4)(actorOf(new Worker(reader)).start)
workers.foreach{_ ! BeginProcessing}
//wait for the workers to stop...

W ten sposób nie więcej niż 4 (lub jak wielu pracowników masz) nieprzetworzonych linii są w pamięci na raz.

Komentarze do odpowiedzi dana Simona dały mi do myślenia. Może spróbujemy zawinąć źródło W Strumień:

def src(source: Source) = Stream[String] = {
  if (source.hasNext) Stream.cons(source.takeWhile( _ != '\n' ).mkString)
  else Stream.empty
}

Wtedy można by je spożywać równolegle w ten sposób:

src(Source.fromFile(path)).par foreach process

Zdaję sobie sprawę, że to stare pytanie, ale może się okazać, że implementacja ParIterator W iterata library jest użyteczną implementacją nie wymaganą przez assembly:

scala> import com.timgroup.iterata.ParIterator.Implicits._
scala> val it = (1 to 100000).toIterator.par().map(n => (n + 1, Thread.currentThread.getId))
scala> it.map(_._2).toSet.size
res2: Int = 8 // addition was distributed over 8 threads

Poniżej pomógł mi osiągnąć

source.getLines.toStream.par.foreach( line => println(line))

W końcu napisaliśmy niestandardowe rozwiązanie w naszej firmie, abyśmy dokładnie zrozumieli równoległość.