如何在Akka Typed中使用PreRestart实现精准消息异常重启？

Alright, I’ve dealt with this exact scenario before when migrating from untyped Akka to Akka Typed—let’s walk through how to replicate that familiar case (reason: Throwable, message: Option[Any]) restart logic you’re used to.

First, let’s recall: in untyped Actors, the preRestart method gives you direct access to the failure reason and the problematic message that triggered the restart. In Akka Typed, this maps to handling the PreRestart signal, but there’s one catch: the signal itself doesn’t automatically carry the failed message. We’ll fix that by tracking the current message in the Actor’s state.

Step 1: Define Your Message Type & Actor State #

Start by defining your Actor’s message hierarchy, plus a simple state to track the message being processed (so we can tie it to the restart later):

import akka.actor.typed._
import akka.actor.typed.scaladsl._

// Define your Actor's message protocol
sealed trait WorkerMessage
case class ProcessTask(taskId: String) extends WorkerMessage
case object Shutdown extends WorkerMessage

// Internal state to track the current message being processed
private case class ActorState(currentTask: Option[WorkerMessage])

Step 2: Wrap Behavior with Supervisor Strategy #

Use Behaviors.supervise to attach a restart strategy to your core Actor behavior. This tells Akka when to trigger a restart (e.g., on any RuntimeException):

object RestartingWorker {
  def apply(): Behavior[WorkerMessage] = Behaviors.setup { context =>
    // Wrap the running behavior with a restart-on-failure strategy
    Behaviors.supervise(running(ActorState(None)))
      .onFailure(SupervisorStrategy.restart) // Restart on any failure
  }

Step 3: Core Behavior with Message Tracking #

Implement the core behavior that updates the state with the current message before processing it. If processing fails, the state will already hold the problematic message for the restart handler:

private def running(state: ActorState): Behavior[WorkerMessage] = Behaviors.receive[WorkerMessage] { (context, msg) =>
    // Update state to track the current message being processed
    val updatedState = state.copy(currentTask = Some(msg))
    
    msg match {
      case ProcessTask(taskId) =>
        // Simulate a failure for specific tasks
        if (taskId.contains("fail")) {
          throw new RuntimeException(s"Failed to process task: $taskId")
        } else {
          context.log.info(s"Successfully processed task: $taskId")
          // Reset state after successful processing
          running(updatedState.copy(currentTask = None))
        }
      case Shutdown =>
        context.log.info("Shutting down gracefully")
        Behaviors.stopped
    }
  }

Step 4: Handle PreRestart Signal for Cleanup & Context #

Add a receiveSignal block to catch the PreRestart signal. Here, you’ll access the failure reason and the problematic message from the Actor’s state—just like the untyped preRestart method:

.receiveSignal { (context, signal) =>
    signal match {
      case PreRestart(reason, _) =>
        // Log the failure details (replace with your cleanup logic)
        state.currentTask match {
          case Some(failedTask) =>
            context.log.error(
              "Restarting actor due to failure: {}\nFailed task: {}",
              reason.getMessage,
              failedTask
            )
            // Add your pre-restart logic here: clean up resources, notify dependencies, etc.
          case None =>
            context.log.error("Restarting actor due to failure: {}", reason.getMessage)
        }
        // Return the initial state for the restarted Actor
        running(ActorState(None))
      case _ =>
        // Ignore other signals, keep running with current state
        running(state)
    }
  }
}

Key Notes #

• Supervisor Strategy: The onFailure(SupervisorStrategy.restart) ensures Akka triggers a restart when your Actor throws an exception. You can customize this to only restart on specific exceptions (e.g., onFailure[MyCustomException](SupervisorStrategy.restart)).
• State Tracking: The ActorState is critical here—it bridges the gap between the message processing and the restart signal, since Akka Typed doesn’t pass the failed message directly in PreRestart.
• Cleanup Logic: Use the PreRestart handler to do anything you’d put in the untyped preRestart method: close connections, stop child Actors, send failure notifications, etc.

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