Typed Actors

Akka Typed Actors is an implementation of the Active Objects pattern. Essentially turning method invocations into asynchronous dispatch instead of synchronous that has been the default way since Smalltalk came out.

Typed Actors consist of 2 "parts", a public interface and an implementation, and if you've done any work in "enterprise" Java, this will be very familiar to you. As with normal Actors you have an external API (the public interface instance) that will delegate method calls asynchronously to a private instance of the implementation.

The advantage of Typed Actors vs. Actors is that with TypedActors you have a static contract, and don't need to define your own messages, the downside is that it places some limitations on what you can do and what you can't, i.e. you can't use become/unbecome.

Typed Actors are implemented using JDK Proxies which provide a pretty easy-worked API to intercept method calls.

注釈

Just as with regular Akka Untyped Actors, Typed Actors process one call at a time.

When to use Typed Actors

Typed actors are nice for bridging between actor systems (the “inside”) and non-actor code (the “outside”), because they allow you to write normal OO-looking code on the outside. Think of them like doors: their practicality lies in interfacing between private sphere and the public, but you don’t want that many doors inside your house, do you? For a longer discussion see this blog post.

A bit more background: TypedActors can easily be abused as RPC, and that is an abstraction which is well-known to be leaky. Hence TypedActors are not what we think of first when we talk about making highly scalable concurrent software easier to write correctly. They have their niche, use them sparingly.

The tools of the trade

Before we create our first Typed Actor we should first go through the tools that we have at our disposal, it's located in akka.actor.TypedActor.

//Returns the Typed Actor Extension
TypedActorExtension extension =
        TypedActor.get(system); //system is an instance of ActorSystem

//Returns whether the reference is a Typed Actor Proxy or not
TypedActor.get(system).isTypedActor(someReference);

//Returns the backing Akka Actor behind an external Typed Actor Proxy
TypedActor.get(system).getActorRefFor(someReference);

//Returns the current ActorContext,
// method only valid within methods of a TypedActor implementation
ActorContext context = TypedActor.context();

//Returns the external proxy of the current Typed Actor,
// method only valid within methods of a TypedActor implementation
Squarer sq = TypedActor.<Squarer>self();

//Returns a contextual instance of the Typed Actor Extension
//this means that if you create other Typed Actors with this,
//they will become children to the current Typed Actor.
TypedActor.get(TypedActor.context());

警告

Same as not exposing this of an Akka Actor, it's important not to expose this of a Typed Actor, instead you should pass the external proxy reference, which is obtained from within your Typed Actor as TypedActor.self(), this is your external identity, as the ActorRef is the external identity of an Akka Actor.

Creating Typed Actors

To create a Typed Actor you need to have one or more interfaces, and one implementation.

The following imports are assumed:

import akka.actor.TypedActor;
import akka.actor.*;
import akka.japi.*;
import akka.dispatch.Futures;

import docs.AbstractJavaTest;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import java.util.concurrent.TimeUnit;

import java.util.List;
import java.util.ArrayList;
import java.util.Random;
import akka.routing.RoundRobinGroup;
public class TypedActorDocTest extends AbstractJavaTest {
    Object someReference = null;
    ActorSystem system = null;

    static
    public interface Squarer {
      void squareDontCare(int i); //fire-forget

      Future<Integer> square(int i); //non-blocking send-request-reply

      Option<Integer> squareNowPlease(int i);//blocking send-request-reply

      int squareNow(int i); //blocking send-request-reply
    }

    static
    class SquarerImpl implements Squarer {
      private String name;

      public SquarerImpl() {
          this.name = "default";
      }

      public SquarerImpl(String name) {
        this.name = name;
      }


      public void squareDontCare(int i) {
          int sq = i * i; //Nobody cares :(
      }

      public Future<Integer> square(int i) {
          return Futures.successful(i * i);
      }

      public Option<Integer> squareNowPlease(int i) {
          return Option.some(i * i);
      }

      public int squareNow(int i) {
          return i * i;
      }
    }

