混合使用ForkJoin+Actor+Future实现一千万个不重复整数的排序(Scala示例)

目标

      实现一千万个不重复整数的排序，可以一次性加载到 2G 的内存里。

      本文适合于想要了解新语言 Scala 并发异步编程框架 Akka, Future 的筒鞋。 读完本文后，将了解如何综合使用 ForkJoin 框架、 Akka 模型、以及 Future 进行并发异步编程，还有一系列小的编程点。 

任务拆分

      首先要进行任务拆分。要实现一千万个不重复整数的排序， 可以拆分为三个子任务：

      (1)  生成一千万的不重复整数并写入文件 NumberGeneratorTask；

      (2) 从文件读取并检测确实生成的是一千万个不重复的整数 CheckUnduplicatedNumbersActor；

      (3)  从文件读取整数进行排序和排序检测 BigfileSortActor。接下来逐一实现这些子任务。

      入口如下。这里使用了 Akka 的框架及 ForkJoin 实例。其中启动 NumberGeneratorTask 抽离到一个工具类 ForkJoinPoolStartup 来实现，更好地维护和复用， 比如启动不同参数的 NumberGeneratorTask 。

      从 system.actorOf(Props(new CheckUnduplicatedNumbersActor(numbers, bigfileSortActor)), name="checkNumberActor") 可以看出如何创建带参数或不带参数的 Actor 实例。注意到，如果任务流程是从 NumberGeneratorTask -> checkNumberTask -> bigfileSortTask , 那么，对应的Actor 顺序正好是反过来：先创建任务流程最靠后的Actor，再创建流程中靠前的Actor ，因为靠前的Actor 需要持有流程中下一个Actor的引用以便向其发送消息。BigFileSortActor 持有 ActorSystem 实例引用 system ， 便于在排序及检测完成后终止整个 Actor 系统。

package scalastudy.concurrent.billionsort

import akka.actor.{ActorSystem, Props}

import scalastudy.concurrent.ForkJoinPoolStartup
import scalastudy.concurrent.config.ActorSystemFactory
import scalastudy.concurrent.billionsort.Constants._

/**
  * Created by shuqin on 16/5/18.
  */
object BillionNumberSort extends App {

    launch()

    def launch(): Unit = {
        ForkJoinPoolStartup.start(createActors(), poolWaitSecs)
    }

    def createActors():NumberGeneratorTask = {
        val system:ActorSystem = ActorSystemFactory.newInstance()
        val bigfileSortActor = system.actorOf(Props(new BigFileSortActor(numbers, system)))
        val checkNumberActor = system.actorOf(Props(new CheckUnduplicatedNumbersActor(numbers, bigfileSortActor)), name="checkNumberActor")
        val numGenTask = new NumberGeneratorTask(numbers, 0, rangeMaxNumber, checkNumberActor)
        return numGenTask
    }

}

package scalastudy.concurrent

import java.util.concurrent.{ForkJoinPool, TimeUnit}

import scalastudy.concurrent.billionsort.NumberGeneratorTask

/**
  * Created by shuqin on 17/4/27.
  */
object ForkJoinPoolStartup {

  def start(entranceTask:NumberGeneratorTask, waitSecs:Int):Unit = {
    val pool = new ForkJoinPool()
    pool.execute(entranceTask)
    pool.shutdown
    pool.awaitTermination(waitSecs, TimeUnit.SECONDS)
    pool.shutdownNow
    assert( pool.isTerminated == true )
  }

}

生成一千万个不重复整数

ForkJoin的使用

　　显然，这个子任务是可以采用 ForkJoin 来完成的。 ForkJoin 是分治思想的框架性实现， 将原问题分解为同样性质的多个子问题，然后将子问题的解组合起来得到原问题的解。通常采用二分法。实现上，通常会采用递归结构， 注意递归不要太深。 actorName ! message 表示向名称为 actorName 的Actor 实例发送 message 消息，message 可以是任意数据结构，字符串、列表、元组、对象等。这里发送了两种类型： 整数列表 randInts.map(i=>i+start).toList 或 整数元组 (start, end) 。生成随机无序整数使用了已有的Java类 RandomSelector 的方法，这表明了，Scala 可以轻易无缝地使用 Java 现有的代码和库。

   　NumberGeneratorTask 的实现如下：

package scalastudy.concurrent.billionsort

import java.util.concurrent.RecursiveAction

import akka.actor.ActorRef
import zzz.study.algorithm.select.RandomSelector

import scalastudy.concurrent.billionsort.Constants.threshold
import scalastudy.concurrent.billionsort.Constants.debug

/**
  *  Created by shuqin on 16/5/19.
  *
  * 在 [start, end] 选出 num 个不重复的整数
  *
  */
class NumberGeneratorTask(num:Int, start:Int, end:Int, checkNumberActor: ActorRef)  extends RecursiveAction {

