并发与流

Scala 并发从 Future（标准库）到 Akka/Pekko Streams、ZIO Streams、FS2（Cats Effect）层层递进。对照 /roadmaps/scala「Concurrency & Streams」节点。

学习目标

• 使用 Future 与 ExecutionContext 做异步组合
• 了解 Pekko Streams 的 Source/Flow/Sink 模型
• 知道 ZIO / FS2 流式处理的基本概念

核心概念

Future 组合

import scala.concurrent.{Future, ExecutionContext}
import scala.util.{Success, Failure}

def fetchTitle(id: Long)(using ec: ExecutionContext): Future[String] =
  Future {
    if id <= 0 then throw new IllegalArgumentException("bad id")
    else s"Title-$id"
  }

def fetchMany(ids: List[Long])(using ec: ExecutionContext): Future[List[String]] =
  Future.sequence(ids.map(fetchTitle))

given ec: ExecutionContext = ExecutionContext.global

val result = fetchMany(List(1L, 2L, 3L))
result.onComplete {
  case Success(xs) => println(xs.mkString(", "))
  case Failure(e)  => println(s"error: ${e.getMessage}")
}
Thread.sleep(500)

Future.sequence 并行多个 Future；生产环境应配置专用线程池而非 global。

Pekko Streams 概览

import org.apache.pekko.actor.typed.{ActorSystem, Behaviors}
import org.apache.pekko.stream.scaladsl.{Source, Sink, Flow}
import org.apache.pekko.stream.Materializer

given system: ActorSystem[Nothing] = ActorSystem(Behaviors.empty, "streams")
given mat: Materializer = Materializer(system)

val stream =
  Source(1 to 100)
    .filter(_ % 2 == 0)
    .map(_ * 10)
    .runWith(Sink.seq)

stream.foreach(seq => println(s"count=${seq.size}, head=${seq.headOption}"))

Source 产数据、Flow 变换、Sink 消费；背压（backpressure）由 Stream 协议处理。

FS2 / ZIO Streams（概念）

// FS2 风格片段
import fs2.Stream

val s: Stream[IO, Int] =
  Stream.emits(1 to 5).filter(_ > 2).map(_ * 2)

// ZIO Stream
import zio.stream.*

val z: ZStream[Any, Throwable, String] =
  ZStream.fromIterable(1 to 3).map(n => s"item-$n")

FS2 与 ZIO Streams 纯函数式、资源安全（bracket），适合 Kafka / 文件管道。

与 Java 对照

Scala	Java