Akka, actors and reactive programming / Piter Publishing House Blog

Hello dear readers.

Today we wanted to talk with you about "everything new is well forgotten old" and recall the actors described by Carl Hewitt in the early 70's. And the thing is that such a book has recently been released :



It is quite voluminous - the translation should have more than 500 pages.

Despite the emphasized elitism of the book (Akka and Scala), its author Von Vernon (the largest specialist in DDD) is confident that the architectural patterns described in this work are fully implementable in .NET and C #, which is described in the application. We place a translation of the article under the cutter, the author of which admits the transfer of the actor paradigm to the Java language. Since the book’s rating on Amazon is consistently high, and the theme is universal, please share your opinions about it and about actor’s architecture in principle.

The first article in this series made an overview of Akka . Now we will delve deeply into the Akka actor sphere, armed with the akka-actor module, which lays the foundation for all other Akka modules.

In our opinion, you can learn to program, even without the practice of reading / writing code. Here we will step by step develop a small actor library: PubSub event bus, operating on the principle of "publication-subscription". Of course, Akka comes with a local and global solution of this kind ready to work, so here we just mess with a well-known example. We will work in the Scala language , simply because it is much more convenient to write Akka-like code on it, but exactly the same results can be achieved in Java.

The actor model

In the actor model — which was invented in 1973 by Carl Hewitt et al. — Actors are “fundamental units of computation that implement processing, storage, and communication.” Ok, let's figure it out in order.

The concept of "fundamental unit of computation" means that when we write a program in accordance with the actor model, our design and implementation work is built around actors. In a terrific interview given to Eric Meyer, Carl Hewitt explains that “actors are everywhere,” and that “there are no single actors, they exist in systems.” We have already summarized this idea: when using the actor model, all of our code will consist of actors.

What does an actor look like? What is finally “processing”, “storage” and “communication”? In essence, communication is asynchronous messaging , storage means that actors can have a state , and processing is that actors can deal with messages. Processing is also referred to as"Behavior" . Doesn't sound too complicated, right? So, let's take the next step and look at the Akka actors.

Akka actor device

As you can see from the following picture, Akka actor consists of several interacting components. Actorref- this is the logical address of the actor, allowing you to asynchronously send messages to the actor on the principle of "sent and forgot." A dispatcher — in this case, by default, there is one dispatcher for each actor system — is responsible for queuing messages to the actor’s mailbox, and also orders this mailbox to remove one or more messages from the queue, but only one at a time - and transfer them to the actor for processing. Last but not least, an actor — usually the only API we have to implement — encapsulates state and behavior.



As will be shown below, Akka does not allow direct access to the actor and therefore ensures that the only way to interact with the actor is asynchronous messages. Unable to call a method in an actor.
In addition, it should be noted that sending a message to an actor and processing of this message by an actor are two separate operations that most likely occur in different threads. Of course, Akka provides the necessary synchronization to ensure that any state changes are visible to all threads.

Accordingly, Akka, as it were, allows us to program the illusion of single-threaded, and we can not use any synchronization primitives like volatile or synchronized in the actor code - moreover, we should not do this.

Actor Implementation

Enough words, move on to the code! In Akka, an actor is a class to which the Actor trait mixes:

class MyActor extends Actor {
  override def receive = ???
}

The receive method returns the so-called original actor behavior. This is just a partially computable function used by Akka to process messages sent to an actor. Since the behavior is PartialFunction [Any, Unit] , it is currently not possible to define actors that only accept messages of a given type. Akka already has an akka-typed experimental module that provides type safety on this platform, but it is still being finalized. By the way, the behavior of the actor can change, and that is why the return value of the receive method is called in the original behavior.

Ok, let's implement a basic actor for our PubSub library:

class PubSubMediator extends Actor {
  override def receive = Actor.emptyBehavior
}

While we do not need PubSubMediator to process any messages, we therefore use the usual partially computable function Actor.emptyBehavior, for which no value is defined.

Actor systems and the creation of actors

As mentioned above, “there are no single actors, they exist in systems”. In Akka, the actor system is an interconnected ensemble whose members are organized hierarchically. Thus, each actor has its own parent actor, as shown in the following picture.



When creating an actor system, Akka - internally using many of the so-called "system actors" - creates three actors: this is the "root guardian" located in the root of the actor hierarchy, as well as the system and user guards. A user guard - often referred to simply as a "guard" - is the parent element for all the top-level actors we create (in this context, we mean "the highest level to which we have access").