    override def compute(): Unit = {

        if (debug) {
            println("Select: " + num  + " unduplicated numbers from [" + start + " " + end + ")");
        }

        if (num <= threshold) {

            if (num > end - start+1) {
                checkNumberActor ! start.to(end).toList
            }
            else {
                val randInts = RandomSelector.selectMDisorderedRandInts2(num, end-start+1)
                checkNumberActor ! randInts.map(i=>i+start).toList
            }
        }
        else {
            val middle = start/2 + end/2
            val leftTask = new NumberGeneratorTask(num/2, start, middle, checkNumberActor)
            val rightTask = new NumberGeneratorTask((num+1)/2, middle+1, end, checkNumberActor)

            if (debug) {
                println("Left: [" + start + "-" + middle + "," + num/2 + "]")
                println("Right: [" + (middle+1) + "-" + end + "," + (num+1)/2 + "]")
            }

            leftTask.fork
            rightTask.fork
            leftTask.join
            rightTask.join
            checkNumberActor ! (start, end)
        }
    }

}

检测生成的一千万个整数不重复

Actor通信

     Akka Actor 并发模型一个重要优势在于为代表单任务的Actor提供了健壮可扩展的消息传递通信机制。继承 Actor 之后，需要覆写指定方法 override def receive: Receive ，使用灵活而强大的 case 语句（偏函数）来匹配消息的类型及消息的值，从而做不同的判断和操作。

     怎样判断整数生成任务完成从而可以开始检测了呢？在 NumberGeneratorTask 生成最后一组整数时并回退到最开始的调用层时，就会发送 (0, Constants.rangeMaxNumber) 作为信号， 而 CheckUnduplicatedNumbersActor 则通过 case (0,Constants.rangeMaxNumber) 可以匹配到这一点。

Trait与Actor的分工

     最开始，接收NumberGeneratorTask 传来的消息进行处理以及检测生成的整数不重复都写在了 CheckUnduplicatedNumbersActor 这一个类里。后来想了想，觉得这个类混杂了不同的功能和职责，因此拆分成了两个类：CheckUnduplicatedNumbersActor 和 CheckUnduplicatedNumbers 。 其中 CheckUnduplicatedNumbersActor 负责任务协作和计算调度， 而 CheckUnduplicatedNumbers 负责检测生成的整数不重复的实际工作。职责明晰，各司其责，分开独立发展。使用 with CheckUnduplicatedNumbers 语法，可以使得具体类混入 trait 的功能，实现多重能力继承，既能利用多重继承的优势，又能避免多重字段继承带来的问题。

策略模式的使用

      CheckUnduplicatedNumbers 使用了策略模式。对于一千万个整数来说，内存占用 40M 左右， 2G 内存是装滴下的， 若是十亿个整数，那么就需要 4G，就不能一次性加载了。因此这里定义了个接口，并实现了一次性加载策略和位图策略。可以使用位图来检测不重复的整数，甚至可以直接进行排序。可参考 《位图排序（位图技术应用）》。 BitMapStrategy 实现了使用位图技术来对一千万个不重复整数进行排序的策略。读者感兴趣可以实现多次加载策略，以应对内存不够的情形。

      此外，Source.fromFile(filename).getLines 这里返回的是迭代器， 如果内存不够用的话，就必须使用这个方法，而不是 Source.fromFile(filename).getLines.toList ， 后者会将所有行全部加载到内存中而导致 OutOfMemoryError . 

package scalastudy.concurrent.billionsort

import java.io.{File, PrintWriter}

import akka.actor.{Actor, ActorRef}

import scala.collection.immutable.List
import scalastudy.concurrent.billionsort.Constants.filename

/**
  * Created by shuqin on 16/5/19.
  */
