1. Introduction
In this article, we’re going to look at Aeron, a multi-language library maintained by Adaptive Financial Consulting, designed for efficient UDP messaging between applications. It’s designed for performance, aiming for high throughput, low latency, and fault tolerance.
2. Dependencies
Before we can use Aeron, we need to include the latest version in our build, which is 1.44.1 at the time of writing.
If we’re using Maven, we can include its dependency in pom.xml:
<dependency>
<groupId>io.aeron</groupId>
<artifactId>aeron-all</artifactId>
<version>1.44.1</version>
</dependency>
Or if we’re using Gradle, we can include it in build.gradle:
implementation("io.aeron:aeron-all:1.44.1")
At this point, we’re ready to start using it in our application.
Note that, at present, some parts of Aeron don’t work out of the box with Java 16 or newer. This is due to specific interactions that JPMS blocks.
3. Media Driver
Aeron works with a level of indirection between the application and the transport. This is known as the Media Driver because it’s the interaction between our application and the transmission media.
Every Aeron process interacts with a media driver and, through that, can interact with other processes – either on the same machine or remotely. It performs this interaction via the file system. We need to point the media driver and all applications to the same directory on disk, where it stores various aspects. Note that we can only have a single media driver running for any given directory simultaneously. Attempting to run more than one will fail.
We’re able to run the media driver embedded within our application when we want to keep things simple:
MediaDriver mediaDriver = MediaDriver.launch();
This will launch a media driver with all of the default settings. In particular, this will run with the default media driver directory.
We also have an alternative launch method that’s designed for embedded use. This acts exactly as before, only it generates a random directory to ensure that multiple instances on the same machine won’t clash:
MediaDriver mediaDriver = MediaDriver.launchEmbedded();
In both of these cases, we can also provide a MediaDriver.Context object to further configure the media driver:
MediaDriver.Context context = new MediaDriver.Context();
context.threadingMode(ThreadingMode.SHARED);
MediaDriver mediaDriver = MediaDriver.launch(context);
When doing this, we need to close the media driver when we’ve finished with it. The interface implements AutoCloseable, so we can use the try-with-resources pattern to manage this.
Alternatively, we can run the media driver as an external application. We can do this using the aeron-all.jar JAR file that we’ve included as our dependency:
$ java -cp aeron-all-1.44.1.jar io.aeron.driver.MediaDriver
This will function precisely the same as MediaDriver.launch() above.
4. Aeron API Client
We perform all API interactions using Aeron via the Aeron class. We need to create a new instance of this and point it at our media driver. Simply creating a new instance will point at the media driver in the default location – precisely as if we’d launched it with MediaDriver.launch():
Aeron aeron = Aeron.connect();
Alternatively, we can provide an Aeron.Context object to configure the connection, including specifying the directory that the media driver is running in:
Aeron.Context ctx = new Aeron.Context();
ctx.aeronDirectoryName(mediaDriver.aeronDirectoryName());
Aeron aeron = Aeron.connect(ctx);
If our media driver is in a non-standard directory, including if we started it with MediaDriver.launchEmbedded(), we must do this. If the directory that we’re pointing at doesn’t have a running media driver, the Aeron.connect() call will block until it does.
We can connect as many Aeron clients as we need to the same media driver. Typically, these would be from different applications, but they can be from the same one if needed. However, if we do this, then we need to use new instances of Aeron.Context as well:
Aeron.Context ctx1 = new Aeron.Context();
ctx1.aeronDirectoryName(mediaDriver.aeronDirectoryName());
aeron1 = Aeron.connect(ctx1);
System.out.println("Aeron 1 connected: " + aeron1);
Aeron.Context ctx2 = new Aeron.Context();
ctx2.aeronDirectoryName(mediaDriver.aeronDirectoryName());
aeron2 = Aeron.connect(ctx2);
System.out.println("Aeron 2 connected: " + aeron2);
As with the MediaDriver, the Aeron instance is AutoCloseable. This means we can wrap it with the try-with-resources pattern to ensure that we close it correctly.
5. Sending and Receiving Messages
Now that we’ve got our Aeron API client, we’re ready to use it to send and receive messages.
5.1. Buffers
Aeron represents all messages – both sending and receiving – as DirectBuffer instances. Ultimately, these are nothing more than a set of bytes, but they provide us with a set of methods to work with a standard set of types.
When we’re sending messages, we need to construct the buffer ourselves from our own data. For this, we’re best off using an UnsafeBuffer instance – named because it uses sun.misc.Unsafe to read and write the values from our underlying buffer. Creating this requires either a byte array or a ByteBuffer instance, and we can then use BufferUtil.allocateDirectAligned() to help with making this most efficiently:
UnsafeBuffer buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned(256, 64));
Once we’ve got our buffer, we then have a whole range of getXyz() and putXyz() methods that we can use to manipulate the data in our buffer:
// Put a string into the buffer starting at index 0.
int length = buffer.putStringWithoutLengthUtf8(0, message);
// Read a string of the given length from the buffer starting from the given offset.
String message = buffer.getStringWithoutLengthUtf8(offset, length);
Note that we need to manage the offsets in the buffer ourselves. Whenever we put data into the buffer, it returns the length of the written data so we can calculate the next offset. When we read from the buffer, we need to know what the length will be.
5.2. Channels and Streams
Sending and Receiving data with Aeron is done using identified streams transmitted over specific channels.
We specify channels as URIs in a particular format, telling Aeron how to transmit the messages. Our media driver then uses this to interact with our transmission media, ensuring that it sends and receives the messages correctly. Streams are identified simply as numbers. The only requirement is that the two ends of the same communication use the same stream ID.
The simplest such channel is aeron:ipc, which transmits and receives using shared memory within the media driver. Note that this can only work if both sides use the same media driver and don’t allow for networking.
More usefully, we can use aeron:udp to send and receive using UDP. This allows us to communicate with any other application anywhere we can connect. In particular, our application will communicate with the media driver, and then the media drivers will communicate with each other:
When specifying a UDP channel, we need to include at least the host and port. On the receiving side, this is where we’ll be listening, and on the sending side, this is where we’ll be sending messages. For example, aeron:udp?endpoint=localhost:20121 will send and receive messages via UDP on localhost:20121.
5.3. Subscriptions
Once our media driver and Aeron client are set up, we’re ready to receive messages. We do this by creating a subscription to a particular stream on a particular channel and then polling this for messages.
Adding a subscription is enough for the media driver to set up everything to be able to receive our messages. We do this with the addSubscription() method on our Aeron instance:
Subscription subscription = aeron.addSubscription("aeron:udp?endpoint=localhost:20121", 1001);
As before, we need to close this when we no longer use it so the media driver knows to stop listening for messages. As always, this is AutoCloseable, so we can use try-with-resources to manage it.
When we have our subscription, we need to receive messages. Aeron performs this with a polling mechanism, giving us complete control over when it processes messages. To poll for messages, we need to provide a FragmentHandler that will process the message received. We can implement this with a lambda if we want to have all of the code inline or as a separate class implementing the interface if we want to reuse it:
FragmentHandler fragmentHandler = (buffer, offset, length, header) -> {
String data = buffer.getStringWithoutLengthUtf8(offset, length);
System.out.printf("Message from session %d (%d@%d) <<%s>>%n",
header.sessionId(), length, offset, data);
};
Aeron calls this with a buffer, the offset into which the data starts, and the length of the data received. We can then process this buffer however we need for our application.
When we’re ready to poll for a new message, we use the Subscription.poll() method:
int fragmentsRead = subscription.poll(fragmentHandler, 10);
Here, we’ve provided our FragmentHandler instance and the number of message fragments to consider when trying to receive a single message. Note that we’ll receive up to one message at a time, even if many are available in the media driver. However, if no messages are available, this will immediately return, and if the messages received are too large, we might receive only part of them.
5.4. Publications
The other side of our messaging is sending messages. We do this with a Publication, which can send messages to a particular stream on a particular channel.
We can add a new publication with the Aeron.addPublication() method. We then need to wait for it to connect, which requires that a subscription is on the receiving end ready to receive the messages:
ConcurrentPublication publication = aeron.addPublication("aeron:udp?endpoint=localhost:20121", 1001);
while (!publication.isConnected()) {
TimeUnit.MILLISECONDS.sleep(100);
}
If there’s no connection, it will immediately fail to send the messages rather than waiting for someone to add a subscription.
As before, we need to close this when we’re no longer using it so that the media driver can free up any allocated resources. As always, this is AutoCloseable, so we can use try-with-resources to manage it.
Once we’ve got a connected publication, we can offer it messages. These are always provided as populated buffers, which will then be sent to the connected subscriber:
UnsafeBuffer buffer = new UnsafeBuffer(BufferUtil.allocateDirectAligned(256, 64));
buffer.putStringWithoutLengthUtf8(0, message);
long result = publication.offer(buffer, 0, message.length());
If the message was sent, we’ll be returned a value indicating the number of bytes transmitted, which might be smaller than the number of bytes we expected to send if the buffer was too large. Alternatively, it might return one of a set of error codes to us, all of which are negative numbers and, therefore, easily distinguishable from the success case:
- Publication.NOT_CONNECTED – The publication wasn’t connected to a subscriber.
- Publication.BACK_PRESSURED – Back pressure from the subscribers means that we can’t send any more messages right now.
- Publication.ADMIN_ACTION – Some administrative actions, such as log rotation, caused the send to fail. In this case, it’s typically safe to immediately retry.
- Publication.CLOSED – The Publication instance has been closed.
- Publication.MAX_POSITION_EXCEEDED – The buffer within the media driver is full. Typically, we can solve this by closing the Publication and creating a new one instead.
6. Conclusion
We’ve seen a quick overview of Aeron, how to set it up, and how to use it for messaging between applications. This library can do much more, so why not try it out and see?
All of the examples are available over on GitHub.