ApsaraMQ for RocketMQ'deki sıralı mesajlar, kesin ilk giren ilk çıkar (FIFO) düzeninde tüketilir. Bu konuda, .NET için TCP istemci SDK'sını kullanarak sıralı mesajların nasıl gönderileceğine ve alınacağına ilişkin örnek kodlar sunulmaktadır.
Sıralama modları
ApsaraMQ for RocketMQ iki sıralama modunu destekler:
Genel olarak sıralı mesajlar: Bir konudaki tüm mesajlar kesin FIFO düzeninde yayınlanır ve tüketilir.
Bölümlere göre sıralı mesajlar: Bir konudaki mesajlar parçalama anahtarı kullanılarak bölümlere dağıtılır. Her bölüm içindeki mesajlar kesin FIFO düzeninde tüketilir.
Parçalama anahtarı, sıralı mesajları belirli bir bölüme yönlendiren bir anahtar alandır. Normal mesajların anahtarlarından farklıdır.
ApsaraMQ for RocketMQ broker'ı, mesaj sırasını tek bir üretici veya iş parçacığının mesajları gönderme dizisine göre belirler. Birden fazla üretici veya iş parçacığı eş zamanlı olarak mesaj gönderirse, broker mesajları varış saatine göre sıralar; bu durum, amaçlanan iş mantığı sırasından farklı olabilir. Sıralamayı garanti altına almak için ilgili tüm mesajları tek bir üreticiden veya iş parçacığından gönderin.
Daha fazla bilgi için bkz. Sıralı mesajlar.
Ön koşullar
Başlamadan önce şunları yaptığınızdan emin olun:
.NET için SDK'yı indirin. Daha fazla bilgi için bkz. Yayın notları
Ortamı hazırlayın. Daha fazla bilgi için bkz. Ortamı hazırlama
ApsaraMQ for RocketMQ konsolunda örnekler, konular ve tüketici grupları oluşturun. Daha fazla bilgi için bkz. Kaynak oluşturma
Alibaba Cloud hesabınız için bir AccessKey çifti edinin. Daha fazla bilgi için bkz. AccessKey çifti oluşturma
Sıralı mesaj gönderme
Aşağıdaki örnek kod, .NET için TCP istemci SDK'sını kullanarak sıralı mesajlar gönderir. Tüm mesajlar aynı parçalama anahtarını paylaştığından aynı bölüme yönlendirilir ve gönderilme sırasına göre tüketilir.
Tam kod deposu için bkz. ApsaraMQ for RocketMQ kod deposu.
Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:
| Yer Tutucu | Açıklama | Örnek |
|---|---|---|
<your-group-id> |
Konsolda oluşturulan üretici grup kimliği | GID_example |
<your-topic> |
Konsolda oluşturulan konu | T_example_topic_name |
<your-tcp-endpoint> |
Konsoldaki Instance Details sayfasındaki TCP uç noktası | NameSrv_Addr |
<your-log-path> |
SDK günlükleri için yerel yol | C://log |
<your-sharding-key> |
Bölüm yönlendirmesini belirleyen parçalama anahtarı | App-Test |
using System;
using ons;
public class OrderProducerExampleForEx
{
public OrderProducerExampleForEx()
{
}
static void Main(string[] args) {
// Configure the producer properties.
ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
// Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
// and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
// AccessKey ID for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
// AccessKey secret for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
// Producer group ID created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<your-group-id>");
// Topic created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
// TCP endpoint from the Instance Details page in the console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<your-tcp-endpoint>");
// Local path for SDK logs.
factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "<your-log-path>");
// Create a producer instance.
// Producer instances are thread-safe and can send messages to different topics.
// In most cases, one producer instance per thread is sufficient.
OrderProducer producer = ONSFactory.getInstance().createOrderProducer(factoryInfo);
// Start the producer.
producer.start();
// Create a message with topic, tag, and body.
Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
// Define the sharding key. Messages with the same sharding key
// are sent to the same partition and consumed in order.
string shardingKey = "<your-sharding-key>";
for (int i = 0; i < 32; i++) {
try
{
SendResultONS sendResult = producer.send(msg, shardingKey);
Console.WriteLine("send success {0}", sendResult.getMessageId());
}
catch (Exception ex)
{
Console.WriteLine("send failure{0}", ex.ToString());
}
}
// Shut down the producer before exiting the thread.
producer.shutdown();
}
}
Sıralı mesaj alma
Aşağıdaki örnek kod, .NET için TCP istemci SDK'sını kullanarak sıralı mesajlar alır. Tüketici, her bölüm içindeki mesajları FIFO düzeninde işler.
Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:
| Yer Tutucu | Açıklama | Örnek |
|---|---|---|
<your-group-id> |
Konsolda oluşturulan tüketici grup kimliği | GID_example |
<your-topic> |
Konsolda oluşturulan konu | T_example_topic_name |
<your-tcp-endpoint> |
Konsoldaki Instance Details sayfasındaki TCP uç noktası | NameSrv_Addr |
<your-log-path> |
SDK günlükleri için yerel yol | C://log |
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)
{
// Configure the consumer properties.
ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
// AccessKey ID for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
// AccessKey secret for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
// Consumer group ID created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "<your-group-id>");
// Topic created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<your-topic>");
// TCP endpoint from the Instance Details page in the console.
factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<your-tcp-endpoint>");
// Local path for SDK logs.
factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "<your-log-path>");
// Create a consumer instance.
OrderConsumer consumer = ONSFactory.getInstance().createOrderConsumer(factoryInfo);
// Subscribe to the topic with a wildcard tag filter.
consumer.subscribe(factoryInfo.getPublishTopics(), "*", new MyMsgOrderListener());
// Start the consumer.
consumer.start();
// Keep the main thread alive to continue receiving messages.
Thread.Sleep(30000);
// Shut down the consumer when it is no longer needed.
consumer.shutdown();
}
}
}
Sonraki adımlar
Sıralı mesaj kavramları hakkında daha fazla bilgi edinin: Sıralı mesajlar