Setup Amazon MQ with Java Application

Overview

Messaging mediates communication among different parts of distributed systems or applications, eliminating the need to understand the application’s design to exchange messages. There are numerous open-source message broker tools available in the market which can be used for messaging applications.

Today, we will walk through one of AWS’s Managed Message Broker services, i.e., Amazon MQ. Explore Amazon MQ, its benefits, and its advantages over Amazon SQS. Also, we will see how to use it in an application.

Introduction to Amazon MQ

Amazon MQ is one of the managed services offered by AWS. It is a message broker for RabbitMQ and Apache ActiveMQ. As Amazon MQ is a managed service, it reduces the burden of operational responsibilities, which involve setting up, maintaining, and provisioning message brokers. With Amazon MQ, one can quickly move to AWS without worrying about rewriting the code as it can connect to existing applications with industry-standard APIs and protocols. Amazon MQ can be set up with a few clicks and supports multiple languages and platforms.

What is Amazon MQ Broker?

We can create a message Broker according to our requirements. A broker can be a single-instance broker or an active/standby broker.

• Single Instance Broker contains one Broker in one availability. It communicates with the application and AWS storage location. It is used for development and testing.
• Active/Standby Broker contains two brokers in two different availability zones for high availability. It will communicate with the application and shared storage location. It can be used for automatic failover. For both modes, Amazon MQ provides data replication across AZs.

Why ActiveMQ, Why not a database to store messages?

Yes, we can use the database to store messages and process them. But, whenever communication happens between two applications, once the message is received, it should be processed and deleted, i.e., for each message entry, there should be an insert and delete operation. It will work smoothly for a few messages without affecting the performance of the database. If we want to handle thousands of messages between these applications, it may affect database performance and tend to fail.

Amazon MQ acts as a message broker service for Apache ActiveMQ, which can be used to handle this use case. It will run the maintenance of ActiveMQ brokers and perform optimizations to avoid overhead. It offers a mechanism of pushing messages to consumers rather than that consumer polling for a new message. Messaging mediates intercommunication amongst distinct parts of distributed systems or applications, eliminating the want the notion of the application’s construct to swap messages. Therefore, this reduces the latency involved in processing new messages.

Benefits of Amazon MQ:

• Application Integration: It provides a way to integrate applications with different languages and different operating systems to communicate with each other.
• Reliability: No need for availability of producers or consumers at the same time. Provides queue to manage any number of requests at any point in time.
• Asynchronous allows one application to send a message and continue with other tasks instead of waiting for a response.
• Decoupled: Since the queue lies between the application and communication is carried through the queue, the failure of one system will not affect other applications.
• Cost Reduction: You need to pay only for the storage and broker instance you use.

Amazon MQ vs Amazon SQS

Creating Broker and connecting it to Java Application

We will create two components i.e., a Producer and a Consumer. Here, the Producer will click to the Broker, create a queue and send a message, and the Consumer will connect to the Broker, create a queue and receive a message sent by the Producer.

Our goal is to create multiple producers with many connections that always use the JMS connection pooling factory, i.e., the PooledConnectionFactory class. However, message consumers should use the JMS connection factory, i.e., ActiveMQConnectionFactory class.

1. First, open the Amazon MQ console by clicking on ‘Get Started.’
   
2. Select the type of engine you want. We are setting Apache ActiveMQ
   
3. Setup Deployment Mode and Storage Type. We are not selecting Blueprints for a network for this example.
   
4. Give Broker a Friendly Name, Select Instance type, and authentication methods. Here, we are using Simple Authentication.
   
   It usually takes around 20 mins to create a Broker.
5. Once the Broker is created, go to connections details and copy the Openwire endpoint.
   
