Simple Message Queue (anciennement MNS) impose une limite de taille de 64 Ko par message. Pour envoyer des messages dépassant 64 Ko sans les fragmenter, stockez le corps du message dans Object Storage Service (OSS) et transmettez une référence à l'objet via la file d'attente. Cette approche, appelée modèle claim-check, préserve votre flux de messagerie tandis qu'OSS gère la charge utile.
Le SDK SMQ fournit MNSExtendedClient, qui gère de manière transparente les téléversements et téléchargements vers OSS. Après configuration, utilisez les mêmes API sendMessage et receiveMessage pour les messages standards et les messages volumineux.
Flux de messages
Le producteur vérifie si le corps du message dépasse 64 Ko.
Si c'est le cas, il téléverse le corps vers un compartiment OSS.
Le producteur envoie la référence de l'objet OSS à la file d'attente SMQ.
Le consommateur lit le message depuis la file d'attente et vérifie si le contenu est une référence à un objet OSS.
Si tel est le cas, il télécharge l'objet depuis OSS et renvoie le corps complet du message à l'application.
Notes d'utilisation
Bande passante : Les messages volumineux consomment une bande passante réseau importante. Assurez-vous que les environnements du producteur et du consommateur disposent d'une bande passante suffisante pour les tailles de charge utile concernées.
Nouvelles tentatives : Les transferts de messages volumineux sont sensibles aux variations du réseau. Implémentez une logique de nouvelle tentative dans votre application pour gérer les échecs temporaires.
Prérequis
Avant de commencer, assurez-vous d'avoir :
Installé le SDK SMQ pour Java. Pour plus d'informations, consultez Installer le SDK pour Java
Configuré un endpoint et des identifiants d'accès. Pour plus d'informations, consultez Configurer les endpoints et les identifiants d'accès
Exemple de code
L'exemple suivant utilise MNSExtendedClient pour envoyer et recevoir des messages standards et volumineux. Le client étendu encapsule le client SMQ standard et achemine les messages volumineux via OSS en fonction du seuil payloadSizeThreshold configuré.
Téléchargez le code source complet : LargeMessageDemo.java
Remplacez les espaces réservés suivants par vos valeurs réelles :
| Espace réservé | Description | Exemple |
|---|---|---|
oss-cn-XXX.aliyuncs.com |
Endpoint OSS pour votre région | oss-cn-hangzhou.aliyuncs.com |
mns-test-XXXXX-bucket |
Nom du compartiment OSS pour stocker les charges utiles des messages volumineux | mns-prod-msg-bucket |
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();
}
}Rubriques connexes
Installer le SDK pour Java -- Configurez le SDK SMQ dans votre projet.
Configurer les endpoints et les identifiants d'accès -- Configurez les endpoints SMQ et l'authentification.