主页 / coding / 问题 / 79408756
Accepted
Kris Rice
Asked: 2025-02-03 20:04:31 +0800 CST

如何正确（且安全地）管理 akka actor 中的对象序列？

  • 772

背景

我有一个Actor名为akka 的程序Client，它管理http与https服务器的连接。客户端有许多功能，包括ping服务和tokenFetcher服务。

代表client一个服务器与另一个服务器之间的“连接”。其设计目的是允许一个服务器server与另一个服务器聊天。

其过程client如下：

  1. 定期检查ping对方server是否在线
  2. 如果对方server在线，则执行auth并获取令牌
  3. 如果令牌有效，则清除向我们请求的所有呼叫

我正在努力解决的是第 3 步。我想知道如何在线程 (actor) 之间安全地实现这一点。

我尝试过的：

我正在使用Seq客户端存储的消息，如下所示：


case class SendApiCall(apiCall: ApiCall, route: String, var sent: Boolean = false)

class Client(server: Server) extends Actor {
    private var apiCalls: Seq[SendApiCall] = Seq.empty[SendApiCall]

    ...

    override def receive: Receive = {
        case sendApiCall@SendApiCall(_, _, _) =>
            if (server.onlineStatus == OFFLINE) {
                apiCalls = apiCalls.appended(sendApiCall)
            }
            else {
                sendApiCall(sendApiCall)
            }
        
        case ServerOnline() => // <- this is send to us from the ping service when it first detects the server is online
            
            apiCalls.iterator.foreach( apiCallRequest =>
                if (!apiCallRequest.sent) {
                    sendApiCall(apiCallRequest)
                    apiCallRequest.sent = true
                }
                apiCallRequest
            )

            apiCalls = apiCalls.filterNot(apiCallRequest => apiCallRequest.sent)
    }
}

但是，我相信apiCalls这是mutable这种情况的状态吗？我想知道：

  1. 这个线程安全吗？
  2. 如果不是线程安全的，我该如何让它变得安全？
multithreading

2 个回答

  1. Best Answer

    假设您的代码在演员内部运行，并且您似乎遵循“规则”，那么对我来说它看起来是线程安全的。

    Actor 的工作方式类似于“邮箱”，它可以从多个来源接收消息（可能同时接收），但只会按顺序receive将消息发送到 Actor 的消息中。因此，尽管函数可能并不总是在同一个线程上运行，但您可以有效地将 Actor 视为单线程。只要您不做任何事情让函数中的代码与自身同时运行，Actor/邮箱模型的顺序性就会使其变得安全。receivereceive

    为了说明非线程安全的东西会损害你的线程安全，考虑一个你可以调用它来通过回调触发一些异步操作的函数；如果你的回调能够直接与你的参与者的内部状态交互，那么这将是非线程安全的：

    def exampleAsyncFunction: Future[Int] = ???

class MyActor extends Actor {

  private var myInternalState: Int = 0

  override def receive = {
    case SomeMessage =>
      // BAD!
      exampleAsyncFunction.onComplete { i =>
        // here, you've captured a reference to this actor, and
        // are directly manipulating its internal state, possibly
        // from outside the sequential context of the `receive` function
        myInternalState += i
      }

    case SomeOtherMessage =>
      // OK
      myInternalState += 1
  }
}

    上述示例不是线程安全的，因为 触发的异步回调函数SomeMessage可能会与 的接收逻辑同时执行SomeOtherMessage，从而导致对 的修改发生冲突myInternalState。

    使上述示例安全的一种潜在方法是让异步回调与参与者的邮箱交互，而不是直接与其内部状态交互：

    def receive = {

  case SomeMessage => 
    val me = self // capture `self` before going into async context

    exampleAsyncFunction.onComplete { i =>
      // BETTER - sending messages to an actor is safe
      me ! IncrementCounter(i)
    }

  case IncrementCounter(i) =>
    // safe to modify internal state because it is a direct response
    // to a received message, as part of the receive loop
    myInternalState += i

  ...
}

    在您在帖子中给出的示例中，看起来您并没有做任何导致对内部状态进行并发修改的事情，因此它是线程安全的。

    • 4

  2. Akka Actor 就其内部状态而言本质上是线程安全的：该receive方法不能同时在同一个 Actor 实例上调用。因此，您的方法从线程安全的角度来看是可行的。

    • 0

相关问题