AWS Compute Blog
Implementing enterprise integration patterns with AWS messaging services: point-to-point channels
This post is courtesy of Christian Mueller, Sr. Solutions Architect, AWS and Dirk Fröhner, Sr. Solutions Architect, AWS
At AWS, we see our customers increasingly moving toward managed services to reduce the time and money that they spend managing infrastructure. This also applies to the messaging domain, where AWS provides a collection of managed services.
Asynchronous messaging is a fundamental approach for integrating independent systems or building up a set of loosely coupled systems that can scale and evolve independently and flexibly. The well-known collection of enterprise integration patterns (EIPs) provides a “technology-independent vocabulary” to “design and document integration solutions.” This blog is the first of two that describes how you can implement the core EIPs using AWS messaging services. Let’s first look at the relevant AWS messaging services.
When organizations migrate their traditional messaging and existing applications to the cloud gradually, they usually want to do it without rewriting their code. Amazon MQ is a managed message broker service for Apache ActiveMQ that makes it easy to set up and operate message brokers in the cloud. It supports industry-standard APIs and protocols such as JMS, AMQP, and MQTT, so you can switch from any standards-based message broker to Amazon MQ without rewriting the messaging code in your applications. Amazon MQ is recommended if you’re using messaging with existing applications and want to move your messaging to the cloud without rewriting existing code.
However, if you build new applications for the cloud, we recommend that you consider using cloud-native messaging services such as Amazon SQS and Amazon SNS. These serverless, fully managed message queue and topic services scale to meet your demands and provide simple, easy-to-use APIs. You can use Amazon SQS and Amazon SNS to decouple and scale microservices, distributed systems, and serverless applications and improve overall reliability.
This blog looks at the first part of some fundamental integration patterns. We describe the patterns and apply them to these AWS messaging services. This will help you apply the right pattern to your use case and architect for scale in a secure and cost-efficient manner. For all variants, we employ both traditional and cloud-native messaging services: Amazon MQ for the former and Amazon SQS and Amazon SNS for the latter.
Integration Patterns
Let’s start with some fundamental integration patterns.
Message exchange patterns
First, we inspect the two major message exchange patterns: one-way and request-response.
One-way messaging
Applying one-way messaging, a message producer (sender) sends out a message to a messaging channel and doesn’t expect or want a response from whatever process (receiver) consumed the message. Examples of one-way messaging include a data transfer and a notification about an event that happened.
Request-response messaging
With request-response messaging, a message producer (requester) sends out a message: for example, a command to instruct the responder to execute something. The requester expects a response from each message consumer (responder) who received that message, likely to know what the result of all executions was. To know where to send the response message to, the request message contains a return address that the responder uses. To make sure that the requester can assign an incoming response to a request, the requester adds a correlation identifier to the request, which the responders echo in their responses.
Messaging channels: point-to-point
Next, we look at the point-to-point messaging channel, one of the most important patterns for messaging channels. We will continue our consideration with publish-subscribe in our second post.
A point-to-point channel is usually implemented by message queues. Message queues operate so that any given message is only consumed by one receiver, although multiple receivers can be connected to the queue. The queue ensures once-only consumption. Messages are usually buffered in queues so that they’re available for consumption for a certain amount of time, even if no receiver is currently connected.
Point-to-point channels are often used for loosely coupled message transmission, though there are two other common uses. First, it can support horizontal scaling of message processing on the receiver side. Depending on the message load in the channel, the number of receiver processes can be elastically adjusted to cope with the load as needed. The queue acts as a buffering load balancer. Second, it can flatten peak loads of messages and prevent your receivers from being flooded when you can’t scale out fast enough or you don’t want additional scaling.
Integration scenarios
In this section, we apply these fundamental patterns to AWS messaging services. The code examples are written in Java, but only by author preference. You can implement the same integration scenarios in C++, .NET, Node.js, Python, Ruby, Go, and other programming languages that AWS provides an SDK and an Apache Active MQ client library is available for.
Point-to-point channels: one-way messaging
The diagrams in the following subsections show the principle of one-way messaging for point-to-point channels, using Amazon MQ queues and Amazon SQS queues. The sender produces a message and sends it into a queue, and the receiver consumes the message from the queue for processing. For traditional messaging (that is, Amazon MQ), the senders and consumers can use protocols such as JMS or AMQP. For cloud-native messaging, they can use the Amazon SQS API.
Traditional messaging
To follow this example, open the Amazon MQ console and create a broker. In the following diagram we see the above explained components for the traditional messaging scenario: A sender sends messages into an Amazon MQ queue, a receiver consumes messages from that queue.
In the following code example, sender and receiver are using the Apache Active MQ client library and the standard Java messaging service (JMS) API to send and receive messages to and from an Amazon MQ queue. You can run the code on every Amazon compute service, your on-premises data center, or your personal computer. For simplicity, the code launches sender and receiver in the same Java virtual machine (JVM).
public class PointToPointOneWayTraditional {
public static void main(String... args) throws Exception {
ActiveMQSslConnectionFactory connFact = new ActiveMQSslConnectionFactory("failover:(ssl://<broker-1>.amazonaws.com:61617,ssl://<broker-2>.amazonaws.com:61617)");
connFact.setConnectResponseTimeout(10000);
Connection conn = connFact.createConnection("user", "password");
conn.setClientID("PointToPointOneWayTraditional");
conn.start();
new Thread(new Receiver(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Queue.PointToPoint.OneWay.Traditional")).start();
new Thread(new Sender(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Queue.PointToPoint.OneWay.Traditional")).start();
}
public static class Sender implements Runnable {
private Session session;
private String destination;
public Sender(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
try {
MessageProducer messageProducer = session.createProducer(session.createQueue(destination));
long counter = 0;
while (true) {
TextMessage message = session.createTextMessage("Message " + ++counter);
message.setJMSMessageID(UUID.randomUUID().toString());
messageProducer.send(message);
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
public static class Receiver implements Runnable, MessageListener {
private Session session;
private String destination;
public Receiver(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
try {
MessageConsumer consumer = session.createConsumer(session.createQueue(destination));
consumer.setMessageListener(this);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public void onMessage(Message message) {
try {
System.out.println(String.format("received message '%s' with message id '%s'", ((TextMessage) message).getText(), message.getJMSMessageID()));
message.acknowledge();
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
}Cloud-native messaging
To follow this example, open the Amazon SQS console and create a standard SQS queue, using the queue name P2POneWayCloudNative. In the following diagram we see the above explained components for the cloud-native messaging scenario: A sender sends messages into an Amazon SQS queue, a receiver consumes messages from that queue.
In the sample code below, the example sender is using the AWS SDK for Java to send messages to an Amazon SQS queue, running in an endless loop. You can run the code on every Amazon compute service, your on-premises data center, or your personal computer.
public class PointToPointOneWayCloudNative {
public static void main(String... args) throws Exception {
final AmazonSQS sqs = AmazonSQSClientBuilder.standard().build();
new Thread(new Sender(sqs, "https://sqs.<region>.amazonaws.com/<account-number>/P2POneWayCloudNative")).start();
}
public static class Sender implements Runnable {
private AmazonSQS sqs;
private String destination;
public Sender(AmazonSQS sqs, String destination) {
this.sqs = sqs;
this.destination = destination;
}
public void run() {
long counter = 0;
while (true) {
sqs.sendMessage(
new SendMessageRequest()
.withQueueUrl(destination)
.withMessageBody("Message " + ++counter)
.addMessageAttributesEntry("MessageID", new MessageAttributeValue().withDataType("String").withStringValue(UUID.randomUUID().toString())));
}
}
}
}We implement the receiver below in a serverless manner as an AWS Lambda function, using Amazon SQS as the event source. The name of the SQS queue is configured outside the function’s code, which is why it doesn’t appear in this code example.
public class Receiver implements RequestHandler<SQSEvent, Void> {
@Override
public Void handleRequest(SQSEvent request, Context context) {
for (SQSEvent.SQSMessage message: request.getRecords()) {
System.out.println(String.format("received message '%s' with message id '%s'", message.getBody(), message.getMessageAttributes().get("MessageID").getStringValue()));
}
return null;
}
}If this approach is new to you, you can find more details in AWS Lambda Adds Amazon Simple Queue Service to Supported Event Sources. Using Lambda comes with a number of benefits. For example, you don’t have to manage the compute environment for the receiver, and you can use an event (or push) model instead of having to poll for new messages.
Point-to-point channels: request-response messaging
In addition to the one-way scenario, we have a return channel option. We would now call the involved processes rather than the requester and responder. The requester sends a message into the request queue, and the responder sends the response into the response queue. Remember that the requester enriches the message with a return address (the name of the response queue) so that the responder knows where to send the response to. The requester also sends a correlation ID that the responder copies into the response message so that the requester can match the incoming response with a request.
Traditional messaging
In this example, we reuse the Amazon MQ broker that we set up earlier. In the following diagram we see the above explained components for the traditional messaging scenario, using an Amazon MQ queue each for the request messages and for the response messages.
Using Amazon MQ, we don’t have to create queues explicitly because they’re implicitly created as needed when we start sending messages to them. This example is similar to the point-to-point one-way traditional example.
public class PointToPointRequestResponseTraditional {
public static void main(String... args) throws Exception {
ActiveMQSslConnectionFactory connFact = new ActiveMQSslConnectionFactory("failover:(ssl://<broker-1>.amazonaws.com:61617,ssl://<broker-2>.amazonaws.com:61617)");
connFact.setConnectResponseTimeout(10000);
Connection conn = connFact.createConnection("user", "password");
conn.setClientID("PointToPointRequestResponseTraditional");
conn.start();
new Thread(new Responder(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Queue.PointToPoint.RequestResponse.Traditional")).start();
new Thread(new Requester(conn.createSession(false, Session.CLIENT_ACKNOWLEDGE), "Queue.PointToPoint.RequestResponse.Traditional")).start();
}
public static class Requester implements Runnable {
private Session session;
private String destination;
public Requester(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
MessageProducer messageProducer = null;
try {
messageProducer = session.createProducer(session.createQueue(destination));
long counter = 0;
while (true) {
TemporaryQueue replyTo = session.createTemporaryQueue();
String correlationId = UUID.randomUUID().toString();
TextMessage message = session.createTextMessage("Message " + ++counter);
message.setJMSMessageID(UUID.randomUUID().toString());
message.setJMSCorrelationID(correlationId);
message.setJMSReplyTo(replyTo);
messageProducer.send(message);
MessageConsumer consumer = session.createConsumer(replyTo, "JMSCorrelationID='" + correlationId + "'");
try {
Message receivedMessage = consumer.receive(5000);
System.out.println(String.format("received message '%s' with message id '%s'", ((TextMessage) receivedMessage).getText(), receivedMessage.getJMSMessageID()));
receivedMessage.acknowledge();
} finally {
if (consumer != null) {
consumer.close();
}
}
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
public static class Responder implements Runnable, MessageListener {
private Session session;
private String destination;
public Responder(Session session, String destination) {
this.session = session;
this.destination = destination;
}
public void run() {
try {
MessageConsumer consumer = session.createConsumer(session.createQueue(destination));
consumer.setMessageListener(this);
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
public void onMessage(Message message) {
try {
String correlationId = message.getJMSCorrelationID();
Destination replyTo = message.getJMSReplyTo();
TextMessage responseMessage = session.createTextMessage(((TextMessage) message).getText() + " with CorrelationID " + correlationId);
responseMessage.setJMSMessageID(UUID.randomUUID().toString());
responseMessage.setJMSCorrelationID(correlationId);
MessageProducer messageProducer = session.createProducer(replyTo);
try {
messageProducer.send(responseMessage);
message.acknowledge();
} finally {
if (messageProducer != null) {
messageProducer.close();
}
}
} catch (JMSException e) {
throw new RuntimeException(e);
}
}
}
}Cloud-native messaging
Open the Amazon SQS console and create two standard SQS queues using the queue names P2PReqRespCloudNative and P2PReqRespCloudNative-Resp. In the following diagram we see the above explained components for the cloud-native scenario, using an Amazon SQS queue each for the request messages and for the response messages.
The following example requester is almost identical to the point-to-point one-way cloud-native example sender. It also provides a reply-to address and a correlation ID.
public class PointToPointRequestResponseCloudNative {
public static void main(String... args) throws Exception {
final AmazonSQS sqs = AmazonSQSClientBuilder.standard().build();
new Thread(new Requester(sqs, "https://sqs.<region>.amazonaws.com/<account-number>/P2PReqRespCloudNative", "https://sqs.<region>.amazonaws.com/<account-number>/P2PReqRespCloudNative-Resp")).start();
}
public static class Requester implements Runnable {
private AmazonSQS sqs;
private String destination;
private String replyDestination;
private Map<String, SendMessageRequest> inflightMessages = new ConcurrentHashMap<>();
public Requester(AmazonSQS sqs, String destination, String replyDestination) {
this.sqs = sqs;
this.destination = destination;
this.replyDestination = replyDestination;
}
public void run() {
long counter = 0;
while (true) {
String correlationId = UUID.randomUUID().toString();
SendMessageRequest request = new SendMessageRequest()
.withQueueUrl(destination)
.withMessageBody("Message " + ++counter)
.addMessageAttributesEntry("CorrelationID", new MessageAttributeValue().withDataType("String").withStringValue(correlationId))
.addMessageAttributesEntry("ReplyTo", new MessageAttributeValue().withDataType("String").withStringValue(replyDestination));
sqs.sendMessage(request);
inflightMessages.put(correlationId, request);
ReceiveMessageResult receiveMessageResult = sqs.receiveMessage(
new ReceiveMessageRequest()
.withQueueUrl(replyDestination)
.withMessageAttributeNames("CorrelationID")
.withMaxNumberOfMessages(5)
.withWaitTimeSeconds(2));
for (Message receivedMessage : receiveMessageResult.getMessages()) {
System.out.println(String.format("received message '%s' with message id '%s'", receivedMessage.getBody(), receivedMessage.getMessageId()));
String receivedCorrelationId = receivedMessage.getMessageAttributes().get("CorrelationID").getStringValue();
SendMessageRequest originalRequest = inflightMessages.remove(receivedCorrelationId);
System.out.println(String.format("Corresponding request message '%s'", originalRequest.getMessageBody()));
sqs.deleteMessage(
new DeleteMessageRequest()
.withQueueUrl(replyDestination)
.withReceiptHandle(receivedMessage.getReceiptHandle()));
}
}
}
}
}The following example responder is almost identical to the point-to-point one-way cloud-native example receiver. It also creates a message and sends it back to the reply-to address provided in the received message.
public class Responder implements RequestHandler<SQSEvent, Void> {
private final AmazonSQS sqs = AmazonSQSClientBuilder.standard().build();
@Override
public Void handleRequest(SQSEvent request, Context context) {
for (SQSEvent.SQSMessage message: request.getRecords()) {
System.out.println(String.format("received message '%s' with message id '%s'", message.getBody(), message.getMessageId()));
String correlationId = message.getMessageAttributes().get("CorrelationID").getStringValue();
String replyTo = message.getMessageAttributes().get("ReplyTo").getStringValue();
System.out.println(String.format("sending message with correlation id '%s' to '%s'", correlationId, replyTo));
sqs.sendMessage(
new SendMessageRequest()
.withQueueUrl(replyTo)
.withMessageBody(message.getBody() + " with CorrelationID " + correlationId)
.addMessageAttributesEntry("CorrelationID", new MessageAttributeValue().withDataType("String").withStringValue(correlationId)));
}
return null;
}
}Go build!
We look forward to hearing about what you build and will continue innovating our services on your behalf.
Additional resources
- Amazon MQ – Managed Message Broker Service for ActiveMQ
- Running ActiveMQ in a Hybrid Cloud Environment with Amazon MQ
- Amazon MQ Workshop on the GitHub website
- Invoking AWS Lambda from Amazon MQ
- AWS Lambda Adds Amazon Simple Queue Service to Supported Event Sources
What’s next?
We have introduced the first fundamental EIPs and shown how you can apply them to the AWS messaging services. If you are keen to dive deeper, continue reading with the second part of this series, where we will cover publish-subscribe messaging.