class CheckUnduplicatedNumbersActor(val numbers:Int, bigfileSortActor: ActorRef) extends Actor
    with CheckUnduplicatedNumbers {


    val fwResult = new PrintWriter(new File(filename))

    var count = 0
    val useBigFileSort = true

    override def receive: Receive = {

        case numberList: List[Int] =>
            fwResult.write(numberList.mkString(" ") + "
");
            count += numberList.length

        case (0, Constants.rangeMaxNumber) =>
            println("Reach End.")
            println("Expected: " + numbers + " , Actual Received: " + count)
            assert(count == numbers)
            fwResult.flush
            fwResult.close

            checkUnduplicatedNumbers(filename, numbers)
            if (useBigFileSort) {
                bigfileSortActor ! filename
            }

        case _ => println("未知消息，请检查原因 ！")
    }

}

package scalastudy.concurrent.billionsort

import java.io.{File, PrintWriter}

import zzz.study.datastructure.vector.EnhancedBigNBitsVector

import scala.collection.mutable.Set
import scala.io.Source

import scalastudy.concurrent.billionsort.Constants.rangeMaxNumber

/**
  * Created by shuqin on 17/4/26.
  */
trait CheckUnduplicatedNumbers {

  def checkUnduplicatedNumbers(filename:String, numbers:Int): Unit = {

    assert(new OnceLoadStrategy().checkUnduplicatedNumbersInFile(filename, numbers) == true)
    assert(new BitMapStrategy().checkUnduplicatedNumbersInFile(filename,numbers) == true)
    println("checkUnduplicatedNumbers passed.")
  }

  /**
    * 一次性加载所有数到内存, 适用于内存可以装下所有数的情况
    * 比如 10000000 个整数占用 40M　空间, 2G 内存是绰绰有余的, 但十亿占用 4G 空间失效
    */
  class OnceLoadStrategy extends CheckUnduplicatedStrategy {

    def checkUnduplicatedNumbersInFile(filename:String, numbers:Int):Boolean = {
      var numbersInFile = 0
      val unDupNumberSet = Set[Int]()
      Source.fromFile(filename).getLines.
        foreach { line =>
          val numbersInLine = line.split("\s+").map(Integer.parseInt(_)).toSet
          numbersInFile += numbersInLine.size;
          unDupNumberSet ++= numbersInLine
                }
      println(s"Expected: ${numbers} , Actual In File: ${numbersInFile} ")
      println("Unduplicated numbers in File: " + unDupNumberSet.size)
      unDupNumberSet.size == numbers
    }
  }

  /**
    * 使用位图技术来检测不重复的数, 实际上还能用于排序
    * N个数只要 4(N/32+1) = N/8 + 4 个字节
    * 十亿个数只要 125000004B = 125MB
    * 反过来, 内存 1G 的机器可以对 80亿 的不重复数进行排序
    */
  class BitMapStrategy extends CheckUnduplicatedStrategy {

    val nbitsVector = new EnhancedBigNBitsVector(rangeMaxNumber)

    override def checkUnduplicatedNumbersInFile(filename: String, numbers:Int): Boolean = {
      Source.fromFile(filename).getLines.
        foreach { line =>
          val numbersInLine = line.split("\s+").map(Integer.parseInt(_)).toSet
          numbersInLine.foreach { num =>
            nbitsVector.setBit(num)
          }
        }

      val undupTotal = checkAndSort(filename)
      println(s"undupTotal: ${undupTotal}")
      assert(undupTotal == numbers)
      return true
    }

    def checkAndSort(filename: String): Integer = {
      val fwFinalResult = new PrintWriter(new File(s"${filename}.sorted.txt"))
      val sorted = nbitsVector.expr()
      var undupTotal = sorted.size()
      fwFinalResult.flush()
      fwFinalResult.close()
      return undupTotal
    }

  }

  trait CheckUnduplicatedStrategy {
    def checkUnduplicatedNumbersInFile(filename:String, numbers:Int):Boolean
  }

} 

大文件排序

      Oh, 终于进入正题了。大文件排序当然采用归并排序了。 在这个实现里，值得注意的是采用了 Future 全异步框架。

Future全异步框架

      可以看到：

　　(1)　def produceFuture(line:String): Future[List[List[Int]]]　将文件的每一行(包含 threshold 个整数)转化为一个对行内整数排序的 Future, 可以在后续获取结果； 对于一个文件，就是获得了 futureTasks =  List[Future[List[List[Int]]]] ;  List[List[Int]] 是为了让后面的 Reduce 语法上走得通。　

　  (2)   val sortedListsFuture = Future.reduce(futureTasks)(cat(_,_)) 将 List[Future[List[List[Int]]]] 整合成一个 TotalFuture, 这个 TotalFuture 的结果是 futureTasks 里面的所有 Future 结果的连接； 每一个 Future 的结果是一个已排序的列表； 那么 TotalFuture 的结果是一个已排序列表的列表。List[Future[List[List[Int]]]] 看着是不是有点头晕目眩？这生动地说明，要想玩转编程，数据结构功底要扎实！

　  (3)   注意到下面这行代码： 是将一个 Future A 转化为另一个 Future B. 其中 B 的结果是基于 A. 在本例中，即是将已排序列表的列表合并为最终列表，但仍然返回的是 Future 而不是最终列表。为什么要这么写， 而不是将 sortedListsFuture 的结果取出来再合并呢？ 这是由于之前的所有动作都是异步的。 如果应用只是取排序的结果，那么也没什么； 但如果应用要将 sortedListsFuture 的结果写入文件呢？ 进而还要做一下排序检测？ 那么， 就不得不在后面加入 TimeUnit.SECONDS.sleep(n) 的代码， 让主线程休息一会了(因为前面整个是异步的， 在 sortedListsFuture 还没完成时，后面的代码就会被执行了)！ 而且你得不断估计前面的排序／合并操作究竟大约需要多少时间从而不断调整休眠的时间！ 之前就是这样实现的！ 但这样并不符合 Future 异步框架的初衷！ 因此后面，我突然觉得要写成全异步的， 也体验到了写成全异步应用的滋味~~ :) 要求确实是有点高，需要不断从 Future 转换成新的 Future ~~ 同时你也发现， Scala Future 也提供了一个帮助编写全异步框架的 API ~~

