0

分散できる集中的なタスクがありますが、最終結果を蓄積して、後で別のメソッドに渡す必要があります。

具体的に言うと、大量のテキスト ファイルのコレクション内の個々のファイルの単語カウントを行っているとします。

これまでに作成したバージョンは次のようになります。

import scala.actors.Actor
import scala.actors.Actor._
import scala.collection.mutable.{ArrayBuffer => mArray}

case object Stop

class Collector(val bin: mArray[(String, Int)], tcount: Int) extends Actor {
  def act() {
    loop {
      receive {
        case (fn: String, val: Int) => {
          // the following "bin" object is what I ultimately need to get back
          bin.append((fn, val))
        }
        case Stop => {
          tcount -= 1
          if(tcount == 0) exit()
        }}}}}

class Processor(col: Collector, flist: Seq[File]) extends Actor {
  def act() {
    for(fn <- flist) {
      val wcount = count words in fn // just a place holder for word counting code
      col ! (fn, wcount)
    }    
    col ! (id, Stop)
  }
}

メインメソッドがコレクターの終了を待ってから「ビン」オブジェクトを処理するようにするいくつかの自作メソッドを考えることができます。

しかし、上記の「ビン」を取得してメインに戻す適切なスカラの方法は何ですか?

4

2 に答える 2

0

メソッドを使用してアクターに ping!!を実行し、将来の結果を作成できます。メッセージを受信したアクターは、 を使用して結果を入力できますreply。

ただし、代わりにエグゼキューターを使用することで、多くの管理作業を節約できます。これは、タスクの並列化に適しています (ここでは、アクターが特に役立つ同時実行状況は実際にはありません)。

以下は、改善された並行フレームワークが含まれているため、Scala 2.10 (近日公開予定、現在 2.10.0-M6 を使用できます) に基づいています。

import java.io.File
import concurrent._
import java.util.concurrent.Executors

// count words in a file - probably not the best possible way to write this
def wc(f: File): Int = io.Source.fromFile(f).getLines.map(
   _.split(' ').filterNot(_ == "").size).sum

// utility method to get files within a directory (non-recursive!)
def filesInDir(dir: File): Seq[File] = dir.listFiles.toSeq.filter(_.isFile)

// process method which takes a list of files, spawns a word-count for each
// and collects the individual future results into a `Map` whose future is returned
def process(files: Seq[File])(implicit exec: ExecutionContext)
: Future[Seq[(File, Either[Throwable, Int])]] = {
  val futs = files.map { f =>
    // `future` submits the body for asynchronous processing. In order to
    // gracefully handle IO errors, a successful result is wrapped in `Right`,
    // and an exception in `Left`. The caller can decide how to handle errors.
    future {
      f -> (try {
        Right(wc(f))
      } catch {
        case e: Throwable => Left(e)
      })
    }
  }
  // collect all the individual results into one result object.
  // the new single `Future` is only complete after all of the individual futures
  // have completed.
  Future.sequence(futs)
}

例：

// use as many threads as there are processor cores
val poolSize = sys.runtime.availableProcessors()

// create a new executor context to be used for spawning
implicit val exec = ExecutionContext.fromExecutor(
  Executors.newFixedThreadPool(poolSize))

// call the process
val map = process(filesInDir(new File("/my/directory")))

// execute some body when the future is completed
map.onSuccess { case m =>
  m foreach println
}
于 2012-08-08T16:48:23.973 に答える
0

アクターでは、アクター メソッドの呼び出しは危険な場合があるため、何も「取得」しません。

代わりに、アクターに回答を送信してもらいます。オブジェクトのケースを追加するGETか、あなたのケースbinでは、ハンドラーのどこかに を送信させることができますStop。

于 2012-08-08T15:03:23.050 に答える