As mensagens ordenadas são um tipo de mensagem do ApsaraMQ for RocketMQ. Elas são publicadas e consumidas em ordem estrita de chegada (FIFO). Este tópico apresenta exemplos de código para enviar e receber mensagens ordenadas com o SDK do cliente TCP para C ou C++.
Informações básicas
As mensagens ordenadas dividem-se nos seguintes tipos:
Mensagens globalmente ordenadas: todas as mensagens de um tópico específico são publicadas e consumidas na ordem FIFO.
Mensagens ordenadas por partição: as mensagens de um tópico específico são distribuídas entre diferentes partições por meio de chaves de fragmentação. Dentro de cada partição, a publicação e o consumo seguem a ordem FIFO. A chave de fragmentação identifica as partições nas mensagens ordenadas e difere da chave de uma mensagem comum.
Para mais informações, consulte Mensagens ordenadas.
Pré-requisitos
Certifique-se de ter concluído as seguintes etapas:
Baixe o SDK para C ou C++. Para mais informações, consulte Notas de versão do SDK do cliente TCP para C++.
Prepare o ambiente. Consulte Preparar o ambiente (v1.x.x).
Crie os recursos no console do ApsaraMQ for RocketMQ antes de referenciá-los no código, incluindo instâncias, tópicos e grupos de consumidores. Consulte Criar recursos.
Obtenha o par de AccessKey da sua conta Alibaba Cloud. Consulte Criar um AccessKey.
Enviar mensagens ordenadas
O broker do ApsaraMQ for RocketMQ define a ordem de geração das mensagens com base na sequência em que o remetente as envia usando um único producer ou thread. Se o envio ocorrer simultaneamente por múltiplos producers ou threads, a ordem dependerá do momento em que o broker do ApsaraMQ for RocketMQ receber cada mensagem. Essa ordem pode divergir da sequência original de envio na aplicação.
O exemplo abaixo demonstra como enviar mensagens ordenadas com o SDK do cliente TCP para C ou C++:
#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(),
// The message tag. A message tag is similar to a Gmail tag and is used by consumers to filter messages in the ApsaraMQ for RocketMQ broker.
"TagA",
// The message body. A message body is data in the binary format. ApsaraMQ for RocketMQ does not process message bodies. The producer and the consumer must agree on the serialization and deserialization methods.
factoryInfo.getMessageContent()
);
// The message key. A key is the business-specific attribute of a message and must be globally unique.
// If you cannot receive a message as expected, you can use the key to query the message in the ApsaraMQ for RocketMQ console.
// Note: You can send and receive a message even if you do not specify the key.
msg.setKey("ORDERID_100");
// 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;
}
Inscrever-se em mensagens ordenadas
O código a seguir ilustra como se inscrever e consumir mensagens ordenadas com o SDK do cliente TCP para C ou C++:
#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;
}