Simple Message Queue (formerly MNS):特大メッセージの送信

背景情報

SMQ キューには、最大 64 KB のメッセージを送信できます。ほとんどの場合、これはメッセージ送信の基本的な要件を満たすことができます。ただし、特殊なシナリオでは、64 KB を超えるメッセージを送信するために、メッセージを分割する必要があります。メッセージを分割したくない場合は、SMQ と OSS を一緒に使用してメッセージを送信できます。

次の例は、メッセージを分割することなく、OSS を使用して 64 KB を超えるメッセージを送信する方法を示しています。

ソリューション

1. メッセージプロデューサーが SMQ キューにメッセージを送信する前に、メッセージ本文のサイズが 64 KB を超える場合、メッセージプロデューサーはメッセージ本文をオブジェクトとして OSS にアップロードします。

2. メッセージプロデューサーは、オブジェクトを SMQ キューに送信します。

3. メッセージコンシューマーは、SMQ キューからメッセージを読み取り、メッセージの内容が OSS オブジェクトであるかどうかを確認します。

4. メッセージの内容が OSS オブジェクトである場合、メッセージコンシューマーは OSS オブジェクトをダウンロードし、メッセージの内容を上位層アプリケーションに返します。

次の図は、メッセージ送信プロセスを示しています。

使用上の注意

• 特大メッセージは、主にネットワーク帯域幅を消費します。このソリューションを使用して特大メッセージを送信する場合、メッセージプロデューサーとメッセージコンシューマーの両方のネットワーク帯域幅が要件を満たしている必要があります。

• 特大メッセージの送信には時間がかかり、ネットワークのジッターの影響を受ける可能性があります。エラーが発生した場合は、上位層で再試行することをお勧めします。

前提条件

• SMQ SDK for Java がインストールされていること。詳細については、SDK for Java のインストールをご参照ください。

• エンドポイントとアクセス認証情報が構成されていること。詳細については、エンドポイントとアクセス認証情報の構成をご参照ください。

サンプルコード

サンプルコードのダウンロード方法については、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";
    /**
     * この例では、サイズが 4 KB を超えるメッセージは OSS に送信されます。
     */
    private final static Long payloadSizeThreshold = 4L;

    public static void main(String[] args) throws ClientException {
        // 環境変数からアクセス認証情報を取得します。
        EnvironmentVariableCredentialsProvider credentialsProvider = CredentialsProviderFactory.newEnvironmentVariableCredentialsProvider();

        // OSSClient インスタンスを作成します。
        OSS ossClient = new OSSClientBuilder().build(OSS_ENDPOINT, credentialsProvider);

        // MNS インスタンスを作成します。
        // Alibaba Cloud の仕様に基づいて、環境内で AccessKey ID と AccessKey シークレットを構成します。
        CloudAccount account = new CloudAccount(ServiceSettings.getMNSAccountEndpoint());
        MNSClient client = account.getMNSClient();
        CloudQueue queue = client.getQueueRef(MNS_QUEUE_NAME);
        CloudTopic cloudTopic = client.getTopicRef(MNS_TOPIC_NAME);

        //reCreate
        ReCreateUtil.reCreateQueue(client,MNS_QUEUE_NAME);
        ReCreateUtil.reCreateTopic(client,MNS_TOPIC_NAME);

        // 特大メッセージのキューの属性を構成します。
        MNSExtendedConfiguration configuration = new MNSExtendedConfiguration()
            .setOssClient(ossClient).setOssBucketName(OSS_BUCKET_NAME)
            .setMNSQueue(queue)
            .setMNSTopic(cloudTopic)
            .setPayloadSizeThreshold(payloadSizeThreshold);

        MNSExtendedClient mnsExtendedClient = new MNSExtendedClientImpl(configuration);

        // 通常サイズのメッセージをキューに送信します。
        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());

        // 特大メッセージをキューに送信します。
        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();
    }
}