Simple Message Queue (formerly MNS):Receive messages from a queue

Configuration

Create or update ${"user.home"}/.aliyun-mns.properties with your endpoint and Base64 encoding preference:

mns.endpoint=http://xxxxxxx
mns.msgBodyBase64Switch=true/false

Store your AccessKey ID and AccessKey secret as environment variables following Alibaba Cloud security best practices.

Code

The sample loops continuously, alternating between a single receive and a long polling batch receive. Each received message is printed and then deleted.

package com.aliyun.mns.sample.queue;

import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.common.ClientException;
import com.aliyun.mns.common.ServiceException;
import com.aliyun.mns.common.ServiceHandlingRequiredException;
import com.aliyun.mns.common.utils.ServiceSettings;
import com.aliyun.mns.model.Message;
import java.util.List;

/**
 * Prerequisites:
 * 1. Set the AccessKey ID and AccessKey secret as environment variables following Alibaba Cloud security best practices.
 * 2. Create ${"user.home"}/.aliyun-mns.properties with the following content:
 *           mns.endpoint=http://xxxxxxx
 *           mns.msgBodyBase64Switch=true/false
 */
public class ReceiveMessageDemo {
    /**
     * Whether to decode the message body from Base64.
     */
    private static final Boolean IS_BASE64 = Boolean.valueOf(ServiceSettings.getMNSPropertyValue("msgBodyBase64Switch","false"));

    public static void main(String[] args) {
        String queueName = "cloud-queue-demo";

        // Set the AccessKey ID and AccessKey secret as environment variables following Alibaba Cloud security best practices.
        CloudAccount account = new CloudAccount(ServiceSettings.getMNSAccountEndpoint());
        // Initialize the client once and reuse it.
        MNSClient client = account.getMNSClient();
        CloudQueue queue = client.getQueueRef(queueName);

        // Receive and process messages in a loop.
        loopReceive(queue, client);

        // Close the client after processing.
        client.close();
    }

    private static void loopReceive(CloudQueue queue, MNSClient client) {
        while (true) {
            try {
                // Single receive: basic method.
                singleReceive(queue);

                // Long polling batch receive: recommended for production.
                longPollingBatchReceive(queue);
            } catch (ClientException ce) {
                System.out.println("Network connection error between client and SMQ service. "
                    + "Check your network and DNS settings.");
                // Retry on transient network errors.
            } catch (ServiceException se) {
                if (se.getErrorCode().equals("QueueNotExist")) {
                    System.out.println("Queue does not exist. Create the queue before calling ReceiveMessage.");
                    client.close();
                    return;
                } else if (se.getErrorCode().equals("TimeExpired")) {
                    System.out.println("Request expired. Check your local machine clock for time synchronization issues.");
                    return;
                }
                // Retry on transient service errors.
            } catch (Exception e) {
                System.out.println("Unexpected error: " + e.getMessage());
            }

        }
    }

    private static void longPollingBatchReceive(CloudQueue queue) throws ServiceHandlingRequiredException {

        System.out.println("=============start longPollingBatchReceive=============");

        // Maximum number of messages to receive per call.
        int batchSize = 15;
        // Long polling wait time in seconds.
        int waitSeconds = 15;

        List<Message> messages = queue.batchPopMessage(batchSize, waitSeconds);
        if (messages != null && messages.size() > 0) {

            for (Message message : messages) {
                printMsgAndDelete(queue,message);
            }
        }

        System.out.println("=============end longPollingBatchReceive=============");

    }

    private static void singleReceive(CloudQueue queue) throws ServiceHandlingRequiredException {
        System.out.println("=============start singleReceive=============");

        Message popMsg = queue.popMessage();
        printMsgAndDelete(queue, popMsg);

        System.out.println("=============end singleReceive=============");
    }

    private static void printMsgAndDelete(CloudQueue queue, Message popMsg) throws ServiceHandlingRequiredException {
        if (popMsg != null) {
            System.out.println("message handle: " + popMsg.getReceiptHandle());
            System.out.println("message body: " + (IS_BASE64 ? popMsg.getMessageBody() : popMsg.getMessageBodyAsRawString()));
            System.out.println("message id: " + popMsg.getMessageId());
            System.out.println("message dequeue count:" + popMsg.getDequeueCount());
            // Add your message processing logic here.

            // Delete the message after successful consumption.
            queue.deleteMessage(popMsg.getReceiptHandle());
            System.out.println("delete message successfully.\n");
        }
    }

}