sortedListsFuture map {
      value:List[List[Int]] =>
            CollectionUtil.mergeKOrderedList(value)
}

　  (4) 由于后面将排序结果写入文件以及从文件检测排序是否 OK 都是同步的，因此，可以在排序 Future 完成后执行。 注意到 Future 的非阻塞写法： f.onComplete { case Success(result) => doWith(result) ;  case Failure(ex) => doWith(ex) } 　 

     (5)  为了将列表链接起来，也试错了好几次：  (x :: y :: Nil).flatten ;　如果写成 reduce(_ :: _ :: Nil) 是会报错的； 写成 reduce(_.flatten :: _.flatten :: Nil) 最终会合并成两个列表不符合预期。 

package scalastudy.concurrent.billionsort

import java.io.{File, PrintWriter}
import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSystem, Props}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.{Failure, Success}
import scalastudy.utils.{CollectionUtil, DefaultFileUtil, PathConstants}


/**
  * Created by shuqin on 16/5/20.
  */
class BigFileSortActor(numbers: Int, actorSystem: ActorSystem) extends Actor with SortChecker {

    override def receive: Receive = {

        case filename:String =>
            println("Received File: " + filename)
            sortFile(filename)
    }

    def produceFuture(line:String): Future[List[List[Int]]] = {
        val origin = line.split("\s+").map( s => Integer.parseInt(s)).toList
        Future {
            List(origin.sorted)
        }
    }

    def cat(x: List[List[Int]],y:List[List[Int]]): List[List[Int]] = {
        return (x :: y :: Nil).flatten
    }

    def obtainSortedFuture(futureTasks:List[Future[List[List[Int]]]]):Future[List[Int]] = {
        val sortedListsFuture = Future.reduce(futureTasks)(cat(_,_))

        sortedListsFuture map {
            value:List[List[Int]] =>
                CollectionUtil.mergeKOrderedList(value)
        }
    }

    def sortFile(filename:String):Unit = {

        val futureTasks = DefaultFileUtil.readFileLines(filename).map(produceFuture(_))
        println("task numbers: " + futureTasks.size)

        val allNumberSortedFuture = obtainSortedFuture(futureTasks)

        allNumberSortedFuture.onComplete {
            case Success(value:List[Int]) =>
                println("sort finished.")
                writeSorted(value, filename)
                checkSorted(filename, numbers)

                println("sleep 3s and then begin to stop all.")
                TimeUnit.SECONDS.sleep(3)
                actorSystem.shutdown
            case Failure(ex) =>
                println("Sort failed: " + ex.getMessage)
        }
    }

