Los mensajes ordenados son un tipo de mensaje que proporciona ApsaraMQ for RocketMQ. Estos mensajes se publican y se consumen en un orden estricto de primero en entrar, primero en salir (FIFO). En este tema se incluye código de ejemplo para enviar y recibir mensajes ordenados mediante el SDK de cliente TCP para C o C++.
Contexto
Los mensajes ordenados se clasifican en los siguientes tipos:
Mensajes ordenados globalmente: todos los mensajes de un topic específico se publican y se consumen en orden FIFO.
Mensajes ordenados por partición: todos los mensajes de un topic específico se distribuyen en distintas particiones mediante claves de fragmentación (sharding keys). Los mensajes de cada partición se publican y se consumen en orden FIFO. Una clave de fragmentación es un campo clave que se utiliza en los mensajes ordenados para identificar las diferentes particiones. Esta clave de fragmentación no es lo mismo que la clave de un mensaje normal.
Para obtener más información, consulte Ordered messages.
Requisitos previos
Compruebe que completó las siguientes operaciones:
Descargó el SDK para C o C++. Para obtener más información, consulte Release notes for the TCP client SDK for C++.
Preparó el entorno. Para obtener más información, consulte Prepare the environment (v1.x.x).
Creó en la consola de ApsaraMQ for RocketMQ los recursos que va a especificar en el código. Estos recursos incluyen instancias, topics y grupos de consumidores. Para obtener más información, consulte Create resources.
Obtuvo el par de AccessKey de su cuenta de Alibaba Cloud. Para obtener más información, consulte Create an AccessKey.
Envío de mensajes ordenados
Un broker de ApsaraMQ for RocketMQ determina el orden de generación de los mensajes según el orden en que el remitente los envía con un solo productor o hilo. Si el remitente usa varios productores o hilos para enviar mensajes de forma concurrente, el orden lo define la secuencia en la que el broker de ApsaraMQ for RocketMQ recibe dichos mensajes. Este orden puede diferir del orden de envío en el lado del negocio.
A continuación se muestra un ejemplo de código para enviar mensajes ordenados con el SDK de cliente TCP para C o 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(),
//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;
}
En el SDK de C++ v1.1.1, el constructor de tres parámetros Message(topic, tag, body) tiene un bug conocido: el parámetro body se asigna internamente de forma incorrecta al campo keys. Esto provoca que el cuerpo del mensaje quede vacío y genera el error 'the message body is empty' cuando se llama a send(). Use el constructor de cuatro parámetros Message(topic, tag, key, body), como se muestra en el ejemplo anterior, para evitar este problema.
Suscripción a mensajes ordenados
A continuación se muestra un ejemplo de código para suscribirse a mensajes ordenados con el SDK de cliente TCP para C o 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;
}