  @Test public void mustGetTheTypedActorExtension() {

    try {

      //Returns the Typed Actor Extension
      TypedActorExtension extension =
              TypedActor.get(system); //system is an instance of ActorSystem

      //Returns whether the reference is a Typed Actor Proxy or not
      TypedActor.get(system).isTypedActor(someReference);

      //Returns the backing Akka Actor behind an external Typed Actor Proxy
      TypedActor.get(system).getActorRefFor(someReference);

      //Returns the current ActorContext,
      // method only valid within methods of a TypedActor implementation
      ActorContext context = TypedActor.context();

      //Returns the external proxy of the current Typed Actor,
      // method only valid within methods of a TypedActor implementation
      Squarer sq = TypedActor.<Squarer>self();

      //Returns a contextual instance of the Typed Actor Extension
      //this means that if you create other Typed Actors with this,
      //they will become children to the current Typed Actor.
      TypedActor.get(TypedActor.context());

    } catch (Exception e) {
      //dun care
    }
  }
  @Test public void createATypedActor() {
    try {
    Squarer mySquarer =
      TypedActor.get(system).typedActorOf(
        new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class));
    Squarer otherSquarer =
      TypedActor.get(system).typedActorOf(
        new TypedProps<SquarerImpl>(Squarer.class,
          new Creator<SquarerImpl>() {
            public SquarerImpl create() { return new SquarerImpl("foo"); }
          }),
        "name");

    mySquarer.squareDontCare(10);

    Future<Integer> fSquare = mySquarer.square(10); //A Future[Int]

    Option<Integer> oSquare = mySquarer.squareNowPlease(10); //Option[Int]

    int iSquare = mySquarer.squareNow(10); //Int

    assertEquals(100, Await.result(fSquare,
      Duration.create(3, TimeUnit.SECONDS)).intValue());

    assertEquals(100, oSquare.get().intValue());

    assertEquals(100, iSquare);

    TypedActor.get(system).stop(mySquarer);

    TypedActor.get(system).poisonPill(otherSquarer);
    } catch(Exception e) {
      //Ignore
    }
  }

  @Test public void createHierarchies() {
    try {
    Squarer childSquarer =
      TypedActor.get(TypedActor.context()).
        typedActorOf(
          new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class)
        );
    //Use "childSquarer" as a Squarer
    } catch (Exception e) {
      //dun care
    }
  }

  @Test public void proxyAnyActorRef() {
    try {
    final ActorRef actorRefToRemoteActor = system.deadLetters();
    Squarer typedActor =
      TypedActor.get(system).
        typedActorOf(
          new TypedProps<Squarer>(Squarer.class),
          actorRefToRemoteActor
        );
    //Use "typedActor" as a FooBar
    } catch (Exception e) {
      //dun care
    }
  }

  interface HasName {
    String name();
  }

  class Named implements HasName {
    private int id = new Random().nextInt(1024);

    @Override public String name() { return "name-" + id; }
  }


  @Test public void typedRouterPattern() {
    try {
      // prepare routees
      TypedActorExtension typed = TypedActor.get(system);

      Named named1 =
        typed.typedActorOf(new TypedProps<Named>(Named.class));

      Named named2 =
        typed.typedActorOf(new TypedProps<Named>(Named.class));

      List<Named> routees = new ArrayList<Named>();
      routees.add(named1);
      routees.add(named2);

      List<String> routeePaths = new ArrayList<String>();
      routeePaths.add(typed.getActorRefFor(named1).path().toStringWithoutAddress());
      routeePaths.add(typed.getActorRefFor(named2).path().toStringWithoutAddress());

      // prepare untyped router
      ActorRef router = system.actorOf(new RoundRobinGroup(routeePaths).props(), "router");

      // prepare typed proxy, forwarding MethodCall messages to `router`
      Named typedRouter = typed.typedActorOf(new TypedProps<Named>(Named.class), router);

      System.out.println("actor was: " + typedRouter.name()); // name-243
      System.out.println("actor was: " + typedRouter.name()); // name-614
      System.out.println("actor was: " + typedRouter.name()); // name-243
      System.out.println("actor was: " + typedRouter.name()); // name-614

      typed.poisonPill(named1);
      typed.poisonPill(named2);
      typed.poisonPill(typedRouter);

    } catch (Exception e) {
      //dun care
    }
  }
}

Our example interface:

public interface Squarer {
  // typed actor iface methods ...
}

Our example implementation of that interface:

class SquarerImpl implements Squarer {
  private String name;

  public SquarerImpl() {
      this.name = "default";
  }

  public SquarerImpl(String name) {
    this.name = name;
  }

  // typed actor impl methods ...
}

The most trivial way of creating a Typed Actor instance of our Squarer:

Squarer mySquarer =
  TypedActor.get(system).typedActorOf(
    new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class));

First type is the type of the proxy, the second type is the type of the implementation. If you need to call a specific constructor you do it like this:

Squarer otherSquarer =
  TypedActor.get(system).typedActorOf(
    new TypedProps<SquarerImpl>(Squarer.class,
      new Creator<SquarerImpl>() {
        public SquarerImpl create() { return new SquarerImpl("foo"); }
      }),
    "name");

