Scala同步机制使用方法及Akka Actor引用复用实现咨询

我来帮你解答这两个Scala/Akka相关的问题：

Scala扎根于JVM生态，所以既可以用Java成熟的并发同步工具，也有适配Scala风格的并发API，常见的同步方案有这些：

• 基础的synchronized代码块：和Java用法一致，依托对象的内置锁保证同一时间只有一个线程执行块内逻辑，适合简单的同步场景：

  class Counter {
    private var count = 0
    def increment(): Unit = this.synchronized {
      count += 1
    }
    def getCount: Int = this.synchronized {
      count
    }
  }
  

• 灵活的显式锁（ReentrantLock等）：来自java.util.concurrent.locks包，比内置锁支持更多特性，比如公平锁、可中断锁，适合复杂的同步需求：

  import java.util.concurrent.locks.ReentrantLock
  
  class Counter {
    private val lock = new ReentrantLock()
    private var count = 0
    def increment(): Unit = {
      lock.lock()
      try {
        count += 1
      } finally {
        lock.unlock() // 务必在finally中释放锁，防止异常导致锁泄漏
      }
    }
  }
  

• 无锁原子类：对于简单的数值操作，java.util.concurrent.atomic下的原子类（比如AtomicInteger）可以实现无锁同步，性能比锁机制更高：

  import java.util.concurrent.atomic.AtomicInteger
  
  class Counter {
    private val count = new AtomicInteger(0)
    def increment(): Unit = count.incrementAndGet()
    def getCount: Int = count.get()
  }
  

• Scala异步同步工具：如果是处理异步任务的协调，可以用Future和Promise，避免阻塞线程（尽量少用Await，因为它会阻塞线程）：

  import scala.concurrent.{Await, Future}
  import scala.concurrent.ExecutionContext.Implicits.global
  import scala.concurrent.duration._
  
  val task1 = Future { /* 执行异步任务1 */ }
  val task2 = Future { /* 执行异步任务2 */ }
  // 等待所有任务完成后获取结果
  val combinedResult = Await.result(Future.sequence(Seq(task1, task2)), 10.seconds)
  

看你给出的代码片段，你想用synchronized块通过actorSelection获取Actor引用，但这个实现有几个明显的问题：

1. 每次调用getTestActorRef都会执行resolveOne，即使Actor已经存在，额外开销很大；
2. 如果Actor还未创建，resolveOne会触发超时异常；
3. 使用Await.ready阻塞线程，违背了Akka异步非阻塞的设计初衷。

这里给你两个更合理的实现方案：

方案一：懒加载创建+缓存ActorRef #

直接在ActorManager中缓存已创建的ActorRef，首次调用时创建Actor，后续直接返回缓存的引用，完全避免resolve的开销：

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import scala.concurrent.duration._

object ActorManager {
  private var testActorRef: Option[ActorRef] = None
  private val timeout = 10.seconds

  // 传入已初始化的ActorSystem
  def getTestActorRef(system: ActorSystem): ActorRef = this.synchronized {
    testActorRef.getOrElse {
      // 首次调用时创建Actor并缓存
      val ref = system.actorOf(Props[TestActor], "testActor")
      testActorRef = Some(ref)
      ref
    }
  }
}

class TestActor extends Actor {
  override def receive: Receive = {
    case msg => println(s"Received message: $msg")
  }
}

方案二：异步获取并缓存（适配Actor由外部创建的场景） #

如果Actor是由其他组件创建的，不能在Manager中直接初始化，可以改用异步方式获取并缓存，避免阻塞线程：

import akka.actor.{ActorRef, ActorSystem}
import scala.concurrent.{ExecutionContext, Future}
import scala.concurrent.duration._

object ActorManager {
  private var testActorRef: Option[ActorRef] = None
  private val actorPath = "akka://testActorSystem/user/testActor"
  private val timeout = 10.seconds

  def getTestActorRef(system: ActorSystem)(implicit ec: ExecutionContext): Future[ActorRef] = this.synchronized {
    testActorRef match {
      case Some(ref) => Future.successful(ref)
      case None =>
        val selection = system.actorSelection(actorPath)
        selection.resolveOne(timeout).map { ref =>
          testActorRef = Some(ref)
          ref
        }
    }
  }
}

额外注意事项 #

• 绝对不要在Actor内部使用Await阻塞线程，这会占用Actor的消息处理线程，严重影响整个Actor系统的吞吐量；
• 如果Actor可能意外终止，可以在Manager中添加watch逻辑，当Actor终止时清空缓存，下次调用时重新创建或获取引用。

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