Este tema proporciona código de ejemplo en C++ para enviar y recibir mensajes normales con el SDK cliente HTTP de ApsaraMQ for RocketMQ. Los mensajes normales son aquellos que ofrece ApsaraMQ for RocketMQ sin características especiales. A diferencia de los mensajes programados, retrasados, ordenados y transaccionales, no incluyen semánticas de entrega particulares. Úselos para la mensajería de propósito general cuando no necesite garantías de orden, tiempo ni transacciones.
Requisitos previos
Antes de comenzar, compruebe que se completaron las siguientes operaciones:
Instale el SDK para C++. Para obtener más información, consulte Prepare the environment.
Cree en la consola de ApsaraMQ for RocketMQ los recursos que desea especificar en el código. Estos recursos incluyen instancias, temas y grupos de consumidores. Para obtener más información, consulte Create resources.
Obtenga el par de AccessKey de su cuenta de Alibaba Cloud. Para obtener más información, consulte Create an AccessKey.
Envío de mensajes normales
Inicialice un objeto MQClient con su endpoint HTTP y su par de AccessKey. Después, publique los mensajes en un tema.
Reemplace a continuación los marcadores de posición con sus valores reales:
|
Marcador de posición |
Descripción |
Dónde encontrarlo |
|
|
Endpoint de acceso HTTP |
Sección HTTP Endpoint de la página Instance Details |
|
|
Nombre del tema de destino |
Se crea en la consola de ApsaraMQ for RocketMQ |
|
|
ID de la instancia. Establezca una cadena vacía si la instancia no tiene espacio de nombres |
Página Instance Details de la consola |
#include <fstream>
#include <time.h>
#include "mq_http_sdk/mq_client.h"
using namespace std;
using namespace mq::http::sdk;
int main() {
MQClient mqClient(
// The HTTP endpoint, found in the HTTP Endpoint section
// of the Instance Details page in the ApsaraMQ for RocketMQ console.
"${HTTP_ENDPOINT}",
// The AccessKey ID and AccessKey secret for authentication.
// Read from environment variables to avoid hardcoding credentials.
System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
);
// The topic to publish messages to. Create the topic in the ApsaraMQ for RocketMQ console.
string topic = "${TOPIC}";
// The instance ID. If the instance has no namespace, set this to an empty string.
// Check the namespace on the Instance Details page in the console.
string instanceId = "${INSTANCE_ID}";
MQProducerPtr producer;
if (instanceId == "") {
producer = mqClient.getProducerRef(topic);
} else {
producer = mqClient.getProducerRef(instanceId, topic);
}
try {
// Send four messages in a loop.
for (int i = 0; i < 4; i++)
{
PublishMessageResponse pmResp;
// The message body.
TopicMessage pubMsg("Hello, mq!have key!");
// Set a custom property on the message.
pubMsg.putProperty("a",std::to_string(i));
// The message key.
pubMsg.setMessageKey("MessageKey" + std::to_string(i));
producer->publishMessage(pubMsg, pmResp);
cout << "Publish mq message success. Topic is: " << topic
<< ", msgId is:" << pmResp.getMessageId()
<< ", bodyMD5 is:" << pmResp.getMessageBodyMD5() << endl;
}
} catch (MQServerException& me) {
cout << "Request Failed: " + me.GetErrorCode() << ", requestId is:" << me.GetRequestId() << endl;
return -1;
} catch (MQExceptionBase& mb) {
cout << "Request Failed: " + mb.ToString() << endl;
return -2;
}
return 0;
}
Puntos clave:
Establezca las variables de entorno
ALIBABA_CLOUD_ACCESS_KEY_IDyALIBABA_CLOUD_ACCESS_KEY_SECRETantes de ejecutar el código.Cada mensaje admite propiedades personalizadas (
putProperty) y una clave de mensaje (setMessageKey).Si la instancia no tiene espacio de nombres, pase una cadena vacía para
instanceId. El SDK selecciona automáticamente la sobrecarga correcta degetProducerRef.
Suscripción a mensajes normales
Funcionamiento del consumo
El consumidor usa sondeo largo (long polling) HTTP para recuperar los mensajes:
El consumidor envía una solicitud al broker.
Si hay mensajes disponibles, el broker responde de inmediato.
Si no hay mensajes disponibles, el broker mantiene abierta la solicitud durante el tiempo de sondeo especificado (máximo: 30 segundos). Responde en cuanto llega un mensaje.
Después de procesar cada lote, el consumidor envía una confirmación (ACK) al broker.
Si el broker no recibe un ACK antes de
NextConsumeTime, vuelve a entregar el mensaje.
Parámetros del consumidor
|
Parámetro |
Descripción |
Valor de ejemplo |
Máximo |
|
Tamaño del lote |
Número máximo de mensajes consumidos por solicitud |
3 |
16 |
|
Duración del sondeo |
Tiempo que el broker espera mensajes (en segundos) |
3 |
30 |
Código de ejemplo
Además de los marcadores de posición de la sección Envío de mensajes normales, reemplace los siguientes:
|
Marcador de posición |
Descripción |
Dónde encontrarlo |
|
|
ID del grupo de consumidores |
Se crea en la consola de ApsaraMQ for RocketMQ |
#include <vector>
#include <fstream>
#include "mq_http_sdk/mq_client.h"
#ifdef _WIN32
#include <windows.h>
#else
#include <unistd.h>
#endif
using namespace std;
using namespace mq::http::sdk;
int main() {
MQClient mqClient(
// The HTTP endpoint, found in the HTTP Endpoint section
// of the Instance Details page in the ApsaraMQ for RocketMQ console.
"${HTTP_ENDPOINT}",
// The AccessKey ID and AccessKey secret for authentication.
// Read from environment variables to avoid hardcoding credentials.
System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
);
// The topic to consume messages from. Create the topic in the ApsaraMQ for RocketMQ console.
string topic = "${TOPIC}";
// The consumer group ID. Create the consumer group in the ApsaraMQ for RocketMQ console.
string groupId = "${GROUP_ID}";
// The instance ID. If the instance has no namespace, set this to an empty string.
// Check the namespace on the Instance Details page in the console.
string instanceId = "${INSTANCE_ID}";
MQConsumerPtr consumer;
if (instanceId == "") {
consumer = mqClient.getConsumerRef(topic, groupId);
} else {
consumer = mqClient.getConsumerRef(instanceId, topic, groupId, "");
}
do {
try {
std::vector<Message> messages;
// Consume messages in long polling mode.
// If no messages are available, the broker holds the request for the specified
// polling duration and responds immediately when a message arrives.
consumer->consumeMessage(
3, // Batch size: max messages per request. Maximum allowed: 16.
3, // Polling duration in seconds. Maximum allowed: 30.
messages
);
cout << "Consume: " << messages.size() << " Messages!" << endl;
// Process each message and collect receipt handles for acknowledgment.
std::vector<std::string> receiptHandles;
for (std::vector<Message>::iterator iter = messages.begin();
iter != messages.end(); ++iter)
{
cout << "MessageId: " << iter->getMessageId()
<< " PublishTime: " << iter->getPublishTime()
<< " Tag: " << iter->getMessageTag()
<< " Body: " << iter->getMessageBody()
<< " FirstConsumeTime: " << iter->getFirstConsumeTime()
<< " NextConsumeTime: " << iter->getNextConsumeTime()
<< " ConsumedTimes: " << iter->getConsumedTimes()
<< " Properties: " << iter->getPropertiesAsString()
<< " Key: " << iter->getMessageKey() << endl;
receiptHandles.push_back(iter->getReceiptHandle());
}
// Send an ACK to the broker. If the broker does not receive an ACK before
// NextConsumeTime, it redelivers the message. Each receipt handle carries
// a unique timestamp that expires after the NextConsumeTime window.
AckMessageResponse bdmResp;
consumer->ackMessage(receiptHandles, bdmResp);
if (!bdmResp.isSuccess()) {
// Log failed ACK items. A handle that has expired causes ACK failure,
// and the broker redelivers the corresponding message.
const std::vector<AckMessageFailedItem>& failedItems =
bdmResp.getAckMessageFailedItem();
for (std::vector<AckMessageFailedItem>::const_iterator iter = failedItems.begin();
iter != failedItems.end(); ++iter)
{
cout << "AckFailedItem: " << iter->errorCode
<< " " << iter->receiptHandle << endl;
}
} else {
cout << "Ack: " << messages.size() << " messages suc!" << endl;
}
} catch (MQServerException& me) {
if (me.GetErrorCode() == "MessageNotExist") {
cout << "No message to consume! RequestId: " + me.GetRequestId() << endl;
continue;
}
cout << "Request Failed: " + me.GetErrorCode() + ".RequestId: " + me.GetRequestId() << endl;
#ifdef _WIN32
Sleep(2000);
#else
usleep(2000 * 1000);
#endif
} catch (MQExceptionBase& mb) {
cout << "Request Failed: " + mb.ToString() << endl;
#ifdef _WIN32
Sleep(2000);
#else
usleep(2000 * 1000);
#endif
}
} while(true);
}
Puntos clave:
El consumidor se ejecuta en un bucle infinito y sondea de forma continua. Cuando no hay mensajes disponibles, captura
MessageNotExisty reintenta la operación.Llame a
ackMessagedespués de procesar un lote para confirmar la recepción. Los mensajes sin confirmación se vuelven a entregar trasNextConsumeTime.Cada identificador de recibo incluye una marca de tiempo única. Si el identificador caduca, el ACK falla y se activa una nueva entrega.
Ante errores distintos de
MessageNotExist, el consumidor hace una pausa de 2 segundos antes de reintentar. Esto evita bucles demasiado rápidos.