    def writeSorted(allNumberSorted: List[Int], filename: String): Unit = {
        val fwResult = new PrintWriter(new File(filename + ".sorted.txt"))
        fwResult.write(allNumberSorted.mkString("
"))
        fwResult.flush
        fwResult.close
    }
}

object BigFileSortActorTest {

    def main(args:Array[String]):Unit = {

        val numbers = 10000000
        val system = ActorSystem("BigFileSortActorTest")
        val bigFileSortActor = system.actorOf(Props(new BigFileSortActor(numbers, system)),name="bigFileSortActor")
        bigFileSortActor ! PathConstants.projPath + "/data/" + numbers +".txt"

        TimeUnit.SECONDS.sleep(640)
        system.shutdown

    }

}

合并有序链表

     合并有序链表是归并排序的核心环节，也是归并排序的性能关键之所在。

     CollectionUtil 实现了一个二路有序列表合并和K路有序列表合并。 其中二路有序列表合并和K路有序列表均分别使用了三种方法来实现：一种是过程式的插入合并，一种是结合foldLeft 或 reduce 的函数式的实现，一种是更高效的实现。读者可以体会三种的差异。过程式的合并时间效率尚可，但空间开销比较大，大数据量时容易导致OOM， 函数式的做法时间效率不够优，而更高效的实现尽可能结合两者的优势。 

      注意到，K路有序列表合并使用到了 klists.par.reduce(merge) ， 将普通列表转化为并行列表，以一定空间开销换取可以并行地合并大量列表的时间效率。实际调试查看大量列表合并进度时，可以在 merge 函数的返回结果行之上加一行 println(result.size)，查看合并后的列表大小。 

      Scala 的 List 是一个链表　(head::(tail::Nil))， 空列表可以用 List(), Nil 来表示； 将元素添加在列表头部可使用 elem :: list , 在列表尾部添加元素使用 list :+ elem ; 列表链接使用 list1 ::: list2。默认的 List 是不可变的，意味着每次操作都会创建一个新的 List , 对于大有序列表合并的空间效率是不能接受的。因此，大有序列表的合并必须采用可变列表 ListBuffer . 至于如何做到 O（n+m）, 还需要探索。

     NOTE： 通过性能测试发现，merge 时间效率是最高的，但是当列表很大时会抛GC exceed limit 异常；  mergeInplace 的性能次之，在列表长度到达 10000 时，性能开始急速下降几十倍； mergeFunctional 的时间性能最差。看上去情况并不如预料。估计应该是使用 List 姿势不对。后续打算专门开一篇文章讨论合并有序列表的优化。

     Scala 的集合性能参考： http://docs.scala-lang.org/overviews/collections/performance-characteristics.html

package scalastudy.utils

import scala.collection.mutable
import scala.collection.mutable.{ListBuffer, Map}

import scala.math.pow

/**
  * Created by lovesqcc on 16-4-2.
  */
object CollectionUtil {

  def main(args: Array[String]): Unit = {

    testSortByValue
    testAllMergeIsRight

    testPerf(merge)
    testPerf(mergeInplace)
    // testPerf(mergeFunctional)
  }

  def testSortByValue():Unit = {
    val map = Map("shuqin" -> 31, "yanni" -> 28)
    sortByValue(map).foreach { println }
  }

  def testAllMergeIsRight(): Unit = {
    testMerge(merge)
    testMerge(mergeFunctional)
    testMerge(mergeInplace)

    testMergeKOrderedList(mergeKOrderedList)
    testMergeKOrderedList(mergeKOrderedListFunctional)
    testMergeKOrderedList(mergeKOrderedListIneffective)
  }

  def testMerge(merge: (List[Int], List[Int]) => List[Int]):Unit = {
    assert(merge(Nil, Nil) == List())
    assert(merge(List(), Nil) == List())
    assert(merge(List(), List()) == List())
    assert(merge(List(), List(1,3)) == List(1,3))
    assert(merge(List(4,2), List()) == List(4,2))
    assert(merge(List(4,2), Nil) == List(4,2))
    assert(merge(List(2,4), List(1,3)) == List(1,2,3,4))
    assert(merge(List(2,4), List(1,3,5)) == List(1,2,3,4,5))
    assert(merge(List(2,4,6), List(1,3)) == List(1,2,3,4,6))
    assert(merge(List(2,4,6), List(8,10)) == List(2,4,6,8,10))
    println("test merge list passed.")
  }

  def testMergeKOrderedList(mergeKOrderedList: List[List[Int]] => List[Int]):Unit = {
    assert(mergeKOrderedList(Nil) == List())
    assert(mergeKOrderedList(List()) == List())
    assert(mergeKOrderedList(List(List())) == List())
    assert(mergeKOrderedList(List(List(1,2))) == List(1,2))
    assert(mergeKOrderedList(List(List(), List())) == List())
    assert(mergeKOrderedList(List(List(), List(1,3))) == List(1,3))
    assert(mergeKOrderedList(List(List(2,4), List())) == List(2,4))
    assert(mergeKOrderedList(List(List(2,4), List(1,3))) == List(1,2,3,4))
    assert(mergeKOrderedList(List(List(2,4), List(1,3,5))) == List(1,2,3,4,5))
    assert(mergeKOrderedList(List(List(2,4,6), List(1,3))) == List(1,2,3,4,6))
    assert(mergeKOrderedList(List(List(2,4,7), List(1,6), List(3,5))) == List(1,2,3,4,5,6,7))
    assert(mergeKOrderedList(List(List(2,4,9), List(1,7), List(3,6), List(5,8))) == List(1,2,3,4,5,6,7,8,9))
    println("test mergeKOrderedList passed.")
  }

  def testPerf(merge: (List[Int], List[Int]) => List[Int]):Unit = {
    val n = 10
    val numbers = (1 to 7).map(pow(n,_).intValue)
    println(numbers)
    numbers.foreach {
      num =>
        val methodName = merge.toString()
        val start = System.currentTimeMillis
        val xList = (1 to num).filter(_ % 2 == 0).toList
        val yList = (1 to num).filter(_ % 2 == 1).toList
        val merged = merge(xList, yList)
        val mergedSize = merged.size
        val end = System.currentTimeMillis
        val cost = end - start
        println(s"method=${methodName}, numbers=${num}, merged size: ${mergedSize}, merge cost: ${cost} ms")
    }
  }

  /**
    * 对指定 Map 按值排序
    */
  def sortByValue(m: Map[String,Int]): Map[String,Int] = {
    val sortedm = new mutable.LinkedHashMap[String,Int]
    m.toList.sortWith{case(kv1,kv2) => kv1._2 > kv2._2}.foreach { t =>
      sortedm(t._1) = t._2
                                                                }
    return sortedm
  }

  /**
    * 合并两个有序列表
    * 将 yList 合并到 xList 上
    * 结合了 mergeFunctional 和 mergeIneffective 的优势
    * 没有空间开销，时间复杂度为 O(n+m), n,m 分别是 xList, yList 的列表长度
    * xList and yList should both be ListBuffer , and return ListBuffer
    *
    * TODO not implemented
    */
  def mergeInplace(xList: List[Int], yList: List[Int]): List[Int] = {
    (xList, yList) match {
      case (Nil, Nil) => List[Int]()
      case (Nil, _) => yList
      case (_, Nil) => xList
      case (hx :: xtail, hy :: ytail) =>
        var result = List[Int]()
        var xListP = List[Int]()
        var yListP = List[Int]()
        if (hx > hy) {
          result = hy :: Nil
          xListP = xList
          yListP = ytail
        }
        else {
          result = hx:: Nil
          yListP = yList
          xListP = xtail
        }
        while (xListP != Nil && yListP != Nil) {
          if (xListP.head > yListP.head) {
            result = result :+ yListP.head
            yListP = yListP.tail
          }
          else {
            result = result :+ xListP.head
            xListP = xListP.tail
          }
        }
        if (xListP == Nil) {
          result = result ::: yListP
        }
        if (yListP == Nil) {
          result = result ::: xListP
        }
        // println("xsize=" + xList.size + ", ysize= " + yList.size + ", merged=" + result.size)
        result
    }
  }

  /**
    * 合并两个有序列表
    * 
    * 由于每次插入 yList 元素到 xList 都要从头遍历，因此算法时间复杂度是 O(n*m)
    */
  def mergeFunctional(xList: List[Int], yList: List[Int]): List[Int] = {
    (xList, yList) match {
      case (Nil, Nil) => List[Int]()
      case (Nil, _) => yList
      case (_, Nil) => xList
      case (hx :: xtail, hy :: ytail) =>
        yList.foldLeft(xList)(insert)
    }
  }

  def insert(xList:List[Int], y:Int): List[Int] = {
    (xList, y) match {
      case (Nil, _) => y :: Nil
      case (hx :: xtail, _) =>
        if (hx > y) {
          y :: xList
        }
        else {
          var result = hx :: Nil
          var pCurr = xtail
          while (pCurr != Nil && pCurr.head < y) {
            result = result :+ pCurr.head
            pCurr = pCurr.tail
          }
          (result :+ y) ::: pCurr
        }
    }
  }

  /**
    * 合并两个有序列表
    * 将 yList 与 xList 合并到一个全新的链表上
    * 由于使用指针是渐进地合并，因此算法时间复杂度是 O(n+m) n,m 分别是 xList, yList 的列表长度
    * 由于有列表复制操作，且是渐进地合并，因此算法空间复杂度也是 O(n+m)
    */
  def merge(xList: List[Int], yList: List[Int]): List[Int] = {
    if (xList.isEmpty) {
      return yList
    }
    if (yList.isEmpty) {
      return xList
    }
    val result = ListBuffer[Int]()
    var xListC = xList
    var yListC = yList
    while (!xListC.isEmpty && !yListC.isEmpty ) {
      if (xListC.head < yListC.head) {
        result.append(xListC.head)
        xListC = xListC.tail
      }
      else {
        result.append(yListC.head)
        yListC = yListC.tail
      }
    }
    if (xListC.isEmpty) {
      result.appendAll(yListC)
    }
    if (yListC.isEmpty) {
      result.appendAll(xListC)
    }

    result.toList
  }

  /**
    * 合并k个有序列表
    * 转化为并行容器进行并行地合并，有空间开销
    */
  def mergeKOrderedList(klists: List[List[Int]]): List[Int] = {
    if (klists.isEmpty) { return List[Int]() }
    if (klists.size == 1) { return klists.head }
    klists.par.reduce(merge)
  }

  /**
    * 合并k个有序列表
    * 使用函数式逐个地合并
    */
  def mergeKOrderedListFunctional(klists: List[List[Int]]): List[Int] = {
    if (klists.isEmpty) { return List[Int]() }
    if (klists.size == 1) { return klists.head }
    klists.reduce(merge)
  }

  /**
    * 合并k个有序列表
    * 使用插入逐个地合并
    */
  def mergeKOrderedListIneffective(klists: List[List[Int]]): List[Int] = {
    if (klists.isEmpty) {
      return List[Int]()
    }
    var nlist = klists.size
    if (nlist == 1) {
      return klists.head
    }
    var klistp = klists;
    val kbuf = ListBuffer[List[Int]]()
    while (nlist > 1) {
      for (i <- 0 to nlist/2-1) {
        kbuf.insert(i, merge(klistp(2*i), klistp(2*i+1)))
        if (nlist%2 == 1) {
          kbuf.append(klistp(nlist-1))
        }
      }
      nlist = nlist - nlist/2
      klistp = kbuf.toList
    }

    kbuf.toList.head
  }

}

排序后检测

     排序后检测，既可以做成一个 Actor ，也可以做成一个 trait. 如果排序检测本身在整个任务协作中占有一席之地，那么做成Actor比较合适；如果只是一个配合性的动作，那么做成 trait 会更直接。这里选择作为一个trait, 而 BigFileSortActor 通过 with SortChecker 来借用它的排序检测能力。 

     这里提供了 checkSort 的过程式实现和函数式实现，读者可体会其中的差异。由于迭代器迭代一次后就变成空，因此迭代过程中要记录迭代次数，来与指定的整数数目进行比较断言。

package scalastudy.concurrent.billionsort

import scala.io.Source

/**
  * Created by shuqin on 17/4/25.
  */
trait SortChecker {

  /**
    * 每次比较列表的两个数, 后一个不小于前一个
    * NOTE: 使用迭代器模式
    */
  def checkSorted(filename:String, numbers:Int): Unit = {
    val numIterator = Source.fromFile(filename + ".sorted.txt").getLines().map(line => Integer.parseInt(line.trim))
    checkSort(numIterator, numbers)
    println("test sorted passed.")
  }

  /**
    * 函数式实现
    */
  def checkSort(numIterator: Iterator[Int], numbers:Int):Unit = {
    var count = 1
    numIterator.reduceLeft((prev,next) => {
      assert(prev <= next); count += 1 ; next;
    } )
    assert(count == numbers)
  }

  /**
    * 过程式实现
    */
  def checkSortProcedural(numIterator: Iterator[Int], numbers:Int): Unit = {
    var last = 0
    var count = 0
    numIterator.foreach {
      num =>
        assert(num >= last)
        last = num
        count += 1
    }
    assert(count == numbers)
  }

  def checkSort(numList: List[Int], numbers:Int): Unit = {
    checkSort(numList.iterator, numbers)
  }

}

辅助类

      (1)  Constants 包含了本示例所需要的常量，便于性能调优。 

      (2)  从 N 个数中选出不重复的 M 个数参见 RandomSelector 的实现。 算法出处：《编程珠玑》第十二章 取样问题。

package scalastudy.concurrent.billionsort

import scalastudy.utils.PathConstants

/**
  * Created by shuqin on 17/4/27.
  */
object Constants {

  // 生成的整数中不超过的最大数
  val rangeMaxNumber = 1000000000

  // 在 [0, rangeMaxNumber] 生成 numbers 个不重复的整数
  val numbers = 10000000

  // 每次生成不超过 threshold 个不重复的整数数组;　
  // 该值不能过小, 否则会因递归层次过深导致内存不足.
  val threshold = numbers / 10

  // 存储生成的不重复整数
  val filename = PathConstants.projPath + s"/data/${numbers}.txt"

  // ForkJoin 池终止前的等待时间
  val poolWaitSecs = 15

  // Debug 选项
  val debug = false

}

package zzz.study.algorithm.select;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;

public class RandomSelector {
    
    private RandomSelector() {
        
    }
    
    private static Random rand = new Random(47);
    
    /**
     * bigRandInt: 返回一个非常大的随机整数，该整数的二进制位数不小于 bits
     */
    public static int bigRandInt(int bits)
    { 
         if (bits >= 32 || bits <= 0) {
             throw new IllegalArgumentException("参数 " + bits + " 错误，必须为小于 32 的正整数！");
         }
         int baseNum = 1 << (bits - 1);
         return rand.nextInt(Integer.MAX_VALUE - baseNum) + baseNum;
    }
    
    /**
     * randRange: 生成给定范围的随机整数
     * @param low  范围下限
     * @param high 范围上限(不包含)
     * @return 给定范围的随机整数
     */
    public static int randRange(int low, int high)
    {
        if (high <= low) {
            throw new IllegalArgumentException("参数 [" + low + "," + high + "] 错误，第一个参数必须小于第二个参数！");
        }
        return bigRandInt(30) % (high-low) + low;    
    }
    
    /**
     * selectMOrderedRandInts : 从指定集合中随机选择指定数目的整数，并以有序输出
     * @param m 需要选取的整数数目
     * @param n 指定整数集合 [0:n-1]
     * @return 随机选取的有序整数列表
     */
    public static int[] selectMOrderedRandInts(int m, int n)
    {
        checkParams(m, n);
        int[] result = new int[m];
        int remaining = n;
        int selector = m;    
        for (int k=0, i=0; i < n; i++) {
            if ((bigRandInt(30) % remaining) < selector) {
                result[k++] = i;
                selector--;
            }
            remaining--;    
        }
        return result;    
    }
    
    /**
     * selectMOrderedRandInts2 : 从指定集合中随机选择指定数目的整数，并以有序输出
     * @param m 需要选取的整数数目
     * @param n 指定整数集合 [0:n-1]
     * @return 随机选取的有序整数列表
     */
    public static int[] selectMOrderedRandInts2(int m, int n)
    {
        checkParams(m, n);
        Set<Integer> holder = new TreeSet<Integer>();
        while (holder.size() < m) {
            holder.add(bigRandInt(30) % n);
        }
        return collectionToArray(holder);
    }
    
    /**
     * selectMOrderedRandInts3 : 从指定集合中随机选择指定数目的整数，并以有序输出
     * @param m 需要选取的整数数目
     * @param n 指定整数集合 [0:n-1]
     * @return 随机选取的有序整数列表
     */
    public static int[] selectMOrderedRandInts3(int m, int n)
    {
        checkParams(m, n);
        int[] arr = selectMDisorderedRandInts3(m, n);
        Arrays.sort(arr);
        return arr;
    }
    
    /**
     * selectMDisorderedRandInts2: 从指定整数集合中随机选择指定数目的整数，并以无序输出
     * @param m 需要选取的整数数目
     * @param n 指定整数集合 [0:n-1]
     * @return 随机选取的无序整数列表
     */
    public static int[] selectMDisorderedRandInts2(int m, int n)
    {
        checkParams(m, n);
        Set<Integer> intSet = new HashSet<Integer>();
        while (intSet.size() < m) {
            intSet.add(bigRandInt(30) % n);
        }
        return collectionToArray(intSet);
    }
    
    /**
     * selectMDisorderedRandInts3: 从指定整数集合中随机选择指定数目的整数，并以无序输出
     * @param m 需要选取的整数数目
     * @param n 指定整数集合 [0:n-1]
     * @return 随机选取的无序整数列表
     */
    public static int[] selectMDisorderedRandInts3(int m, int n)
    {
        checkParams(m, n);
        int[] arr = new int[n];
        for (int i=0; i < n; i++) {
            arr[i] = i;
        }
        for (int k=0; k < m; k++) {
            int j = randRange(k, n);
            int tmp = arr[k];
            arr[k] = arr[j];
            arr[j] = tmp;
        }
        return Arrays.copyOf(arr, m);
    }
    
    public static void checkParams(int m, int n)
    {
        if (m > n || m <= 0 || n <= 0 ) {
            throw new IllegalArgumentException("参数 [" + m + "," + n + "] 错误，必须均为正整数，且第一个参数必须小于或等于第二个参数！");
        }
    }
    
    /**
     * collectionToArray : 将指定整数集合转化为整型数组列表
     * @param collection 指定整数集合
     * @return 要返回的整型数组列表，若给定集合为空，则返回 null
     */
    public static int[] collectionToArray(Collection<Integer> collection)
    {
        if (collection == null || collection.size() == 0) {
            return null;
        }
        int[] result = new int[collection.size()];
        int k = 0;
        for (Integer integer : collection) {
            result[k] = integer;
            k++;
        }
        return result;
    }
    
    /**
     * printArray: 打印数组的便利方法，每打印十个数换行 
     * @param arr 指定要打印的数组
     */
    public static void printArray(int[] arr)
    {
        for (int i=0; i < arr.length; i++) {
            System.out.printf("%d%c", arr[i], i%10==9 ? '
' : ' ');
        }
    }
    
}　　　　　

     

小结

      本来只是想写一个 ForkJoin 的示例，但写着写着就加入了 akka, future 的元素， 是在解决问题的过程中逐渐引入的。我觉得这种学习的方式很好，就是在解决一个问题的过程中，可以综合地探索和学习到很多不同的东西。传统的学习讲究"循序渐进"的方式，但是"跳跃式＋快速试错"也许是学习新技术的更好的方法。 :)

      对于Scala并发异步编程，可以总结如下：

      (1)  ForkJoin 非常适合于数据并发或数据并行的计算，在分布式计算架构之上就演变成 Map-Reduce 计算模型了；

      (2)  Akka-Actor 并发模型非常适合于任务协作和通信的并发任务。多线程与锁同步机制的问题就在于，线程之间没有通信的通道，只好通过在内存区域开辟若干共享可变的状态来协调线程之间的协作； 而 Actor 模型则为代表任务的Actor之间的通信和协作通过了消息传递机制。正应了那句话：通过通信来共享内存，而不是通过内存来共享通信。

      (3)  Scala Future API 提供了一个全异步的框架。不像 Java 那样只能生成一个 Future 随后取数据， Scala Future 可以通过各种计算操作映射成各种各样的 Future, 而且可以级联、组合这些 Future 得到新的 Future,  然后才从转换后的最终 Future 中获取结果，并且提供了非阻塞的处理结果的方式， 是灵活、可扩展的异步编程框架。

      (4)  在执行某些容器的大量独立操作时，可以采用并行计算。Scala 提供了并行容器的实现以及简便的串行容器转并行容器的方法，充分利用多核的能力做并行计算。