Início / coding / Perguntas / 79408756
Accepted
Kris Rice
Asked: 2025-02-03 20:04:31 +0800 CST

Como gerenciar corretamente (e com segurança) uma sequência de objetos em um ator akka?

  • 772

Fundo

Tenho um akka Actorchamado Clientque gerencia httpe httpsconecta a um servidor. O cliente tem muitos recursos, incluindo um pingserviço e um tokenFetcherserviço.

O clientrepresenta uma 'conexão' entre um servidor e outro. Ele é projetado para permitir que um serverconverse com outro.

O processo clienté o seguinte:

  1. Periodicamente pingo outro serverpara ver se está online
  2. Se o outro serverestiver online, faça authe ganhe um token
  3. Se o token for válido, limpe todas as chamadas que foram solicitadas de nós

É com o passo 3 que estou tendo dificuldades. Gostaria de saber como implementaria isso com segurança entre threads (atores).

O que eu tentei:

Estou usando uma série Seqde mensagens que o cliente armazenaria, como estas:


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)
    }
}

No entanto, acredito apiCallsque é um mutableestado neste cenário? Gostaria de saber:

  1. Este tópico é seguro?
  2. Como eu tornaria isso seguro para threads, se não for?
multithreading

2 respostas

  1. Best Answer

    Considerando que seu código está sendo executado dentro de um ator e você parece estar seguindo "as regras", ele parece seguro para threads para mim.

    Os atores trabalham em termos de uma "caixa de correio", que pode receber mensagens de muitas fontes, possivelmente simultaneamente, mas só despachará as mensagens para a receivemensagem do seu ator sequencialmente . Então, embora a receivefunção nem sempre possa ser executada no mesmo thread, você pode efetivamente considerar seu ator como single-threaded. Contanto que você não faça nada para fazer o código dentro da sua receivefunção ser executado simultaneamente com ele mesmo, a sequencialidade do modelo ator/caixa de correio o torna seguro.

    Para ilustrar algo não seguro para threads que comprometeria sua segurança de threads, considere uma função que você poderia chamar para acionar alguma ação assíncrona com um retorno de chamada; se seu retorno de chamada fosse capaz de interagir diretamente com o estado interno do seu ator, isso não seria seguro para threads:

    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
  }
}

    O exemplo acima não seria seguro para threads, porque a função de retorno de chamada assíncrona acionada por SomeMessagepoderia ser executada simultaneamente com a lógica de recebimento para SomeOtherMessage, causando modificações conflitantes em myInternalState.

    Uma maneira possível de tornar o exemplo acima seguro é fazer com que o retorno de chamada assíncrono interaja com a caixa de correio do ator em vez de diretamente com seu estado interno:

    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

  ...
}

    No exemplo que você deu em sua postagem, não parece que você está fazendo nada para causar modificações simultâneas em seu estado interno, então ele é seguro para threads.

    • 4

  2. Os atores Akka são inerentemente thread-safe com relação ao seu estado interno: o receivemétodo não pode ser invocado na mesma instância de ator simultaneamente. Como tal, sua abordagem funciona de uma perspectiva de thread-safety.

    • 0

relate perguntas