Scala 并发编程进阶：Future、ExecutionContext 与结构化并发

要点清单
- Future.apply 立即把任务提交给 ExecutionContext，返回句柄而非配方
- map / flatMap / filter / recover / recoverWith / transform 是核心组合子
- sequence 把 List[Future[A]] 翻转为 Future[List[A]]，遇首个失败即整体失败
- traverse 等价于 map 加 sequence，但只遍历一次，更省一次分配
- zip 并行等待两个独立结果，比 flatMap 嵌套快一倍左右
- andThen 只做副作用，不改变结果，其回调抛出的异常会被忽略并上报

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.*

case class User(id: Long, name: String)

def fetchUser(id: Long): Future[User] =
  Future { Thread.sleep(50); User(id, s"user-$id") }

def fetchOrders(id: Long): Future[List[Long]] =
  Future { Thread.sleep(30); List(id * 10, id * 10 + 1) }

// zip 并行等待，总耗时约等于较慢的那个
val profile: Future[(User, List[Long])] = fetchUser(1).zip(fetchOrders(1))

// traverse 一次遍历，比 list.map(fetchUser).sequence 少一次分配
val names: Future[List[String]] =
  Future.traverse(List(1L, 2L, 3L))(fetchUser(_).map(_.name))

// recover 把失败分支收敛成同类型的成功值
val safe: Future[String] = fetchUser(42).map(_.name).recover {
  case e: RuntimeException => s"fallback: ${e.getMessage}"
}

要点清单
- global 是 ForkJoinPool，默认并行度等于 CPU 核数
- blocking { } 只是向 FJP 申请临时扩容，不改变任务本身的阻塞性质
- 多 EC 隔离故障域：CPU 密集与 IO 密集必须分开，避免互相拖垮
- MDC 不会自动跨线程传播，需要包装 Executor 手动复制上下文
- 不要在 EC 线程上用 Await.result 等另一个 Future，会死锁

import java.util.concurrent.{Executors, ThreadFactory, Executor}
import java.util.concurrent.atomic.AtomicInteger
import scala.concurrent.ExecutionContext

val counter = new AtomicInteger(0)
val factory: ThreadFactory = r => {
  val t = new Thread(r, s"app-ec-${counter.incrementAndGet()}")
  t.setDaemon(true)
  t
}

// 固定线程池：适合阻塞 IO，线程数按下游连接池与超时估算
val blockingEc: ExecutionContext =
  ExecutionContext.fromExecutor(Executors.newFixedThreadPool(32, factory))

// 虚拟线程：JDK 21 起，每任务一线程，阻塞成本极低
val virtualEc: ExecutionContext =
  ExecutionContext.fromExecutor(Executors.newVirtualThreadPerTaskExecutor())

// MDC 不会跨线程传播，包装 Executor 手动复制上下文
def mdcAware(delegate: Executor): Executor = (command: Runnable) => {
  val ctx = Option(org.slf4j.MDC.getCopyOfContextMap)
  delegate.execute { () =>
    val prev = Option(org.slf4j.MDC.getCopyOfContextMap)
    ctx.foreach(org.slf4j.MDC.setContextMap)
    try command.run() finally prev.fold(org.slf4j.MDC.clear())(org.slf4j.MDC.setContextMap)
  }
}

四大缺陷
- 急切求值：一构造就提交，无法复用、无法重试、无法只描述不执行
- 无法取消：没有取消协议，只能靠 AtomicBoolean 之类的协作式标志位自救
- 异常吞噬：失败只在 onComplete 与 recover 处可见，忘挂回调就静默丢失
- 阻塞传播：一个阻塞任务占住工作线程，容易引发线程饥饿与级联超时
- 衍生问题：全局隐式 EC 让依赖关系不可见，测试难以注入确定性调度

import java.util.concurrent.atomic.AtomicBoolean

// 缺陷一：急切求值。下面两行都会立即执行，没有延迟语义
val a = Future { expensiveCall() }
val b = Future { expensiveCall() }

