About Adam Warski

Adam is one of the co-founders of SoftwareMill, a company specialising in delivering customised software solutions. He is also involved in open-source projects, as a founder, lead developer or contributor to: Hibernate Envers, a Hibernate core module, which provides entity versioning/auditing capabilities; ElasticMQ, an SQS-compatible messaging server written in Scala; Veripacks, a tool to specify and verify inter-package dependencies, and others.

Making the Reactive Queue durable with Akka Persistence

Some time ago I wrote how to implement a reactive message queue with Akka Streams. The queue supports streaming send and receive operations with back-pressure, but has one downside: all messages are stored in-memory, and hence in case of a restart are lost.

But this can be easily solved with the experimental akka-persistence module, which just got an update in Akka 2.3.4.
 
 
 
 

Queue actor refresher

To make the queue durable, we only need to change the queue actor; the reactive/streaming parts remain intact. Just as a reminder, the reactive queue consists of:

  • a single queue actor, which holds an internal priority queue of messages to be delivered. The queue actor accepts actor-messages to send, receive and delete queue-messages
  • a Broker, which creates the queue actor, listens for connections from senders and receivers, and creates the reactive streams when a connection is established
  • a Sender, which sends messages to the queue (for testing, one message each second). Multiple senders can be started. Messages are sent only if they can be accepted (back-pressure from the broker)
  • a Receiver, which receives messages from queue, as they become available and as they can be processed (back-pressure from the receiver).

reactmq-actors1

Going persistent (remaining reactive)

The changes needed are quite minimal.

First of all, the QueueActor needs to extend PersistentActor, and define two methods:

  • receiveCommand, which defines the “normal” behaviour when actor-messages (commands) arrive
  • receiveRecover, which is used during recovery only, and where replayed events are sent

But in order to recover, we first need to persist some events! This should of course be done when handling the message queue operations.

For example, when sending a message, a MessageAdded event is persisted using persistAsync:

def handleQueueMsg: Receive = {
  case SendMessage(content) =>
    val msg = sendMessage(content)
    persistAsync(msg.toMessageAdded) { msgAdded =>
      sender() ! SentMessage(msgAdded.id)
      tryReply()
    }
 
   // ...
}

persistAsync is one way of persisting events using akka-persistence. The other, persist (which is also the default one), buffers subsequent commands (actor-messages) until the event is persisted; this is a bit slower, but also easier to reason about and remain consistent. However in case of the message queue such behaviour isn’t necessary. The only guarantee that we need is that the message send is acknowledged only after the event is persisted; and that’s why the reply is sent in the after-persist event handler. You can read more about persistAsync in the docs.

Similarly, events are persisted for other commands (actor-messages, see QueueActorReceive). Both for deletes and receives we are using persistAsync, as the queue aims to provide an at-least-once delivery guarantee.

The final component is the recovery handler, which is defined in QueueActorRecover (and then used in QueueActor). Recovery is quite simple: the events correspond to adding a new message, updating the “next delivery” timestamp or deleting.

The internal representation uses both a priority queue and a by-id map for efficiency, so when the events are handled during recovert we only build the map, and use the RecoveryCompleted special event to build the queue as well. The special event is sent by akka-persistence automatically.

And that’s all! If you now run the broker, send some messages, stop the broker, start it again, you’ll see that the messages are recovered, and indeed, they get received if a receiver is run.

The code isn’t production-ready of course. The event log is going to constantly grow, so it would certainly make sense to make use of snapshots, plus delete old events/snapshots to make the storage size small and recovery fast.

Replication

Now that the queue is durable, we can also have a replicated persistent queue almost for free: we simply need to use a different journal plugin! The default one relies on LevelDB and writes data to the local disk. Other implementations are available: for Cassandra, HBase, and Mongo.

Making a simple switch of the persistence backend we can have our messages replicated across a cluster.

Summary

With the help of two experimental Akka modules, reactive streams and persistence, we have been able to implement a durable, reactive queue with a quite minimal amount of code. And that’s just the beginning, as the two technologies are only starting to mature!

Related Whitepaper:

Java Essential Training

Author David Gassner explores Java SE (Standard Edition), the language used to build mobile apps for Android devices, enterprise server applications, and more!

The course demonstrates how to install both Java and the Eclipse IDE and dives into the particulars of programming. The course also explains the fundamentals of Java, from creating simple variables, assigning values, and declaring methods to working with strings, arrays, and subclasses; reading and writing to text files; and implementing object oriented programming concepts. Exercise files are included with the course.

Get it Now!  

Leave a Reply


eight × = 56



Java Code Geeks and all content copyright © 2010-2014, Exelixis Media Ltd | Terms of Use | Privacy Policy
All trademarks and registered trademarks appearing on Java Code Geeks are the property of their respective owners.
Java is a trademark or registered trademark of Oracle Corporation in the United States and other countries.
Java Code Geeks is not connected to Oracle Corporation and is not sponsored by Oracle Corporation.

Sign up for our Newsletter

20,709 insiders are already enjoying weekly updates and complimentary whitepapers! Join them now to gain exclusive access to the latest news in the Java world, as well as insights about Android, Scala, Groovy and other related technologies.

As an extra bonus, by joining you will get our brand new e-books, published by Java Code Geeks and their JCG partners for your reading pleasure! Enter your info and stay on top of things,

  • Fresh trends
  • Cases and examples
  • Research and insights
  • Two complimentary e-books