O Simple Message Queue (anteriormente MNS) limita cada mensagem a 64 KB. Para enviar mensagens maiores sem dividi-las, armazene o corpo da mensagem no Object Storage Service (OSS) e passe apenas uma referência ao objeto pela fila. Esse padrão Claim-Check permite que produtores e consumidores troquem cargas úteis de qualquer tamanho, mantendo-se dentro do limite de tamanho da fila.
Como funciona
O produtor verifica o tamanho do corpo da mensagem. Se o corpo exceder 64 KB, ele o envia como um objeto para o OSS.
O produtor envia uma referência ao objeto do OSS (não o corpo completo) para a fila do SMQ.
O consumidor lê a mensagem da fila e verifica se o corpo é uma referência a um objeto do OSS.
Se o corpo for uma referência a um objeto do OSS, o consumidor baixa o objeto do OSS e retorna o corpo completo da mensagem à aplicação.
O diagrama a seguir ilustra esse fluxo de trabalho.
Observações de uso
Mensagens muito grandes consomem largura de banda significativa. Certifique-se de que tanto o produtor quanto o consumidor tenham largura de banda suficiente para os tamanhos de mensagem esperados.
Transferências de mensagens grandes demandam tempo e podem sofrer impacto de instabilidades na rede. Implemente retentativas em sua aplicação para lidar com erros transitórios.
Pré-requisitos
Antes de começar, verifique se você tem:
SDK do SMQ para Java instalado
Um endpoint e credenciais de acesso configurados
Código de exemplo
O exemplo a seguir demonstra como implementar o padrão Claim-Check com SMQ e OSS. Baixe o source completo em LargeMessageDemo.java.
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 {
// Get 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 SMQ settings.
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 normal-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();
}
}