Since you supply a Props, you can specify which dispatcher to use, what the default timeout should be used and more. Now, our Squarer doesn't have any methods, so we'd better add those.

public interface Squarer {
  void squareDontCare(int i); //fire-forget

  Future<Integer> square(int i); //non-blocking send-request-reply

  Option<Integer> squareNowPlease(int i);//blocking send-request-reply

  int squareNow(int i); //blocking send-request-reply
}

Alright, now we've got some methods we can call, but we need to implement those in SquarerImpl.

class SquarerImpl implements Squarer {
  private String name;

  public SquarerImpl() {
      this.name = "default";
  }

  public SquarerImpl(String name) {
    this.name = name;
  }


  public void squareDontCare(int i) {
      int sq = i * i; //Nobody cares :(
  }

  public Future<Integer> square(int i) {
      return Futures.successful(i * i);
  }

  public Option<Integer> squareNowPlease(int i) {
      return Option.some(i * i);
  }

  public int squareNow(int i) {
      return i * i;
  }
}

Excellent, now we have an interface and an implementation of that interface, and we know how to create a Typed Actor from that, so let's look at calling these methods.

Method dispatch semantics

Methods returning:

void will be dispatched with fire-and-forget semantics, exactly like ActorRef.tell

scala.concurrent.Future<?> will use send-request-reply semantics, exactly like ActorRef.ask

akka.japi.Option<?> will use send-request-reply semantics, but will block to wait for an answer, and return akka.japi.Option.None if no answer was produced within the timeout, or akka.japi.Option.Some<?> containing the result otherwise. Any exception that was thrown during this call will be rethrown.

Any other type of value will use send-request-reply semantics, but will block to wait for an answer, throwing java.util.concurrent.TimeoutException if there was a timeout or rethrow any exception that was thrown during this call. Note that due to the Java exception and reflection mechanisms, such a TimeoutException will be wrapped in a java.lang.reflect.UndeclaredThrowableException unless the interface method explicitly declares the TimeoutException as a thrown checked exception.

Messages and immutability

While Akka cannot enforce that the parameters to the methods of your Typed Actors are immutable, we strongly recommend that parameters passed are immutable.

One-way message send

mySquarer.squareDontCare(10);

As simple as that! The method will be executed on another thread; asynchronously.

Request-reply message send

Option<Integer> oSquare = mySquarer.squareNowPlease(10); //Option[Int]

This will block for as long as the timeout that was set in the Props of the Typed Actor, if needed. It will return None if a timeout occurs.

int iSquare = mySquarer.squareNow(10); //Int

This will block for as long as the timeout that was set in the Props of the Typed Actor, if needed. It will throw a java.util.concurrent.TimeoutException if a timeout occurs. Note that here, such a TimeoutException will be wrapped in a java.lang.reflect.UndeclaredThrowableException by the Java reflection mechanism, because the interface method does not explicitly declare the TimeoutException as a thrown checked exception. To get the TimeoutException directly, declare throws java.util.concurrent.TimeoutException at the interface method.

Request-reply-with-future message send

Future<Integer> fSquare = mySquarer.square(10); //A Future[Int]

This call is asynchronous, and the Future returned can be used for asynchronous composition.

Stopping Typed Actors

Since Akka's Typed Actors are backed by Akka Actors they must be stopped when they aren't needed anymore.

TypedActor.get(system).stop(mySquarer);

This asynchronously stops the Typed Actor associated with the specified proxy ASAP.

TypedActor.get(system).poisonPill(otherSquarer);

This asynchronously stops the Typed Actor associated with the specified proxy after it's done with all calls that were made prior to this call.

Typed Actor Hierarchies

Since you can obtain a contextual Typed Actor Extension by passing in an ActorContext you can create child Typed Actors by invoking typedActorOf(..) on that.

Squarer childSquarer =
  TypedActor.get(TypedActor.context()).
    typedActorOf(
      new TypedProps<SquarerImpl>(Squarer.class, SquarerImpl.class)
    );
//Use "childSquarer" as a Squarer

You can also create a child Typed Actor in regular Akka Actors by giving the UntypedActorContext as an input parameter to TypedActor.get(…).

Supervisor Strategy

By having your Typed Actor implementation class implement TypedActor.Supervisor you can define the strategy to use for supervising child actors, as described in スーパービジョンとモニタリング and 耐障害性.

Receive arbitrary messages

If your implementation class of your TypedActor extends akka.actor.TypedActor.Receiver, all messages that are not MethodCall instances will be passed into the onReceive-method.

