Simple Message Queue (formerly MNS):Transmit oversized messages

Message flow

1. The producer checks whether the message body exceeds 64 KB.

2. If it does, the producer uploads the body to an OSS bucket.

3. The producer sends the OSS object reference to the SMQ queue.

4. The consumer reads the message from the queue and checks whether the content is an OSS object reference.

5. If it is, the consumer downloads the object from OSS and returns the full message body to the application.

Usage notes

• Bandwidth: Oversized messages consume significant network bandwidth. Make sure both the producer and consumer environments have enough bandwidth for the payload sizes involved.

• Retries: Large message transfers are sensitive to network jitter. Implement retry logic in your application to handle transient failures.

Prerequisites

Before you begin, make sure that you have:

• SMQ SDK for Java installed. For more information, see Install SDK for Java

• An endpoint and access credential configured. For more information, see Configure endpoints and access credentials

Sample code

The following example uses MNSExtendedClient to send and receive both regular and oversized messages. The extended client wraps the standard SMQ client and routes oversized messages through OSS based on the configured payloadSizeThreshold.

Download the complete source: LargeMessageDemo.java

Replace the following placeholders with your actual values:

package com.aliyun.mns.sample.scenarios.largeMessage;

import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.CloudTopic;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.common.utils.ServiceSettings;
import com.aliyun.mns.model.Message;
import com.aliyun.mns.sample.scenarios.largeMessage.service.MNSExtendedClient;
import com.aliyun.mns.sample.scenarios.largeMessage.service.bean.MNSExtendedConfiguration;
import com.aliyun.mns.sample.scenarios.largeMessage.service.impl.MNSExtendedClientImpl;
import com.aliyun.mns.sample.utils.ReCreateUtil;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.common.auth.CredentialsProviderFactory;
import com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider;
import com.aliyuncs.exceptions.ClientException;
import org.junit.Assert;

public class LargeMessageDemo {

    private final static String OSS_ENDPOINT = "oss-cn-XXX.aliyuncs.com";
    private final static String OSS_BUCKET_NAME = "mns-test-XXXXX-bucket";
    private final static String MNS_QUEUE_NAME = "test-largeMessage-queue";
    private final static String MNS_TOPIC_NAME = "test-largeMessage-topic";
    /**
     * In this example, messages whose size is larger than 4 KB are sent to OSS.
     */
    private final static Long payloadSizeThreshold = 4L;

    public static void main(String[] args) throws ClientException {
        // Obtain access credentials from environment variables.
        EnvironmentVariableCredentialsProvider credentialsProvider =
            CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider();

        // Create an OSS client.
        OSS ossClient = new OSSClientBuilder().build(OSS_ENDPOINT, credentialsProvider);

        // Create an SMQ client.
        // Configure the AccessKey ID and AccessKey secret as environment variables.
        CloudAccount account = new CloudAccount(ServiceSettings.getMNSAccountEndpoint());
        MNSClient client = account.getMNSClient();
        CloudQueue queue = client.getQueueRef(MNS_QUEUE_NAME);
        CloudTopic cloudTopic = client.getTopicRef(MNS_TOPIC_NAME);

        // Re-create the queue and topic for a clean demo environment.
        ReCreateUtil.reCreateQueue(client, MNS_QUEUE_NAME);
        ReCreateUtil.reCreateTopic(client, MNS_TOPIC_NAME);

        // Configure the extended client with OSS and the size threshold.
        MNSExtendedConfiguration configuration = new MNSExtendedConfiguration()
            .setOssClient(ossClient).setOssBucketName(OSS_BUCKET_NAME)
            .setMNSQueue(queue)
            .setMNSTopic(cloudTopic)
            .setPayloadSizeThreshold(payloadSizeThreshold);

        MNSExtendedClient mnsExtendedClient = new MNSExtendedClientImpl(configuration);

        // Send and receive a regular-sized message.
        Message normalMessage = new Message();
        normalMessage.setMessageBodyAsRawString("1");
        mnsExtendedClient.sendMessage(normalMessage);
        Message message = mnsExtendedClient.receiveMessage(10);
        System.out.println("[normal]ReceiveMsg:" + message.getMessageBodyAsRawString());
        mnsExtendedClient.deleteMessage(message.getReceiptHandle());

        // Send and receive an oversized message.
        String largeMsgBody = "largeMessage";
        Assert.assertTrue(largeMsgBody.getBytes().length > payloadSizeThreshold);

        Message largeMessage = new Message();
        largeMessage.setMessageBodyAsRawString(largeMsgBody);

        mnsExtendedClient.sendMessage(largeMessage);
        Message receiveMessage = mnsExtendedClient.receiveMessage(10);
        System.out.println("[large]ReceiveMsg:" + receiveMessage.getMessageBodyAsRawString());
        mnsExtendedClient.deleteMessage(receiveMessage.getReceiptHandle());

        client.close();
        ossClient.shutdown();
    }
}

Related topics

• Install SDK for Java -- Set up the SMQ SDK in your project.

• Configure endpoints and access credentials -- Configure SMQ endpoints and authentication.