Mengirim pesan dari satu klien produser ke beberapa klien konsumen - Simple Message Queue (formerly MNS)

Simple Message Queue (sebelumnya MNS) menyediakan model pesan yang memungkinkan beberapa klien konsumen menarik pesan dari satu klien produser. Topik ini menjelaskan cara menggunakan model ini untuk mengirim pesan dari satu klien produser ke beberapa klien konsumen.

Catatan

Dalam contoh ini, SDK untuk Java digunakan. Untuk informasi lebih lanjut tentang SDK untuk bahasa pemrograman lainnya, lihat Referensi untuk SDK Versi Baru (Direkomendasikan).

Prasyarat

IntelliJ IDEA telah diinstal. Untuk informasi lebih lanjut, lihat IntelliJ IDEA.
Anda dapat menggunakan IntelliJ IDEA atau Eclipse. Dalam contoh ini, IntelliJ IDEA digunakan.
JDK 1.8 atau yang lebih baru telah diinstal. Untuk informasi lebih lanjut, lihat Unduhan Java.
Maven 2.5 atau yang lebih baru telah diinstal. Untuk informasi lebih lanjut, lihat Mengunduh Apache Maven.

Informasi latar belakang

SMQ menyediakan model pesan berbasis antrian dan topik yang dapat memenuhi kebutuhan sebagian besar skenario bisnis.

Model pesan berbasis antrian: Satu klien mengirimkan pesan ke antrian SMQ. Beberapa klien mengonsumsi pesan dengan menarik pesan dari antrian tersebut.
Model pesan berbasis topik: Satu klien mengirimkan pesan ke topik SMQ. Pesan secara otomatis didorong dari server SMQ ke beberapa klien.

SMQ memungkinkan klien untuk mengonsumsi pesan secara real-time dengan mendorong pesan dari server SMQ ke klien. Namun, titik akhir klien konsumen terpapar untuk menerima pesan dalam skenario ini. Anda tidak dapat mengekspos titik akhir klien konsumen dalam skenario tertentu, seperti di jaringan pribadi perusahaan. Dalam hal ini, klien konsumen dapat mengonsumsi pesan dengan menarik pesan dari server SMQ. Anda dapat menggabungkan antrian dan topik SMQ untuk mengirim pesan dari satu klien ke beberapa klien konsumen tanpa mengekspos titik akhir klien konsumen.

Solusi

Buat langganan ke topik menggunakan titik akhir antrian tertentu. Kemudian, pesan dapat didorong dari topik ke antrian, dan klien konsumen dapat menarik pesan dari antrian. Dengan cara ini, Anda dapat mengirim pesan dari satu klien ke beberapa klien tanpa mengekspos titik akhir klien konsumen. Gambar berikut menunjukkan prosesnya. 消息流

Instal pustaka dependensi Java

Buat proyek Java di IntelliJ IDEA.

Tambahkan dependensi berikut ke file pom.xml untuk mengimpor pustaka dependensi Java:

<dependency>
    <groupId>com.aliyun.mns</groupId>
    <artifactId>aliyun-sdk-mns</artifactId>
    <version>1.1.9.2</version>
</dependency>

Operasi API

SDK for Java 1.1.8 menyediakan kelas CloudPullTopic untuk mendukung solusi di atas. Anda dapat memanggil operasi API berikut yang disediakan oleh MNSClient untuk membuat objek CloudPullTopic:

public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector<String> queueNameList, boolean needCreateQueue, QueueMeta queueMetaTemplate)

public CloudPullTopic createPullTopic(TopicMeta topicMeta, Vector<String> queueNameList)

Tabel berikut menjelaskan parameter yang dapat dikonfigurasi dalam pernyataan di atas.

TopicMeta: menentukan metadata dari sebuah topik.
QueueMeta: menentukan metadata dari sebuah antrian.
queueNameList: menentukan daftar antrian ke mana pesan dalam topik tertentu didorong.
needCreateQueue: menentukan apakah akan membuat antrian yang ditentukan oleh parameter queueNameList.
queueMetaTemplate: menentukan template metadata dari sebuah antrian.

Kode contoh

package doc;

// Tentukan kelas SMQ.
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.QueueMeta;
import com.aliyun.mns.model.TopicMeta;
import com.aliyun.mns.client.CloudPullTopic;
import com.aliyun.mns.model.TopicMessage;
import com.aliyun.mns.model.RawTopicMessage;
import com.aliyun.mns.model.Message;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.common.ServiceException;
import com.aliyun.mns.common.ClientException;
import java.util.Vector;

public class DemoTopicMessageBroadcast {
    public static void main(String[] args) {
      
// Dapatkan ID AccessKey dan Rahasia AccessKey dari akun Alibaba Cloud serta Titik Akhir MNS. 
    CloudAccount account = new CloudAccount(
    ServiceSettings.getMNSAccessKeyId(),
    ServiceSettings.getMNSAccessKeySecret(),
    ServiceSettings.getMNSAccountEndpoint());
    MNSClient client = account.getMNSClient();

// Buat konsumen. 
Vector<String> consumerNameList = new Vector<String>();
String consumerName1 = "consumer001";
String consumerName2 = "consumer002";
String consumerName3 = "consumer003";
consumerNameList.add(consumerName1);
consumerNameList.add(consumerName2);
consumerNameList.add(consumerName3);
QueueMeta queueMetaTemplate = new QueueMeta();
queueMetaTemplate.setPollingWaitSeconds(30);

try{

    // Buat topik. 
    String topicName = "demo-topic-for-pull";
    TopicMeta topicMeta = new TopicMeta();
    topicMeta.setTopicName(topicName);
    CloudPullTopic pullTopic = client.createPullTopic(topicMeta, consumerNameList, true, queueMetaTemplate);

    // Publikasikan pesan ke topik. 
    String messageBody = "broadcast message to all the consumers:hello the world.";
    // Jika pesan asli dikirim, gunakan getMessageBodyAsRawString untuk mengurai isi pesan. 
    TopicMessage tMessage = new RawTopicMessage();
    tMessage.setBaseMessageBody(messageBody);
    pullTopic.publishMessage(tMessage);


    // Terima pesan. 
    CloudQueue queueForConsumer1 = client.getQueueRef(consumerName1);
    CloudQueue queueForConsumer2 = client.getQueueRef(consumerName2);
    CloudQueue queueForConsumer3 = client.getQueueRef(consumerName3);

    Message consumer1Msg = queueForConsumer1.popMessage(30);
    if(consumer1Msg != null)
    {
        System.out.println("consumer1 menerima pesan:" + consumer1Msg.getMessageBodyAsRawString());
    } else{
        System.out.println("antrian kosong");
    }

    Message consumer2Msg = queueForConsumer2.popMessage(30);
    if(consumer2Msg != null)
    {
        System.out.println("consumer2 menerima pesan:" + consumer2Msg.getMessageBodyAsRawString());
    }else{
        System.out.println("antrian kosong");
    }

    Message consumer3Msg = queueForConsumer3.popMessage(30);
    if(consumer3Msg != null)
    {
        System.out.println("consumer3 menerima pesan:" + consumer3Msg.getMessageBodyAsRawString());
    }else{
        System.out.println("antrian kosong");
    }

    // Hapus topik. 
    pullTopic.delete();
} catch(ClientException ce) {
    System.out.println("Ada masalah dengan koneksi jaringan antara klien dan layanan SMQ."
            + "Silakan periksa ketersediaan jaringan dan DNS Anda.");
    ce.printStackTrace();
} catch(ServiceException se) {

    se.printStackTrace();
}

    client.close();
    }
}