ZeroMQ
Loading

ZeroMQ

Akka provides a ZeroMQ module which abstracts a ZeroMQ connection and therefore allows interaction between Akka actors to take place over ZeroMQ connections. The messages can be of a proprietary format or they can be defined using Protobuf. The socket actor is fault-tolerant by default and when you use the newSocket method to create new sockets it will properly reinitialize the socket.

ZeroMQ is very opinionated when it comes to multi-threading so configuration option akka.zeromq.socket-dispatcher always needs to be configured to a PinnedDispatcher, because the actual ZeroMQ socket can only be accessed by the thread that created it.

The ZeroMQ module for Akka is written against an API introduced in JZMQ, which uses JNI to interact with the native ZeroMQ library. Instead of using JZMQ, the module uses ZeroMQ binding for Scala that uses the native ZeroMQ library through JNA. In other words, the only native library that this module requires is the native ZeroMQ library. The benefit of the scala library is that you don't need to compile and manage native dependencies at the cost of some runtime performance. The scala-bindings are compatible with the JNI bindings so they are a drop-in replacement, in case you really need to get that extra bit of performance out.

Note

The currently used version of zeromq-scala-bindings is only compatible with zeromq 2; zeromq 3 is not supported.

Connection

ZeroMQ supports multiple connectivity patterns, each aimed to meet a different set of requirements. Currently, this module supports publisher-subscriber connections and connections based on dealers and routers. For connecting or accepting connections, a socket must be created. Sockets are always created using the akka.zeromq.ZeroMQExtension, for example:

import akka.zeromq.Bind;
import akka.zeromq.ZeroMQExtension;
ActorRef pubSocket = ZeroMQExtension.get(system).newPubSocket(
  new Bind("tcp://127.0.0.1:1233"));

Above examples will create a ZeroMQ Publisher socket that is Bound to the port 21231 on localhost.

Similarly you can create a subscription socket, with a listener, that subscribes to all messages from the publisher using:

import akka.zeromq.Connect;
import akka.zeromq.Listener;
import akka.zeromq.Subscribe;
ActorRef listener = system.actorOf(Props.create(ListenerActor.class));
ActorRef subSocket = ZeroMQExtension.get(system).newSubSocket(
  new Connect("tcp://127.0.0.1:1233"),
  new Listener(listener), Subscribe.all());
public class ListenerActor extends UntypedActor {
  public void onReceive(Object message) throws Exception {
    //...
  }
}

The following sub-sections describe the supported connection patterns and how they can be used in an Akka environment. However, for a comprehensive discussion of connection patterns, please refer to ZeroMQ -- The Guide.

Publisher-Subscriber Connection

In a publisher-subscriber (pub-sub) connection, the publisher accepts one or more subscribers. Each subscriber shall subscribe to one or more topics, whereas the publisher publishes messages to a set of topics. Also, a subscriber can subscribe to all available topics. In an Akka environment, pub-sub connections shall be used when an actor sends messages to one or more actors that do not interact with the actor that sent the message.

When you're using zeromq pub/sub you should be aware that it needs multicast - check your cloud - to work properly and that the filtering of events for topics happens client side, so all events are always broadcasted to every subscriber.

An actor is subscribed to a topic as follows:

ActorRef subTopicSocket = ZeroMQExtension.get(system).newSubSocket(
  new Connect("tcp://127.0.0.1:1233"),
  new Listener(listener), new Subscribe("foo.bar"));

It is a prefix match so it is subscribed to all topics starting with foo.bar. Note that if the given string is empty or Subscribe.all() is used, the actor is subscribed to all topics.

To unsubscribe from a topic you do the following:

import akka.zeromq.Unsubscribe;
subTopicSocket.tell(new Unsubscribe("foo.bar"), ActorRef.noSender());

To publish messages to a topic you must use two Frames with the topic in the first frame.

import akka.util.ByteString;
import akka.zeromq.ZMQMessage;
pubSocket.tell(ZMQMessage.withFrames(ByteString.fromString("foo.bar"), 
  ByteString.fromArray(payload)), ActorRef.noSender());

Pub-Sub in Action

The following example illustrates one publisher with two subscribers.

The publisher monitors current heap usage and system load and periodically publishes Heap events on the "health.heap" topic and Load events on the "health.load" topic.

import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import org.junit.*;
import scala.concurrent.duration.Duration;
import akka.serialization.SerializationExtension;
import akka.serialization.Serialization;
import java.io.Serializable;
import java.lang.management.ManagementFactory;
public final Object TICK = "TICK";

public class Heap implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long timestamp;
  public final long used;
  public final long max;

  public Heap(long timestamp, long used, long max) {
    this.timestamp = timestamp;
    this.used = used;
    this.max = max;
  }
}

public class Load implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long timestamp;
  public final double loadAverage;

  public Load(long timestamp, double loadAverage) {
    this.timestamp = timestamp;
    this.loadAverage = loadAverage;
  }
}

public class HealthProbe extends UntypedActor {

  ActorRef pubSocket = ZeroMQExtension.get(getContext().system()).newPubSocket(
    new Bind("tcp://127.0.0.1:1237"));
  MemoryMXBean memory = ManagementFactory.getMemoryMXBean();
  OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean();
  Serialization ser = SerializationExtension.get(getContext().system());

  @Override
  public void preStart() {
    getContext().system().scheduler()
        .schedule(Duration.create(1, "second"), Duration.create(1, "second"),
          getSelf(), TICK, getContext().dispatcher(), ActorRef.noSender());
  }

