Simple Message Queue (formerly MNS):Transmit oversized messages

Message flow

1. O produtor verifica se o corpo da mensagem ultrapassa 64 KB.

2. Se ultrapassar, o produtor faz o upload do corpo para um bucket do OSS.

3. O produtor envia a referência ao objeto OSS para a fila do SMQ.

4. O consumidor lê a mensagem da fila e verifica se o conteúdo é uma referência a um objeto OSS.

5. Caso seja, o consumidor baixa o objeto do OSS e retorna o corpo completo da mensagem à aplicação.

Usage notes

• Largura de banda: Mensagens de grande porte consomem largura de banda de rede significativa. Certifique-se de que os ambientes do produtor e do consumidor disponham de largura de banda suficiente para os tamanhos de payload envolvidos.

• Retentativas: Transferências de mensagens grandes são sensíveis a instabilidades de rede. Implemente lógica de retentativa na sua aplicação para lidar com falhas transitórias.

Prerequisites

Antes de começar, verifique se você possui:

• SDK do SMQ para Java instalado. Para mais informações, consulte Instale SDK for Java

• Um endpoint e uma credencial de acesso configurados. Para mais informações, consulte Configure endpoints and access credentials

Sample code

O exemplo a seguir usa MNSExtendedClient para enviar e receber mensagens regulares e de grande porte. O cliente estendido encapsula o cliente SMQ padrão e roteia mensagens de grande porte pelo OSS com base no payloadSizeThreshold configurado.

Baixe o source completo: LargeMessageDemo.java

Substitua os seguintes placeholders pelos seus valores reais:

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

• Instale SDK for Java -- Configure o SDK do SMQ no seu projeto.

• Configure endpoints and access credentials -- Configure os endpoints e a autenticação do SMQ.