Sıralı mesajlar, ApsaraMQ for RocketMQ tarafından sunulan bir mesaj türüdür. Bu mesajlar kesin ilk giren ilk çıkar (FIFO) düzeninde yayınlanır ve tüketilir. Bu konuda, C veya C++ için TCP istemci SDK'sını kullanarak sıralı mesajların nasıl gönderileceği ve alınacağına ilişkin örnek kodlar sunulmaktadır.
Arka plan bilgileri
Sıralı mesajlar aşağıdaki türlere ayrılır:
Genel olarak sıralı mesajlar: Belirtilen bir konuya ait tüm mesajlar, ilk giren ilk çıkar (FIFO) düzeninde yayınlanır ve tüketilir.
Bölümlenmiş sıralı mesajlar: Belirtilen bir konuya ait tüm mesajlar, parçalama anahtarları kullanılarak farklı bölümlere dağıtılır. Her bölümdeki mesajlar FIFO düzeninde yayınlanır ve tüketilir. Parçalama anahtarı, farklı bölümleri tanımlamak için sıralı mesajlarda kullanılan bir anahtar alandır. Parçalama anahtarı, normal bir mesajın anahtarından farklıdır.
Daha fazla bilgi için bkz. Sıralı mesajlar.
Önkoşullar
Aşağıdaki işlemlerin tamamlandığından emin olun:
C veya C++ için SDK'yı indirin. Daha fazla bilgi için bkz. C++ için TCP istemci SDK'sının sürüm notları.
Ortamı hazırlayın. Daha fazla bilgi için bkz. Ortamı hazırlayın (v1.x.x).
Kodda belirtmek istediğiniz kaynakları ApsaraMQ for RocketMQ konsolunda oluşturun. Bu kaynaklar arasında örnekler, konular ve tüketici grupları bulunur. Daha fazla bilgi için bkz. Kaynak oluşturma.
Alibaba Cloud hesabınızın AccessKey çiftini edinin. Daha fazla bilgi için bkz. AccessKey oluşturma.
Sıralı mesajları gönderme
Bir ApsaraMQ for RocketMQ broker'ı, mesajların oluşturulma sırasını göndericinin tek bir üretici veya iş parçacığı kullanarak mesajları gönderme sırasına göre belirler. Gönderici, mesajları eş zamanlı olarak göndermek için birden çok üretici veya iş parçacığı kullanıyorsa mesaj sırası, ApsaraMQ for RocketMQ broker'ının mesajları aldığı sıraya göre belirlenir. Bu sıra, iş tarafındaki gönderme sırasından farklı olabilir.
Aşağıdaki örnek kod, C veya C++ için TCP istemci SDK'sını kullanarak sıralı mesajların nasıl gönderileceğini gösterir:
#include "ONSFactory.h"
#include "ONSClientException.h"
#include <iostream>
using namespace ons;
int main()
{
// The parameter that is required to create and use a producer.
ONSFactoryProperty factoryInfo;
// The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
// The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX");
// The topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
// The message content.
factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
// Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
// The AccessKey ID that is used for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
// The AccessKey secret that is used for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
// Create the producer.
OrderProducer *pProducer = ONSFactory::getInstance()->createOrderProducer(factoryInfo);
// Before you send the message, call the start() method only once to start the producer.
pProducer->start();
Message msg(
//Message Topic
factoryInfo.getPublishTopics(),
//Message Tag
"TagA",
//Message Key. Set a key to uniquely identify the message. We recommend that the key be globally unique so that if a message cannot be received, you can query it in the ApsaraMQ for RocketMQ console using the key and resend it. Note: Not setting a key does not affect message sending or receiving.
"ORDERID_100",
//Message Body
factoryInfo.getMessageContent()
);
// The key field that is used to identify partitions for partitionally ordered messages.
// This field can be set to a non-empty string for globally ordered messages.
std::string shardingKey = "abc";
// Messages that have the same sharding key are sent in order.
try
{
// Send the message. If no exception is thrown, the message is sent.
SendResultONS sendResult = pProducer->send(msg, shardingKey);
std::cout << "send success" << std::endl;
}
catch(ONSClientException & e)
{
// Specify the logic to handle errors.
}
// Before you exit your application, destroy the producer. If you do not destroy the producer, issues such as memory leaks may occur.
pProducer->shutdown();
return 0;
}
C++ SDK v1.1.1'de, üç parametreli Message(topic, tag, body) oluşturucusunda, body parametresinin dahili olarak yanlışlıkla keys alanına eşlendiği bilinen bir hata bulunmaktadır. Bu durum, mesaj gövdesinin boş olmasına neden olur ve send() çağrıldığında 'the message body is empty' hatasıyla sonuçlanır. Bu sorunu önlemek için yukarıdaki örnekte gösterildiği gibi dört parametreli Message(topic, tag, key, body) oluşturucusunu kullanın.
Sıralı mesajlara abone olma
Aşağıdaki örnek kod, C veya C++ için TCP istemci SDK'sını kullanarak sıralı mesajlara nasıl abone olunacağını gösterir:
#include "ONSFactory.h"
using namespace std;
using namespace ons;
// Create the consumer instance.
//After the push consumer pulls the message, the push consumer calls the consumeMessage function of the instance.
class ONSCLIENT_API MyMsgListener : public MessageOrderListener
{
public:
MyMsgListener()
{
}
virtual ~MyMsgListener()
{
}
virtual OrderAction consume(Message &message, ConsumeOrderContext &context)
{
// Consume messages based on your business requirements.
return Success; //CONSUME_SUCCESS;
}
};
int main(int argc, char* argv[])
{
// The parameter that is required to create and use the consumer.
ONSFactoryProperty factoryInfo;
// The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "XXX");
// The topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
// Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured.
// The AccessKey ID that is used for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
// The AccessKey secret that is used for authentication.
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
// The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX");
// Create the consumer.
OrderConsumer* orderConsumer = ONSFactory::getInstance()->createOrderConsumer(factoryInfo);
MyMsgListener msglistener;
// The message topic and tag to which the consumer subscribes.
orderConsumer->subscribe(factoryInfo.getPublishTopics(), "*",&msglistener );
// Register the instance to listen to messages. After the consumer pulls the messages, the consumer calls the consumeMessage function of the message listening class.
//Start the consumer.
orderConsumer->start();
for(volatile int i = 0; i < 1000000000; ++i) {
//wait
}
// Destroy the consumer. Before you exit the application, destroy the consumer. Otherwise, issues such as memory leaks may occur.
orderConsumer->shutdown();
return 0;
}