  @Override
  public void postRestart(Throwable reason) {
    // don't call preStart, only schedule once
  }

  @Override
  public void onReceive(Object message) {
    if (message.equals(TICK)) {
      MemoryUsage currentHeap = memory.getHeapMemoryUsage();
      long timestamp = System.currentTimeMillis();

      // use akka SerializationExtension to convert to bytes
      ByteString heapTopic = ByteString.fromString("health.heap", "UTF-8");
      ByteString heapPayload = ByteString.fromArray(
              ser.serialize(
                  new Heap(timestamp,
                        currentHeap.getUsed(),
                        currentHeap.getMax())
              ).get());
      // the first frame is the topic, second is the message
      pubSocket.tell(ZMQMessage.withFrames(heapTopic, heapPayload), getSelf());

      // use akka SerializationExtension to convert to bytes
      ByteString loadTopic = ByteString.fromString("health.load", "UTF-8");
      ByteString loadPayload = ByteString.fromArray(
              ser.serialize(new Load(timestamp, os.getSystemLoadAverage())).get()
      );
      // the first frame is the topic, second is the message
      pubSocket.tell(ZMQMessage.withFrames(loadTopic, loadPayload), getSelf());
    } else {
      unhandled(message);
    }
  }

}
system.actorOf(Props.create(HealthProbe.class), "health");

Let's add one subscriber that logs the information. It subscribes to all topics starting with "health", i.e. both Heap and Load events.

public class Logger extends UntypedActor {

  ActorRef subSocket = ZeroMQExtension.get(getContext().system()).newSubSocket(
    new Connect("tcp://127.0.0.1:1237"),
      new Listener(getSelf()), new Subscribe("health"));
  Serialization ser = SerializationExtension.get(getContext().system());
  SimpleDateFormat timestampFormat = new SimpleDateFormat("HH:mm:ss.SSS");
  LoggingAdapter log = Logging.getLogger(getContext().system(), this);

  @Override
  public void onReceive(Object message) {
    if (message instanceof ZMQMessage) {
      ZMQMessage m = (ZMQMessage) message;
      String topic = m.frame(0).utf8String();
      // the first frame is the topic, second is the message
      if ("health.heap".equals(topic)) {
        Heap heap = ser.deserialize(m.frame(1).toArray(), Heap.class).get();
        log.info("Used heap {} bytes, at {}", heap.used,
          timestampFormat.format(new Date(heap.timestamp)));
      } else if ("health.load".equals(topic)) {
        Load load = ser.deserialize(m.frame(1).toArray(), Load.class).get();
        log.info("Load average {}, at {}", load.loadAverage,
          timestampFormat.format(new Date(load.timestamp)));
      }
    } else {
      unhandled(message);
    }
  }

}
system.actorOf(Props.create(Logger.class), "logger");

Another subscriber keep track of used heap and warns if too much heap is used. It only subscribes to Heap events.

public class HeapAlerter extends UntypedActor {

  ActorRef subSocket = ZeroMQExtension.get(getContext().system()).newSubSocket(
    new Connect("tcp://127.0.0.1:1237"),
    new Listener(getSelf()), new Subscribe("health.heap"));
  Serialization ser = SerializationExtension.get(getContext().system());
  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  int count = 0;

  @Override
  public void onReceive(Object message) {
    if (message instanceof ZMQMessage) {
      ZMQMessage m = (ZMQMessage) message;
      String topic = m.frame(0).utf8String();
      // the first frame is the topic, second is the message
      if ("health.heap".equals(topic)) {
        Heap heap = ser.<Heap>deserialize(m.frame(1).toArray(), Heap.class).get();
        if (((double) heap.used / heap.max) > 0.9) {
          count += 1;
        } else {
          count = 0;
        }
        if (count > 10) {
          log.warning("Need more memory, using {} %",
            (100.0 * heap.used / heap.max));
        }
      }
    } else {
      unhandled(message);
    }
  }

}
system.actorOf(Props.create(HeapAlerter.class), "alerter");

Router-Dealer Connection

While Pub/Sub is nice the real advantage of zeromq is that it is a "lego-box" for reliable messaging. And because there are so many integrations the multi-language support is fantastic. When you're using ZeroMQ to integrate many systems you'll probably need to build your own ZeroMQ devices. This is where the router and dealer socket types come in handy. With those socket types you can build your own reliable pub sub broker that uses TCP/IP and does publisher side filtering of events.

To create a Router socket that has a high watermark configured, you would do:

ActorRef highWatermarkSocket = ZeroMQExtension.get(system).newRouterSocket(
    new SocketOption[] { new Listener(listener),
      new Bind("tcp://127.0.0.1:1233"), new HighWatermark(50000) });

The akka-zeromq module accepts most if not all the available configuration options for a zeromq socket.

Push-Pull Connection

Akka ZeroMQ module supports Push-Pull connections.

You can create a Push connection through the:

ActorRef newPushSocket(SocketOption[] socketParameters);

You can create a Pull connection through the:

ActorRef newPullSocket(SocketOption[] socketParameters);

More documentation and examples will follow soon.

Rep-Req Connection

Akka ZeroMQ module supports Rep-Req connections.

You can create a Rep connection through the:

ActorRef newRepSocket(SocketOption[] socketParameters);

You can create a Req connection through the:

ActorRef newReqSocket(SocketOption[] socketParameters);

More documentation and examples will follow soon.

Configuration

There are several configuration properties for the zeromq module, please refer to the reference configuration.

Contents