ApsaraMQ for RocketMQ:Kirim dan terima pesan terurut

Klasifikasi pesan terurut

Pesan terurut diklasifikasikan menjadi jenis-jenis berikut:

• Pesan terurut global: Semua pesan dalam topik tertentu diterbitkan dan dikonsumsi dalam urutan FIFO yang ketat.

• Pesan terurut partisi: Semua pesan dalam topik tertentu didistribusikan ke partisi berbeda menggunakan kunci sharding. Pesan dalam setiap partisi dikonsumsi dalam urutan FIFO yang ketat. Kunci sharding adalah bidang kunci yang digunakan untuk mengidentifikasi partisi berbeda dan berbeda dari kunci pesan normal.

Untuk informasi lebih lanjut, lihat Pesan Terurut.

Prasyarat

• SDK untuk .NET telah diunduh. Untuk informasi lebih lanjut, lihat Catatan Rilis.

• Lingkungan telah dipersiapkan. Untuk informasi lebih lanjut, lihat Persiapkan Lingkungan.

• Sumber daya yang ingin Anda tentukan dalam kode telah dibuat di konsol ApsaraMQ for RocketMQ. Sumber daya tersebut mencakup instance, topik, dan grup konsumen. Untuk informasi lebih lanjut, lihat Buat Sumber Daya.

• Pasangan AccessKey akun Alibaba Cloud Anda telah diperoleh. Untuk informasi lebih lanjut, lihat Buat Pasangan AccessKey.

Kirim pesan terurut

Untuk informasi lebih lanjut tentang kode contoh rinci, lihat repositori kode ApsaraMQ for RocketMQ.

Kode contoh berikut menunjukkan cara mengirim pesan terurut menggunakan TCP client SDK untuk .NET:

using System;
using ons;

public class OrderProducerExampleForEx
{
    public OrderProducerExampleForEx()
    {
    }

    static void Main(string[] args) {
        // Konfigurasikan akun Anda. Anda dapat memperoleh informasi akun di Konsol Manajemen Alibaba Cloud.
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET dikonfigurasi.
        // ID AccessKey yang digunakan untuk otentikasi.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
	      // Rahasia AccessKey yang digunakan untuk otentikasi.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // ID grup konsumen yang Anda buat di konsol ApsaraMQ for RocketMQ.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "GID_example");
        // Topik yang Anda buat di konsol ApsaraMQ for RocketMQ.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
        // Titik akhir TCP. Anda dapat memperoleh titik akhir di bagian TCP Endpoint halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
        // Jalur log.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

        // Buat instance produser.
        // Catatan: Instance produser bersifat thread-safe dan dapat digunakan untuk mengirim pesan dari topik yang berbeda. Dalam kebanyakan kasus, setiap thread hanya memerlukan satu instance produser.
        OrderProducer producer = ONSFactory.getInstance().createOrderProducer(factoryInfo);

        // Mulai instance produser.
        producer.start();

        // Buat pesan.
        Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Contoh isi pesan");
        string shardingKey = "App-Test";
        for (int i = 0; i < 32; i++) {
            try
            {
                SendResultONS sendResult = producer.send(msg, shardingKey);
                Console.WriteLine("pengiriman berhasil {0}", sendResult.getMessageId());
            }
            catch (Exception ex)
            {
                Console.WriteLine("pengiriman gagal{0}", ex.ToString());
            }
        }

        // Sebelum keluar dari thread Anda, hentikan instance produser.
        producer.shutdown();

    }
}

Berlangganan pesan terurut

Kode contoh berikut menunjukkan cara berlangganan pesan terurut menggunakan TCP client SDK untuk .NET:

using System;
using System.Text;
using System.Threading;
using ons;

namespace demo
{

    public class MyMsgOrderListener : MessageOrderListener
    {
        public MyMsgOrderListener()
        {

        }

        ~MyMsgOrderListener()
        {
        }

        public override ons.OrderAction consume(Message value, ConsumeOrderContext context)
        {
            Byte[] text = Encoding.Default.GetBytes(value.getBody());
            Console.WriteLine(Encoding.UTF8.GetString(text));
            return ons.OrderAction.Success;
        }
    }

    class OrderConsumerExampleForEx
    {
        static void Main(string[] args)
        {
            // Konfigurasikan akun Anda. Anda dapat memperoleh informasi akun di Konsol Manajemen Alibaba Cloud.
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            // ID AccessKey yang digunakan untuk otentikasi.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
            // Rahasia AccessKey yang digunakan untuk otentikasi.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
            // ID grup konsumen yang Anda buat di konsol ApsaraMQ for RocketMQ.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "GID_example");
            // Topik yang Anda buat di konsol ApsaraMQ for RocketMQ.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
            // Titik akhir TCP. Anda dapat memperoleh titik akhir di bagian TCP Endpoint halaman Detail Instance di konsol ApsaraMQ for RocketMQ.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
            // Jalur log.
            factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

            // Buat instance konsumen.
            OrderConsumer consumer = ONSFactory.getInstance().createOrderConsumer(factoryInfo);

            // Berlangganan ke topik.
            consumer.subscribe(factoryInfo.getPublishTopics(), "*",new MyMsgOrderListener());

            // Mulai instance konsumen.
            consumer.start();

            // Jeda utas utama selama periode waktu tertentu.
            Thread.Sleep(30000);

            // Jika instance konsumen tidak lagi digunakan, matikan instance konsumen.
            consumer.shutdown();
        }
    }
}