Suppose, but how to create a system of actors? You just need to call the factory method provided by the lone object ActorSystem :

val system = ActorSystem("pub-sub-mediator-spec-system")

And why do we even create an ActorSystem ? Why not just create actors? The latter is impossible, because when the actor constructor is called directly, the system will throw an exception. Instead, we have to use the factory method provided by - you guessed it - ActorSystem to create a top-level actor:

system.actorOf(Props(new PubSubMediator), "pub-sub-mediator")

Of course, actorOf does not return an Actor instance, but an ActorRef . So Akka does not allow us to access the instance of Actor, which, in turn, guarantees: exchange of information with the actor is possible only through asynchronous messages. The name indicated by us must be unique among the siblings of this actor, otherwise an exception will be thrown. If we do not provide a name, Akka will create it for us, as each actor must have a name.

And what kind of contraption Props ? This is just a configuration object for an actor. It accepts the constructor as a parameter passed by name (i.e., lazily) and may contain other important information - for example, routing or deployment.

When it comes to remote communications, it’s important to consider that Props can be serialized, so it’s already been the practice to add a Props factory to a companion actor object. It is also convenient to set a constant corresponding to the name of the actor.

Knowing all this, let's add PubSubMediator and create a test for it using ScalaTest and Akka Testkit - another Akka module that simplifies testing of Akka actors:

object PubSubMediator {
  final val Name = "pub-sub-mediator"
  def props: Props = Props(new PubSubMediator)
}
class PubSubMediator extends Actor {
  override def receive = Actor.emptyBehavior
}
class PubSubMediatorSpec extends WordSpec with Matchers with BeforeAndAfterAll {
  implicit val system = ActorSystem("pub-sub-mediator-spec-system")
  "A PubSubMediator" should {
    "be suited for getting started" in {
      EventFilter.debug(occurrences = 1, pattern = s"started.*${classOf[PubSubMediator].getName}").intercept {
        system.actorOf(PubSubMediator.props)
      }
    }
  }
  override protected def afterAll() = {
    Await.ready(system.terminate(), Duration.Inf)
    super.afterAll()
  }
}

As you can see, we are creating an ActorSystem and an PubSubMediator actor in PubSubMediatorSpec . The test itself is a little far-fetched, because our PubSubMediator is still pretty crude. It uses life-cycle debugging and is expected to log a debug message like “started ... PubSubMediator ...” . The full code for its current version is located at GitHub under the tag step-01 .

Communication

So, having learned how to create actors, let's talk about communication, which - as mentioned above - is based on asynchronous messages and is closely related to two other properties of the actor: behavior (that is, the ability to process messages) and state.

To send a message to an actor, you need to know its address, that is, ActorRef :

mediator ! GetSubscribers("topic")

As you can see, there is an operator in ActorRef ! - the so-called “bang”, which sends the specified message to the corresponding actor. As soon as the message is delivered, the operation is completed, and the sending code continues to work. It is implied that there is no return value (except Unit ), therefore, messages really go away on the principle of "sent and forgot."

Although simple, we often need a response. Thanks to the operator ! implicitly accepts the sender as an ActorRef , this can be done without difficulty:

override def receive = {
  case Subscribe(topic) =>
    // ИМЕННО ТУТ обрабатывается подписка
    sender() ! Subscribed
}

In this example, the behavior of the recipient actor processes a specific message — the Subscribe command — and sends the message — the Subscribed event — back to the sender. Then, the sender method is used to access the sender of the message that is currently being processed.

With all this in mind, let's further refine PubSubMediator and the related test.
First, add the message protocol — the set of all messages related to PubSubMediator — to the companion object:

object PubSubMediator {
  case class Publish(topic: String, message: Any)
  case class Published(publish: Publish)
  case class Subscribe(topic: String, subscriber: ActorRef)
  case class Subscribed(subscribe: Subscribe)
  case class AlreadySubscribed(subscribe: Subscribe)
  case class Unsubscribe(topic: String, subscriber: ActorRef)
  case class Unsubscribed(unsubscribe: Unsubscribe)
  case class NotSubscribed(unsubscribe: Unsubscribe)
  case class GetSubscribers(topic: String)
  final val Name = "pub-sub-mediator"
  def props: Props = Props(new PubSubMediator)
}

Next, let's implement a behavior that has so far remained unfilled:

