Akka Actor内创建自定义线程的问题及正确处理方式

问题背景 #

运行基于Akka的应用时，远程服务不可用场景下出现致命错误：

[ERROR] [2024-01-02 07:30:41.524] [xxxx.dispatcher-22680] Uncaught error from thread [xxxx.dispatcher-22680]: unable to create new native thread, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for ActorSystem[....]
java.lang.OutOfMemoryError: unable to create new native thread
    at java.lang.Thread.start0(Native Method)
    at java.lang.Thread.start(Thread.java:719)
    ....

排查后发现错误根源是在Actor内部手动创建并启动自定义线程：

val refreshThread = new Thread() {....}
refreshThread.start()

针对该场景，核心问题解答如下：

1. 此类自定义线程如何被Akka管理？ #

Akka完全不会管理这些手动创建的线程。Akka的Dispatcher仅负责调度Actor的消息处理逻辑，所有Actor的执行都在Dispatcher管控的线程池中完成。而你手动创建的线程脱离了Akka的线程模型，属于独立的JVM线程，Akka无法追踪其状态、调度其执行，也无法在Actor停止时自动终止这些线程，完全由JVM负责生命周期管理。

2. 该做法是否符合Akka的规范？ #

完全不符合Akka的设计规范。Akka的核心设计哲学之一就是屏蔽底层线程管理，让开发者通过其提供的异步、调度API来处理并发任务，手动创建线程会带来一系列严重问题：

• 线程资源耗尽：如你遇到的OutOfMemoryError，远程服务不可用可能导致Actor反复创建新线程，最终耗尽系统允许的最大线程数。
• 线程泄漏：Actor停止后，自定义线程可能继续运行，占用系统资源。
• 线程安全风险：Actor的状态设计为单线程访问，自定义线程可能并发修改Actor状态，引发数据不一致。
• 破坏Akka的监控和调优：Akka无法监控这些线程的运行状态，无法通过Dispatcher配置进行资源管控。

3. 该场景下的正确处理方式是什么？ #

根据业务场景，选择对应的Akka原生API替代手动线程：

• 定时/周期性任务：使用Akka Scheduler，它会利用Akka的Dispatcher调度任务，且能和Actor生命周期绑定（Actor停止时自动取消任务）：

  import scala.concurrent.duration._
  import context.dispatcher
  
  // 示例：每隔10秒执行一次任务
  val cancellable = context.system.scheduler.scheduleAtFixedRate(
    initialDelay = 0.seconds,
    interval = 10.seconds
  ) { () =>
    // 你的任务逻辑
  }
  
  // 可在Actor停止时手动取消（也可依赖Actor生命周期自动取消）
  override def postStop(): Unit = {
    cancellable.cancel()
  }
  

• 异步计算任务：使用Akka Future配合Dispatcher，让任务在Akka管控的线程池中执行：

  import scala.concurrent.Future
  import context.dispatcher
  
  val asyncTask = Future {
    // 你的异步逻辑
  }
  
  // 可通过onComplete处理任务结果
  asyncTask.onComplete { result =>
    // 处理结果
  }
  

• 调用阻塞API：如果必须执行阻塞操作，使用Akka专门配置的阻塞Dispatcher（单独线程池，限制线程数），或用blocking包装代码告知Akka调整线程池负载：

  import scala.concurrent.blocking
  
  Future {
    blocking {
      // 阻塞API调用
    }
  }(context.system.dispatchers.lookup("blocking-dispatcher"))
  

内容的提问来源于stack exchange，提问作者Mandroid