Async consumer for Amazon sqs

Question

I am new to working with queues. I am able to successfully post messages and receive them synchronously However, I am now trying to async now.

The reference links provided by sqs suggests using jmsclient wrapper. And the link also mentions to use it if you already have a code that is integrated to a jms client.

http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/jmsclient.html#samples

But I am starting afresh I referred this example to send and recv messages synchronously.

https://github.com/aws/aws-sdk-java/blob/master/src/samples/AmazonSimpleQueueService/SimpleQueueServiceSample.java

Can I use the same code but implement it with a message listener? Any code examples will be appreciated.

Answer 1

There's a code sample in the section about Using JMS with Amazon SQS of the Amazon SQS Developer Guide that shows how to asynchronously receive messages using JMS.

First you implement the MessageListener interface:

class MyListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        try {
            // Cast the received message as TextMessage and print the text to screen.
            if (message != null) {
                System.out.println("Received: " + ((TextMessage) message).getText());
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

And then you set it as the MessageListener for a MessageConsumer:

// Create a consumer for the 'TestQueue'.
MessageConsumer consumer = session.createConsumer(queue);

// Instantiate and set the message listener for the consumer.
consumer.setMessageListener(new MyListener());

// Start receiving incoming messages.
connection.start();

// Wait for 1 second. The listener onMessage() method will be invoked when a message is received.
Thread.sleep(1000);

Answer 2

You can use sqslistener annotation from SpringCloud framework. If you are developing application with Spring and AWS and you are not using Spring Cloud , it is good time for you to switch.

Here is a sample code to asynchronously receive message from SQS using sqslistener annotation. A good thing is you have to almost zero configuration for using this :

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.aws.messaging.listener.SqsMessageDeletionPolicy;
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
import org.springframework.stereotype.Component;
import com.example.my.RecoverableException;

@Component
@Slf4j
public class CustomMessageQueue {

    @SqsListener(value = "${build_request_queue.name}", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
    public void receive(String message) {
        try {
            // write message processing here
        } catch (RecoverableException e) {
            // handle errors here for which message from queue should not be deleted
            // throwing an exception will make receive method fail and hence message does not get deleted
            throw e;

        } catch (Exception e) {
            // suppress exceptions for which message should be deleted.
        }
    }
}

The great thing about sqslistener annotation is its deletionPolicy. So you can decide when a message from SQS gets deleted.