class PubSubMediator extends Actor {
  import PubSubMediator._
  private var subscribers = Map.empty[String, Set[ActorRef]].withDefaultValue(Set.empty)
  override def receive = {
    case publish @ Publish(topic, message) =>
      subscribers(topic).foreach(_ ! message)
      sender() ! Published(publish)
    case subscribe @ Subscribe(topic, subscriber) if subscribers(topic).contains(subscriber) =>
      sender() ! AlreadySubscribed(subscribe)
    case subscribe @ Subscribe(topic, subscriber) =>
      subscribers += topic -> (subscribers(topic) + subscriber)
      sender() ! Subscribed(subscribe)
    case unsubscribe @ Unsubscribe(topic, subscriber) if !subscribers(topic).contains(subscriber) =>
      sender() ! NotSubscribed(unsubscribe)
    case unsubscribe @ Unsubscribe(topic, subscriber) =>
      subscribers += topic -> (subscribers(topic) - subscriber)
      sender() ! Unsubscribed(unsubscribe)
    case GetSubscribers(topic) =>
      sender() ! subscribers(topic)
  }
}

As you can see, the behavior processes all commands - for example, Publish or Subscribe - and always sends an affirmative or negative response to the sender. The fact whether the team is valid and whether it gives a positive result — for example, Subscribed — depends both on the team and on the state represented as a private mutable subscribers field.

As mentioned above, only one message is processed at a time, and Akka guarantees that state changes will remain visible when processing the next message, so you do not need to manually synchronize all access to subscribers. Competition is no problem!

Finally, let's look at a fragment of an extended test:

val subscribe01 = Subscribe(topic01, subscriber01.ref)
mediator ! subscribe01
sender.expectMsg(Subscribed(subscribe01))
val subscribe02 = Subscribe(topic01, subscriber02.ref)
mediator ! subscribe02
sender.expectMsg(Subscribed(subscribe02))
val subscribe03 = Subscribe(topic02, subscriber03.ref)
mediator ! subscribe03
sender.expectMsg(Subscribed(subscribe03))

As you can see, we send Subscribe messages to the intermediary using the operator ! and expect to receive appropriate feedback. As above, the entire project code as of the current moment is located at GitHub under the tag step-02 .

Life cycle

So far, we have neglected one important aspect of the actors: their existence is finite - that is, they end, and the actor can be completed at any time.

Having access to ActorRef , we do not know whether the corresponding actor is "alive". In particular, we will not receive an exception if we send messages to the completed actor. In that case, ActorRefremains valid, but Akka forwards and for messages sent to dead mailboxes, non-guaranteed delivery is in effect. Thus, these messages are logged, which is useful for testing, but this method is by no means suitable to implement something like retransmission or even guaranteed delivery.

But sometimes we really need to know whether the actor is still “alive” or not. In this case, we need the ability to get rid of completed subscribers, because otherwise PubSubMediator sends unnecessary messages and may even sooner or later use up all the memory.

For all these reasons, Akka provides the ability to track the life cycle of actors. Since we can only observe the completion of the actor, this mechanism is called " dead watch » (death watch). To track an actor, we simply call the watch method provided by ActorContext , available in Actor via context :

context.watch(subscriber)

Akka will then send the Terminated message to the observer after the observer has completed. This message is guaranteed to be the last one received from the actor, even with a remote connection.

Ok, let's finish PubSubMediator :

class PubSubMediator extends Actor {
  import PubSubMediator._
  ...
  override def receive = {
    ...
    case subscribe @ Subscribe(topic, subscriber) =>
      subscribers += topic -> (subscribers(topic) + subscriber)
      context.watch(subscriber)
      sender() ! Subscribed(subscribe)
    ...
    case Terminated(subscriber) =>
      subscribers = subscribers.map { case (topic, ss) => topic -> (ss - subscriber) }
  }
}

As you can see, we track all subscribers by processing the valid Subscribe command and deleting each completed subscriber when working with the corresponding Terminated message . Again, the full actual code for this example is on GitHub under the tag step-03 .

Conclusion

This concludes our preliminary introduction to the Akka actors. So, we examined the most important aspects of the actor model - communication, behavior and state, and also talked about actor systems. We also discussed the implementation of these concepts with Akka and talked about dead watch.

Of course, I had to omit a lot of interesting and important material: the creation of subsidiary actors, supervision, etc. We refer you to interesting additional resources, for example, excellent Akka documentation .