This allows you to react to DeathWatch Terminated-messages and other types of messages, e.g. when interfacing with untyped actors.

Lifecycle callbacks

By having your Typed Actor implementation class implement any and all of the following:

TypedActor.PreStart

TypedActor.PostStop

TypedActor.PreRestart

TypedActor.PostRestart

You can hook into the lifecycle of your Typed Actor.

Proxying

You can use the typedActorOf that takes a TypedProps and an ActorRef to proxy the given ActorRef as a TypedActor. This is usable if you want to communicate remotely with TypedActors on other machines, just pass the ActorRef to typedActorOf.

Lookup & Remoting

Since TypedActors are backed by Akka Actors, you can use typedActorOf to proxy ActorRefs potentially residing on remote nodes.

Squarer typedActor =
  TypedActor.get(system).
    typedActorOf(
      new TypedProps<Squarer>(Squarer.class),
      actorRefToRemoteActor
    );
//Use "typedActor" as a FooBar

Typed Router pattern

Sometimes you want to spread messages between multiple actors. The easiest way to achieve this in Akka is to use a Router, which can implement a specific routing logic, such as smallest-mailbox or consistent-hashing etc.

Routers are not provided directly for typed actors, but it is really easy to leverage an untyped router and use a typed proxy in front of it. To showcase this let's create typed actors that assign themselves some random id, so we know that in fact, the router has sent the message to different actors:

  @Test public void typedRouterPattern() {
    try {
      // prepare routees
      TypedActorExtension typed = TypedActor.get(system);

      Named named1 =
        typed.typedActorOf(new TypedProps<Named>(Named.class));

      Named named2 =
        typed.typedActorOf(new TypedProps<Named>(Named.class));

      List<Named> routees = new ArrayList<Named>();
      routees.add(named1);
      routees.add(named2);

      List<String> routeePaths = new ArrayList<String>();
      routeePaths.add(typed.getActorRefFor(named1).path().toStringWithoutAddress());
      routeePaths.add(typed.getActorRefFor(named2).path().toStringWithoutAddress());

      // prepare untyped router
      ActorRef router = system.actorOf(new RoundRobinGroup(routeePaths).props(), "router");

      // prepare typed proxy, forwarding MethodCall messages to `router`
      Named typedRouter = typed.typedActorOf(new TypedProps<Named>(Named.class), router);

      System.out.println("actor was: " + typedRouter.name()); // name-243
      System.out.println("actor was: " + typedRouter.name()); // name-614
      System.out.println("actor was: " + typedRouter.name()); // name-243
      System.out.println("actor was: " + typedRouter.name()); // name-614

      typed.poisonPill(named1);
      typed.poisonPill(named2);
      typed.poisonPill(typedRouter);

    } catch (Exception e) {
      //dun care
    }
  }
}

In order to round robin among a few instances of such actors, you can simply create a plain untyped router, and then facade it with a TypedActor like shown in the example below. This works because typed actors of course communicate using the same mechanisms as normal actors, and methods calls on them get transformed into message sends of MethodCall messages.

// prepare routees
TypedActorExtension typed = TypedActor.get(system);

Named named1 =
  typed.typedActorOf(new TypedProps<Named>(Named.class));

Named named2 =
  typed.typedActorOf(new TypedProps<Named>(Named.class));

List<Named> routees = new ArrayList<Named>();
routees.add(named1);
routees.add(named2);

List<String> routeePaths = new ArrayList<String>();
routeePaths.add(typed.getActorRefFor(named1).path().toStringWithoutAddress());
routeePaths.add(typed.getActorRefFor(named2).path().toStringWithoutAddress());

// prepare untyped router
ActorRef router = system.actorOf(new RoundRobinGroup(routeePaths).props(), "router");

// prepare typed proxy, forwarding MethodCall messages to `router`
Named typedRouter = typed.typedActorOf(new TypedProps<Named>(Named.class), router);

System.out.println("actor was: " + typedRouter.name()); // name-243
System.out.println("actor was: " + typedRouter.name()); // name-614
System.out.println("actor was: " + typedRouter.name()); // name-243
System.out.println("actor was: " + typedRouter.name()); // name-614

Contents

Typed Actors

When to use Typed Actors

The tools of the trade

Creating Typed Actors

Method dispatch semantics

Messages and immutability

One-way message send

Request-reply message send

Request-reply-with-future message send

Stopping Typed Actors

Typed Actor Hierarchies

Supervisor Strategy

Receive arbitrary messages

Lifecycle callbacks

Proxying

Lookup & Remoting

Typed Router pattern

Akka

Contribute

Company