// 缺陷二：协作式取消。Future 本身不支持取消，只能自查标志位
def longTask(flag: AtomicBoolean): Future[Unit] = Future {
  while (!flag.get()) { step() }
}

// 缺陷三：异常吞噬。这个 Future 的失败没有任何人观察
def fireAndForget(): Unit = Future { risky() }

// 缺陷四：阻塞传播。在 EC 线程上等待，可能直接耗尽线程池
def deadlockProne(): Unit = Await.result(Future { blockingCall() }, 5.seconds)

要点清单
- Promise 与 Future 一一对应，promise.future 拿到只读端
- tryComplete 系列返回 Boolean，表示本次是否真的写入成功
- 典型用途：超时、first-completed 竞速、回调式 API 桥接
- Promise 线程安全，可在任意线程完成，无需额外加锁
- 用 Promise 实现超时时务必取消定时器，避免线程与内存泄漏

import scala.concurrent.{Future, Promise}
import scala.concurrent.duration.*
import java.util.concurrent.{Executors, TimeUnit, TimeoutException}

private val scheduler = Executors.newSingleThreadScheduledExecutor()

// 给任意 Future 加超时：谁先完成谁生效
def withTimeout[A](f: Future[A], d: FiniteDuration)(using ec: ExecutionContext): Future[A] = {
  val p = Promise[A]()
  val task = scheduler.schedule(
    () => p.tryFailure(new TimeoutException(s"timed out after $d")),
    d.toMillis, TimeUnit.MILLISECONDS)
  f.onComplete { r => p.tryComplete(r); task.cancel(false) }
  p.future
}

要点清单
- volatile 保证可见性与有序性，不保证复合操作的原子性
- AtomicInteger 与 AtomicReference 基于 CAS，适合低争用计数
- LongAdder 在高争用计数下显著优于 AtomicLong，代价是读取弱一致
- ConcurrentHashMap 的 computeIfAbsent 是原子的，函数体内不要再操作同一张表
- synchronized 在 Scala 中写作 lock.synchronized { }，是可重入互斥锁
- 无锁不等于无竞争，CAS 自旋失败会浪费 CPU，高争用时锁反而更快

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.{AtomicLong, LongAdder}

class Metrics {
  private val requests = new LongAdder          // 高争用计数首选
  private val latency = new AtomicLong(0L)      // 需要精确读改写时用 Atomic
  private val gauges = new ConcurrentHashMap[String, Long]()

  def record(ms: Long): Unit = { requests.increment(); latency.addAndGet(ms) }

  def gauge(name: String, v: Long): Long =
    gauges.merge(name, v, (a: Long, b: Long) => a + b)
}

@volatile private var running = true
def stop(): Unit = running = false              // volatile 写，建立 happens-before
def loop(): Unit = while (running) { work() }   // volatile 读，可见性有保证

private val lock = new Object
def transfer(from: Account, to: Account, amount: Long): Unit = lock.synchronized {
  from.withdraw(amount)
  to.deposit(amount)
}

要点清单
- Scala 2.13 起必须显式依赖 scala-parallel-collections，导入 CollectionConverters
- 默认使用公共 ForkJoinPool，无法隔离，一个阻塞任务拖垮全局
- 拆分粒度由 Spliterator 决定，Vector 拆分均匀，List 拆分代价高
- 归约操作必须满足结合律，否则并行结果与串行结果不一致
- 顺序敏感的 foldLeft 在 Par 上语义会变，需用 fold 显式给结合律
- 现代替代：Future.traverse 配专用 EC，或流式库的并行算子

// build.sbt: "org.scala-lang.modules" %% "scala-parallel-collections" % "1.0.4"
import scala.collection.parallel.CollectionConverters.*

val xs = (1 to 1000000).toVector
val sum: Long = xs.par.map(_ * 2L).sum   // 结合律成立，并行安全

