Tüm Ürünler
Search
Belge Merkezi

ApsaraMQ for RocketMQ:Send and receive normal messages by using the C++ HTTP SDK

Son Güncelleme:Oct 04, 2026

Bu konu, ApsaraMQ for RocketMQ HTTP istemci SDK'sını kullanarak normal mesajların nasıl gönderilip alınacağını gösteren C++ örnek kodlarını sunar. Normal mesajlar, ApsaraMQ for RocketMQ tarafından sağlanan ve özel özellik içermeyen mesajlardır; zamanlanmış, gecikmeli, sıralı ve işlemli mesajlardan farklı olarak herhangi bir özel teslimat semantiği taşımazlar. Sıralama, zamanlama ve işlem garantilerinin gerekli olmadığı genel amaçlı mesajlaşma senaryolarında normal mesajları kullanın.

Önkoşullar

Başlamadan önce aşağıdaki işlemleri tamamladığınızdan emin olun:

  • C++ SDK'yı yükleyin. Daha fazla bilgi için bkz. Ortamı hazırlayın.

  • ApsaraMQ for RocketMQ konsolunda kodda belirtmek istediğiniz kaynakları oluşturun. Bu kaynaklar arasında örnekler, konular ve tüketici grupları bulunur. Daha fazla bilgi için bkz. Kaynakları oluşturun.

  • Alibaba Cloud hesabınızın AccessKey çiftini edinin. Daha fazla bilgi için bkz. AccessKey oluşturma.

Normal mesaj gönderme

HTTP uç noktanız ve AccessKey çiftiniz ile bir MQClient örneği başlatın, ardından mesajları bir konuya yayınlayın.

Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:

Yer Tutucu Açıklama Nereden Bulunur
${HTTP_ENDPOINT} HTTP erişim uç noktası HTTP Endpoint bölümünde, Instance Details sayfasında
${TOPIC} Hedef konu adı ApsaraMQ for RocketMQ konsolunda oluşturulmuştur
${INSTANCE_ID} Örnek kimliği. Örneğin ad alanı yoksa boş dize olarak ayarlayın Konsoldaki Instance Details sayfasında
#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;
}

Önemli noktalar:

  • Kodu çalıştırmadan önce ALIBABA_CLOUD_ACCESS_KEY_ID ve ALIBABA_CLOUD_ACCESS_KEY_SECRET ortam değişkenlerini ayarlayın.

  • Her mesaj özel özellikleri (putProperty) ve bir mesaj anahtarını (setMessageKey) destekler.

  • Örneğin ad alanı yoksa instanceId parametresi için boş bir dize iletin. SDK doğru getProducerRef aşırı yüklemesini otomatik olarak seçer.

Normal mesajlara abonelik

Tüketim mekanizması

Tüketici, mesajları almak için HTTP uzun yoklama (long polling) yöntemini kullanır:

  1. Tüketici aracıya (broker) bir istek gönderir.

  2. Mesajlar mevcutsa aracı hemen yanıt verir.

  3. Mesaj mevcut değilse aracı, belirtilen yoklama süresi boyunca (maksimum: 30 saniye) isteği açık tutar ve bir mesaj geldiğinde hemen yanıt verir.

  4. Her toplu işi işledikten sonra tüketici, aracıya bir onay (ACK) gönderir.

  5. Aracı, NextConsumeTime süresi dolmadan bir ACK almazsa mesajı yeniden teslim eder.

Tüketici parametreleri

Parametre Açıklama Örnek değer Maksimum
Toplu iş boyutu İstek başına tüketilen maksimum mesaj sayısı 3 16
Yoklama süresi Aracının mesajlar için beklediği süre (saniye) 3 30

Örnek kod

Normal mesaj gönderme bölümünde listelenenlere ek olarak aşağıdaki yer tutucuları değiştirin:

Yer Tutucu Açıklama Nereden Bulunur
${GROUP_ID} Tüketici grubu kimliği ApsaraMQ for RocketMQ konsolunda oluşturulmuştur
#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);
}

Önemli noktalar:

  • Tüketici sonsuz bir döngü içinde sürekli yoklama yaparak çalışır. Mesaj mevcut olmadığında MessageNotExist hatasını yakalar ve yeniden dener.

  • Bir toplu işi işledikten sonra alımı onaylamak için ackMessage yöntemini çağırın. Onaylanmayan mesajlar NextConsumeTime sonrasında yeniden teslim edilir.

  • Her teslimat tanıtıcısı benzersiz bir zaman damgası taşır. Süresi dolmuş bir tanıtıcı ACK başarısızlığına neden olur ve bu durum yeniden teslimi tetikler.

  • MessageNotExist dışındaki hatalarda tüketici, sık döngülerden kaçınmak için yeniden denemeden önce 2 saniye bekler.