如何在Akka Actor中安全正确地管理Api调用请求Seq？

背景 #

我有一个名为Client的Akka Actor，负责管理与服务器的HTTP/HTTPS连接，内置ping服务和tokenFetcher服务，用于实现服务器间通信。它的工作流程如下：

• 定期ping目标服务器，检测在线状态；
• 若服务器在线，执行认证操作获取token；
• 若token有效，处理所有已缓存的调用请求。
  我在实现第3步时遇到了线程安全相关问题。

已尝试的方案 #

我用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() => // ping服务首次检测到服务器在线时发送此消息
            
            apiCalls.iterator.foreach( apiCallRequest =>
                if (!apiCallRequest.sent) {
                    sendApiCall(apiCallRequest)
                    apiCallRequest.sent = true
                }
            )

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

我认为当前实现中的apiCalls属于可变状态，因此想了解两个问题：

1. 上述实现是否线程安全？
2. 若不安全，如何改造为线程安全的实现？

解答 #

1. 当前实现是否线程安全？ #

不安全，核心问题有两点：

• SendApiCall的可变字段sent：如果sendApiCall方法是异步执行（比如触发了非Actor线程的HTTP请求），其他线程可能在Actor处理ServerOnline()消息的过程中修改sent字段，导致状态不一致。即使是单线程处理，filterNot依赖的sent状态也可能被后续异步操作修改，使得过滤结果不符合预期。
• 状态修改的非原子性：遍历修改sent字段和后续过滤的操作不是原子的，若遍历过程中有新的消息加入（虽然Actor单线程处理不会出现，但结合异步回调仍有风险），会导致状态混乱。

2. 线程安全改造方案 #

遵循Akka Actor的核心设计原则：Actor内部状态仅在单线程消息处理流程中修改，杜绝共享可变状态，具体改造如下：

方案一：使用不可变消息，简化状态管理

把SendApiCall改为不可变类，去掉可变的sent字段，直接批量处理待请求后清空队列：

// 改为不可变类，移除可变字段
case class SendApiCall(apiCall: ApiCall, route: String)

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

    ...

    override def receive: Receive = {
        case sendApiCall: SendApiCall =>
            if (server.onlineStatus == OFFLINE) {
                pendingApiCalls = pendingApiCalls.appended(sendApiCall)
            }
            else {
                sendApiCall(sendApiCall)
            }
        
        case ServerOnline() =>
            // 批量发送所有待处理请求
            pendingApiCalls.foreach(sendApiCall)
            // 清空待处理队列
            pendingApiCalls = Seq.empty
    }
}

如果需要处理发送失败的场景，可在sendApiCall方法中添加异步回调，当请求失败时将消息重新发送给当前Actor，由Actor重新加入待处理队列，确保所有状态变更都在Actor单线程中完成。

方案二：封装Actor内部状态，跟踪请求生命周期

如果必须跟踪每个请求的发送状态，不要在消息对象中存储可变状态，而是在Actor内部用不可变结构封装状态：

case class SendApiCall(apiCall: ApiCall, route: String)
// Actor内部状态封装为不可变类
case class ClientState(pendingCalls: Seq[SendApiCall] = Seq.empty)

class Client(server: Server) extends Actor {
    private var state = ClientState()

    ...

    override def receive: Receive = {
        case sendApiCall: SendApiCall =>
            if (server.onlineStatus == OFFLINE) {
                state = state.copy(pendingCalls = state.pendingCalls.appended(sendApiCall))
            }
            else {
                sendWithFailureHandler(sendApiCall)
            }
        
        case ServerOnline() =>
            state.pendingCalls.foreach(sendWithFailureHandler)
            state = state.copy(pendingCalls = Seq.empty)
        
        // 处理发送失败的回调消息
        case ApiCallFailed(call) =>
            state = state.copy(pendingCalls = state.pendingCalls.appended(call))
    }

    private def sendWithFailureHandler(call: SendApiCall): Unit = {
        // 假设sendApiCall返回Future，处理失败场景
        sendApiCall(call).onFailure {
            case _ => self ! ApiCallFailed(call)
        }(context.dispatcher)
    }
}

// 自定义失败回调消息
case class ApiCallFailed(call: SendApiCall)

关键原则总结

• 禁止Actor外部线程修改Actor状态，包括消息对象中的可变字段；
• 所有状态变更必须在Actor的receive方法或其同步调用的方法中完成；
• 优先使用不可变数据结构和不可变消息，从根源避免可变状态带来的线程安全问题。

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