6. Create a Java application named AmazonMQConnection. Create two classes that will act as two components in this application. One is the Producer class, and another one is the Consumer.
   
   Producer Class

   Java

   import org.apache.activemq.jms.pool.PooledConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Producer { public static void main(String[] args) throws JMSException { //Enter the Openwire link of the Amazon MQ broker created earlier ActiveMQConnectionFactory connFactory = new ActiveMQConnectionFactory("link"); // Specify the master username and password. connFactory.setUserName("username"); connFactory.setPassword("password"); PooledConnectionFactory pooledConnFactory = new PooledConnectionFactory(); pooledConnFactory.setConnectionFactory(connFactory); pooledConnFactory.setMaxConnections(10); Connection producerConnection = pooledConnFactory.createConnection(); producerConnection.start(); Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination producerDest = producerSession.createQueue("MesgQueue"); MessageProducer producer = producerSession.createProducer(producerDest); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); String text = "Message from Producer using Amazon MQ"; TextMessage producerMessage = producerSession.createTextMessage(text); producer.send(producerMessage); System.out.println("Message sent successfully"); producer.close(); producerSession.close(); producerConnection.close(); } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40

   import org.apache.activemq.jms.pool.PooledConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Producer { public static void main(String[] args) throws JMSException {          //Enter the Openwire link of the Amazon MQ broker created earlier ActiveMQConnectionFactory connFactory = new ActiveMQConnectionFactory("link");        // Specify the master username and password.        connFactory.setUserName("username");        connFactory.setPassword("password");               PooledConnectionFactory pooledConnFactory = new PooledConnectionFactory();        pooledConnFactory.setConnectionFactory(connFactory);        pooledConnFactory.setMaxConnections(10);        Connection producerConnection = pooledConnFactory.createConnection();        producerConnection.start();        Session producerSession =        producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);        Destination producerDest = producerSession.createQueue("MesgQueue");        MessageProducer producer = producerSession.createProducer(producerDest);        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);        String text = "Message from Producer using Amazon MQ";        TextMessage producerMessage = producerSession.createTextMessage(text);        producer.send(producerMessage);        System.out.println("Message sent successfully");        producer.close();        producerSession.close();        producerConnection.close();       } }

   Consumer Class

   Java

   import org.apache.activemq.jms.pool.PooledConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Consumer { public static void main(String[] args) throws JMSException { //Enter the Openwire link of the Amazon MQ broker created earlier ActiveMQConnectionFactory connFactory = new ActiveMQConnectionFactory("link"); // Specify the master username and password. connFactory.setUserName("username"); connFactory.setPassword("password"); Connection consumerConn = connFactory.createConnection(); consumerConn.start(); Session consumerSession = consumerConn.createSession(false,Session.AUTO_ACKNOWLEDGE); Destination consumerDest = consumerSession.createQueue("MesgQueue"); MessageConsumer consumer = consumerSession.createConsumer(consumerDest); Message consumerMesg = consumer.receive(1000); TextMessage consumerTextMesg = (TextMessage) consumerMesg; System.out.println("Received Message is : " + consumerTextMesg.getText()); consumer.close(); consumerSession.close(); consumerConn.close(); } }

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34

   import org.apache.activemq.jms.pool.PooledConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class Consumer { public static void main(String[] args) throws JMSException {   //Enter the Openwire link of the Amazon MQ broker created earlier ActiveMQConnectionFactory connFactory = new ActiveMQConnectionFactory("link");        // Specify the master username and password.        connFactory.setUserName("username");        connFactory.setPassword("password");                 Connection consumerConn = connFactory.createConnection();        consumerConn.start(); Session consumerSession =                                                              consumerConn.createSession(false,Session.AUTO_ACKNOWLEDGE);        Destination consumerDest = consumerSession.createQueue("MesgQueue");        MessageConsumer consumer = consumerSession.createConsumer(consumerDest);        Message consumerMesg = consumer.receive(1000);        TextMessage consumerTextMesg = (TextMessage) consumerMesg;        System.out.println("Received Message is : " + consumerTextMesg.getText());        consumer.close();        consumerSession.close();        consumerConn.close();       } }

    

7. Convert Java application into Maven project. Update pom.xml file with the following code. This will allow the application to work with ActiveMQ.
   XHTML

   <dependencies> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-client</artifactId> <version>5.16.4</version> </dependency> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-pool </artifactId> <version>5.16.4</version> </dependency> </dependencies>

   1 2 3 4 5 6 7 8 9 10 11 12 13

   <dependencies>         <dependency>             <groupId>org.apache.activemq</groupId>             <artifactId>activemq-client</artifactId>             <version>5.16.4</version>         </dependency>         <dependency>             <groupId>org.apache.activemq</groupId>             <artifactId>activemq-pool             </artifactId>             <version>5.16.4</version>         </dependency> </dependencies>

8.  Run Producer class to send a message and Run Consumer class to receive the message sent by the Producer.

Conclusion

In this blog, we saw how easy it is to set up Amazon MQ. We have connected two components in an application using the Amazon MQ service. Message Queuing Services are used extensively to decouple applications or communication between two applications.