Twitter Scala School 项目解析：Scala并发编程指南

引言

并发编程是现代软件开发中不可或缺的重要技能。Scala作为一门运行在JVM上的多范式编程语言，继承了Java强大的并发模型，同时提供了更高级的抽象和更优雅的语法。本文将深入探讨Scala中的并发编程概念和实践。

基础概念

Runnable与Callable接口

在Java并发模型中，Runnable和Callable是两个基础接口：

trait Runnable {
  def run(): Unit
}

trait Callable[V] {
  def call(): V
}

关键区别在于：

• Runnable不返回任何值
• Callable可以返回计算结果

线程基础

Scala的并发模型构建在Java并发模型之上。在Sun JVM上，对于IO密集型工作负载，单机可以运行数万个线程。

创建线程的基本方式：

val hello = new Thread(new Runnable {
  def run() {
    println("hello world")
  }
})
hello.start

线程池与Executor框架

直接使用线程存在资源管理问题，Java 5引入了ExecutorService抽象，提供了更高级的线程管理能力。

固定大小线程池示例

import java.util.concurrent.{Executors, ExecutorService}

class NetworkService(port: Int, poolSize: Int) extends Runnable {
  val pool: ExecutorService = Executors.newFixedThreadPool(poolSize)
  
  def run() {
    try {
      while (true) {
        val socket = serverSocket.accept()
        pool.execute(new Handler(socket))
      }
    } finally {
      pool.shutdown()
    }
  }
}

这种模式下，线程会被复用，提高了资源利用率。

Future异步编程

Future代表一个异步计算的结果。当需要结果时，可以调用阻塞的Await.result()方法获取。

val future = new FutureTask[String](new Callable[String]() {
  def call(): String = {
    searcher.search(target)
}})
executor.execute(future)

val blockingResult = Await.result(future)

线程安全问题与解决方案

典型问题示例

class Person(var name: String) {
  def set(changedName: String) {
    name = changedName
  }
}

在多线程环境下，两个线程同时调用set方法会导致不可预测的结果。

解决方案

1. 同步块(synchronized)

   def set(changedName: String) {
     this.synchronized {
       name = changedName
     }
   }
   

2. volatile关键字

   class Person(@volatile var name: String)
   

3. 原子引用(AtomicReference)

   import java.util.concurrent.atomic.AtomicReference
   class Person(val name: AtomicReference[String])
   

性能比较：

• synchronized通常性能最好，因为JVM对其有优化
• volatile每次访问都会同步
• AtomicReference需要方法调用开销

实战：构建线程安全的搜索引擎

非线程安全版本

class InvertedIndex(val userMap: mutable.Map[String, User]) {
  def add(user: User) {
    tokenizeName(user.name).foreach { term =>
      userMap += term -> user
    }
  }
}

改进方案

1. 细粒度同步

   def add(user: User) {
     val tokens = tokenizeName(user.name)
     tokens.foreach { term =>
       userMap.synchronized {
         add(term, user)
       }
     }
   }
   

2. 使用SynchronizedMap特质

   class SynchronizedInvertedIndex extends InvertedIndex(
     new mutable.HashMap[String, User] with SynchronizedMap[String, User]
   )
   

3. 使用ConcurrentHashMap

   class ConcurrentInvertedIndex extends InvertedIndex(
     new ConcurrentHashMap[String, User] asScala
   )
   

生产者-消费者模式实践

高效处理文件索引的经典模式：

class Producer[T](path: String, queue: BlockingQueue[T]) extends Runnable {
  def run() {
    Source.fromFile(path).getLines.foreach { line =>
      queue.put(line)
    }
  }
}

abstract class Consumer[T](queue: BlockingQueue[T]) extends Runnable {
  def run() {
    while (true) {
      consume(queue.take())
    }
  }
}

// 初始化
val queue = new LinkedBlockingQueue[String]()
val pool = Executors.newFixedThreadPool(8)

// 启动生产者
new Thread(new Producer("users.txt", queue)).start()

// 启动消费者
for (_ <- 1 to 8) {
  pool.submit(new IndexerConsumer(index, queue))
}

总结

Scala提供了丰富的并发编程工具，从基础的线程操作到高级的Future抽象，再到各种线程安全集合。理解这些工具的特性和适用场景，可以帮助开发者构建高效、可靠的并发系统。在实际开发中，应根据具体场景选择合适的并发策略，平衡性能与正确性的需求。