Scala - Akka Actors Basics
The actor model treats actors as the fundamental units of computation. Each actor encapsulates state and behavior, communicating exclusively through asynchronous message passing. When an actor...
Key Insights
- Akka actors provide a message-passing concurrency model that eliminates shared mutable state and traditional locking mechanisms, making concurrent systems easier to reason about and more resilient to failures
- Each actor processes messages sequentially from its mailbox, guaranteeing thread-safety within the actor’s state while the actor system manages thread allocation and scheduling across potentially thousands of actors
- The actor hierarchy and supervision strategies create self-healing systems where parent actors can decide how to handle child actor failures through restart, resume, stop, or escalate directives
Understanding the Actor Model
The actor model treats actors as the fundamental units of computation. Each actor encapsulates state and behavior, communicating exclusively through asynchronous message passing. When an actor receives a message, it can create new actors, send messages to other actors, or change its internal state for processing the next message.
Akka implements this model on the JVM, providing location transparency where actors can exist locally or remotely without changing the code. The framework handles message delivery, threading, and failure recovery.
import akka.actor.{Actor, ActorSystem, Props}
class SimpleActor extends Actor {
def receive: Receive = {
case msg: String => println(s"Received: $msg")
case _ => println("Unknown message")
}
}
object ActorBasics extends App {
val system = ActorSystem("MySystem")
val simpleActor = system.actorOf(Props[SimpleActor], "simpleActor")
simpleActor ! "Hello, Actor!"
Thread.sleep(1000)
system.terminate()
}
The ! operator (pronounced “tell” or “fire-and-forget”) sends messages asynchronously. The sender doesn’t wait for a response and continues execution immediately.
Actor Lifecycle and State Management
Actors maintain private state that only they can modify. This eliminates race conditions inherent in shared mutable state. The actor’s receive method defines how it processes messages, and you can change behavior dynamically using context.become.
import akka.actor.{Actor, ActorRef, Props}
case class Deposit(amount: Double)
case class Withdraw(amount: Double)
case object GetBalance
class BankAccount extends Actor {
private var balance: Double = 0.0
def receive: Receive = {
case Deposit(amount) =>
balance += amount
println(s"Deposited $amount, new balance: $balance")
case Withdraw(amount) =>
if (balance >= amount) {
balance -= amount
sender() ! s"Withdrew $amount, remaining: $balance"
} else {
sender() ! s"Insufficient funds. Balance: $balance"
}
case GetBalance =>
sender() ! balance
}
}
// Usage
val account = system.actorOf(Props[BankAccount], "account")
account ! Deposit(100.0)
account ! Withdraw(30.0)
The sender() reference allows actors to reply to the message originator. This reference is only valid during message processing and should not be captured in closures or used asynchronously.
Request-Response Patterns with Ask
While ! provides fire-and-forget semantics, the ? operator (ask pattern) returns a Future for scenarios requiring responses.
import akka.actor.{Actor, ActorSystem, Props}
import akka.pattern.ask
import akka.util.Timeout
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Success, Failure}
case class Calculate(x: Int, y: Int)
case class Result(value: Int)
class Calculator extends Actor {
def receive: Receive = {
case Calculate(x, y) =>
sender() ! Result(x + y)
}
}
object AskPattern extends App {
implicit val system: ActorSystem = ActorSystem("AskSystem")
implicit val timeout: Timeout = Timeout(3.seconds)
val calculator = system.actorOf(Props[Calculator], "calculator")
val futureResult = calculator ? Calculate(5, 3)
futureResult.onComplete {
case Success(Result(value)) =>
println(s"Calculation result: $value")
system.terminate()
case Failure(exception) =>
println(s"Calculation failed: $exception")
system.terminate()
}
}
The ask pattern should be used sparingly as it introduces back-pressure and potential timeout handling. For high-throughput systems, prefer tell with explicit reply handling.
Actor Hierarchies and Supervision
Actors form hierarchies where each actor can create child actors. Parent actors supervise their children, defining strategies for handling failures.
import akka.actor.{Actor, ActorRef, OneForOneStrategy, Props, SupervisorStrategy}
import akka.actor.SupervisorStrategy._
import scala.concurrent.duration._
case class Process(data: String)
case object GetStats
class Worker extends Actor {
private var processed = 0
def receive: Receive = {
case Process(data) =>
if (data.isEmpty) throw new IllegalArgumentException("Empty data")
processed += 1
println(s"Worker processed: $data (total: $processed)")
case GetStats =>
sender() ! processed
}
}
class Supervisor extends Actor {
override val supervisorStrategy: SupervisorStrategy =
OneForOneStrategy(maxNrOfRetries = 3, withinTimeRange = 1.minute) {
case _: IllegalArgumentException =>
println("Restarting worker due to invalid input")
Restart
case _: Exception =>
println("Stopping worker due to unexpected error")
Stop
}
val worker: ActorRef = context.actorOf(Props[Worker], "worker")
def receive: Receive = {
case msg => worker.forward(msg)
}
}
// Usage
val supervisor = system.actorOf(Props[Supervisor], "supervisor")
supervisor ! Process("valid data")
supervisor ! Process("") // Will cause restart
supervisor ! Process("more data")
The OneForOneStrategy applies the directive only to the failed child. AllForOneStrategy applies it to all children. Supervision strategies create resilient systems where failures are isolated and handled without affecting the entire application.
Stateful Actors with Context Become
The context.become method allows actors to change their behavior dynamically, useful for implementing state machines.
import akka.actor.{Actor, Props}
sealed trait AuthMessage
case class Login(username: String, password: String) extends AuthMessage
case object Logout extends AuthMessage
case class SecureAction(action: String) extends AuthMessage
class AuthenticatedActor extends Actor {
def receive: Receive = unauthenticated
def unauthenticated: Receive = {
case Login(user, pass) if pass == "secret" =>
println(s"$user logged in")
context.become(authenticated(user))
sender() ! "Login successful"
case Login(user, _) =>
println(s"Failed login attempt for $user")
sender() ! "Login failed"
case SecureAction(_) =>
sender() ! "Please login first"
}
def authenticated(username: String): Receive = {
case SecureAction(action) =>
println(s"$username performing: $action")
sender() ! s"Action completed: $action"
case Logout =>
println(s"$username logged out")
context.become(unauthenticated)
sender() ! "Logout successful"
case Login(_, _) =>
sender() ! "Already logged in"
}
}
// Usage
val authActor = system.actorOf(Props[AuthenticatedActor], "auth")
authActor ! SecureAction("read data") // Rejected
authActor ! Login("alice", "secret") // Success
authActor ! SecureAction("read data") // Allowed
authActor ! Logout // Back to unauthenticated
This pattern eliminates conditional logic based on state flags, making the code more maintainable and the state transitions explicit.
Actor Routers for Load Distribution
Akka provides routing strategies to distribute messages across multiple actor instances for parallel processing.
import akka.actor.{Actor, Props}
import akka.routing.{RoundRobinPool, SmallestMailboxPool}
case class Work(id: Int, data: String)
class WorkerActor extends Actor {
def receive: Receive = {
case Work(id, data) =>
Thread.sleep(100) // Simulate work
println(s"${self.path.name} processed work $id: $data")
}
}
// Round-robin routing
val roundRobinRouter = system.actorOf(
RoundRobinPool(5).props(Props[WorkerActor]),
"roundRobinRouter"
)
// Smallest mailbox routing
val smartRouter = system.actorOf(
SmallestMailboxPool(5).props(Props[WorkerActor]),
"smartRouter"
)
// Send work
(1 to 20).foreach { i =>
roundRobinRouter ! Work(i, s"task-$i")
}
Routers handle the complexity of distributing work while maintaining the simple message-passing interface. The SmallestMailboxPool strategy sends messages to the actor with the fewest queued messages, providing automatic load balancing.
Production Considerations
Configure actor systems through application.conf for production deployments:
akka {
actor {
default-dispatcher {
throughput = 10
fork-join-executor {
parallelism-min = 8
parallelism-max = 64
}
}
}
actor.deployment {
/myRouter {
router = round-robin-pool
nr-of-instances = 10
}
}
}
Always use typed messages (case classes) rather than primitive types for clarity and type safety. Implement proper shutdown procedures using CoordinatedShutdown to ensure graceful termination. Monitor mailbox sizes and processing times to identify bottlenecks. Use Akka’s built-in logging instead of direct println statements for production systems.