// 现代写法：显式控制并发上限，失败可观察，可加超时
def batchCompute(xs: Vector[Long])(using ec: ExecutionContext): Future[Long] =
  Future.traverse(xs.grouped(10000).toList)(c => Future(c.map(_ * 2L).sum)).map(_.sum)

要点清单
- boundary 定义作用域边界，break 从任意深度跳出并返回一个值
- 相比非局部 return 抛异常的实现，boundary 由编译器支持，无异常开销
- Java 21 的 StructuredTaskScope 为预览 API，Java 25 正式转正
- ShutdownOnFailure 遇首个失败即取消兄弟任务，ShutdownOnSuccess 用于竞速
- 虚拟线程让一任务一线程可行，sleep 不再昂贵，但不解决 CPU 密集
- 虚拟线程遇 synchronized 会钉住载体线程，需换成 ReentrantLock

import scala.util.boundary, boundary.break

// 用 boundary 实现提前返回，编译期支持，无异常开销
def firstPositive(xs: List[Int]): Int = boundary {
  for x <- xs do if x > 0 then break(x)
  -1
}

// 结构化并发：任一子任务失败，其余自动取消，且必须全部收敛
import java.util.concurrent.StructuredTaskScope
final case class Profile(user: User, orders: List[Long])

def loadProfile(id: Long): Profile = {
  val scope = new StructuredTaskScope.ShutdownOnFailure()
  try {
    val user   = scope.fork(() => fetchUser(id))
    val orders = scope.fork(() => fetchOrders(id))
    scope.join()            // 等全部子任务结束或首个失败
    scope.throwIfFailed()   // 有失败则抛出，兄弟任务已被取消
    Profile(user.get(), orders.get())
  } finally scope.close()   // 作用域退出时确保所有子任务终止
}

// Cats Effect：重试、超时、并发全部由 Temporal 统一表达
import cats.effect.IO
import scala.concurrent.duration.*

def fetch(id: Long): IO[String] = IO { blockingCall() }

val program: IO[String] = for {
  a <- fetch(1).timeout(2.seconds).handleError(_ => "fallback")
  b <- fetch(2).retry(3, 200.millis)
  c <- (fetch(3), fetch(4)).parTupled.map(_._1)
} yield s"$a-$b-$c"

要点清单
- 超时必须同时设置连接超时与读超时，只用 Future 层超时无法真正止损
- 重试用指数退避加抖动，避免重试风暴；只重试幂等或可安全重放的操作
- 熔断器按下游维度隔离，半开状态放少量流量探测恢复
- 指标至少覆盖进行中任务数、队列长度、失败率与 P99 延迟
- 追踪上下文要显式跨线程传播，否则异步链路会断成两截

import scala.concurrent.{Future, Promise, ExecutionContext}
import scala.concurrent.duration.*
import java.util.concurrent.{Executors, TimeUnit}
import scala.util.Random

private val scheduler = Executors.newSingleThreadScheduledExecutor()

def delayed[A](d: FiniteDuration)(body: => Future[A])(using ec: ExecutionContext): Future[A] = {
  val p = Promise[A]()
  scheduler.schedule(() => p.completeWith(body), d.toMillis, TimeUnit.MILLISECONDS)
  p.future
}

// 指数退避加抖动，最多重试 max 次，仅对可重试异常生效
def retry[A](max: Int, base: FiniteDuration, isRetryable: Throwable => Boolean)
            (op: => Future[A])(using ec: ExecutionContext): Future[A] =
  op.recoverWith {
    case e if max > 0 && isRetryable(e) =>
      delayed(base + Random.nextLong(50).millis)(retry(max - 1, base * 2, isRetryable)(op))
  }

// 测试：确定性 EC 让回调同步执行，断言不再依赖 sleep
final class DeterministicExecutionContext extends ExecutionContext {
  def execute(r: Runnable): Unit = r.run()
  def reportFailure(t: Throwable): Unit = t.printStackTrace()
}

given ExecutionContext = new DeterministicExecutionContext