Sample code used to connect RocketMQ 3.x or 4.x SDK for C++ to an ApsaraMQ for RocketMQ 5.x instance to send and receive messages - ApsaraMQ for RocketMQ

ApsaraMQ for RocketMQ 5.x örnekleri, RocketMQ 3.x ve 4.x SDK istemcileriyle uyumludur. Bu konuda normal, sıralı, zamanlanmış/gecikmeli ve işlemsel mesajların gönderilmesi ve alınması için C++ örnek kodları sunulmaktadır.

Önemli

En son RocketMQ 5.x SDK'larını kullanmanızı öneririz. Bu SDK'lar, ApsaraMQ for RocketMQ 5.x broker'larıyla tam uyumludur ve daha fazla işlev ile geliştirilmiş özellikler sunar. Daha fazla bilgi için bkz. C++ SDK sürüm bilgileri.
Alibaba Cloud yalnızca RocketMQ 4.x, 3.x ve TCP istemci SDK'larını destekler. Bunları yalnızca mevcut iş süreçlerinde kullanmanızı öneririz.

Önkoşullar

Başlamadan önce aşağıdaki adımları tamamladığınızdan emin olun:

C++ dinamik kütüphanesini yükleyin. Ayrıntılar için bkz. C++ dinamik kütüphanesini yükleme
ApsaraMQ for RocketMQ konsolunda bir konu ve grup kimliği oluşturun.
ApsaraMQ for RocketMQ konsolundan örnek uç noktasını alın.

Ortak yapılandırma

Bu konudaki tüm örnekler aynı bağlantı kurulumunu paylaşır. Kodu çalıştırmadan önce aşağıdaki yer tutucuları kendi değerlerinizle değiştirin.

Yer Tutucu	Açıklama	Örnek
`<your-group-id>`	Konsolda oluşturulan grup kimliği	GID_example
`<your-access-point>`	`host:port` biçimindeki örnek uç noktası. `http://` veya `https://` eklemeyin ve çözümlenmiş bir IP adresi kullanmayın.	rmq-cn-xxx.rmq.aliyuncs.com:8080
`<your-topic>`	Konsolda oluşturulan konu	normal_topic_01
`<instance-username>`	Access Control sayfasındaki Intelligent Authentication sekmesinden alınan örnek kullanıcı adı	N/A
`<instance-password>`	Access Control sayfasındaki Intelligent Authentication sekmesinden alınan örnek parolası	N/A

Kimlik Doğrulama:

İnternet erişimi: Kullanıcı adı ve parola gerekir.
VPC erişimi: Kullanıcı adı veya parolaya gerek yoktur.
İnternet üzerinden sunucusuz örnek: Kullanıcı adı ve parola gerekir.
Kimlik doğrulamasız erişimin etkinleştirildiği VPC içindeki sunucusuz örnek: Kullanıcı adı veya parolaya gerek yoktur.

Önemli

ApsaraMQ for RocketMQ 5.0 örneğiyle RocketMQ 3.x veya 4.x C++ SDK'sını kullanırken örnek kimliğini belirtmeyin. Belirtmeniz bağlantı hatalarına neden olur.

Ortak producer kurulumu:

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

// Initialize the producer with your group ID
DefaultMQProducer producer("<your-group-id>");

// Set the endpoint (domain:port only, no http/https prefix)
producer.setNamesrvAddr("<your-access-point>");

// Set credentials (required for Internet access; skip for VPC access)
producer.setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

producer.start();

Ortak consumer kurulumu:

#include <iostream>
#include <thread>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;

// Initialize the push consumer with your group ID
DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("<your-group-id>");

// Set the endpoint (domain:port only, no http/https prefix)
consumer->setNamesrvAddr("<your-access-point>");

// Set credentials (required for Internet access; skip for VPC access)
consumer->setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

Önemli

Bir consumer başlatmadan önce:

start() çağrısından önce tüm abonelikleri yapılandırın.
Aynı gruptaki tüm consumer'ların özdeş abonelikler kullandığından emin olun.

Normal mesajlar

Sıralama, gecikmeli teslimat veya işlem desteği gerektirmeyen genel amaçlı mesajlaşma senaryolarında normal mesajları kullanın.

Normal mesajları gönderme

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

int main() {
    cout << "=======Before sending messages=======" << endl;

    DefaultMQProducer producer("<your-group-id>");
    producer.setNamesrvAddr("<your-access-point>");
    producer.setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");
    producer.start();

    auto start = chrono::system_clock::now();
    int count = 32;

    for (int i = 0; i < count; ++i) {
        MQMessage msg("<your-topic>", "HiTAG", "HelloCPPSDK.");
        try {
            SendResult sendResult = producer.send(msg);
            cout << "SendResult:" << sendResult.getSendStatus()
                 << ", Message ID: " << sendResult.getMsgId() << endl;
            this_thread::sleep_for(chrono::seconds(1));
        } catch (MQException& e) {
            cout << "ErrorCode: " << e.GetError()
                 << " Exception:" << e.what() << endl;
        }
    }

    auto interval = chrono::system_clock::now() - start;
    cout << "Send " << count << " messages OK, costs "
         << chrono::duration_cast<chrono::milliseconds>(interval).count()
         << "ms" << endl;

    producer.shutdown();
    cout << "=======After sending messages=======" << endl;
    return 0;
}

Normal mesajlara abone olma

#include <iostream>
#include <thread>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;

class ExampleMessageListener : public MessageListenerConcurrently {
public:
    ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {
        for (auto item = msgs.begin(); item != msgs.end(); item++) {
            std::cout << "Received Message Topic:" << item->getTopic()
                      << ", MsgId:" << item->getMsgId() << std::endl;
        }
        return CONSUME_SUCCESS;
    }
};

int main(int argc, char *argv[]) {
    std::cout << "=======Before consuming messages=======" << std::endl;

    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("<your-group-id>");
    consumer->setNamesrvAddr("<your-access-point>");
    consumer->setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

    // Register the listener and subscribe to the topic
    ExampleMessageListener *messageListener = new ExampleMessageListener();
    consumer->subscribe("<your-topic>", "*");
    consumer->registerMessageListener(messageListener);

    // Start after subscriptions are configured
    consumer->start();

    // Keep the main thread alive
    std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}

Sıralı mesajlar

Aynı iş varlığına ait mesajların sırayla işlenmesi gerektiğinde (örneğin sipariş durumu güncellemeleri veya stok değişiklikleri) sıralı mesajları kullanın.

Sıralı mesajlar, aynı bölüm anahtarına sahip mesajları aynı kuyruğa yönlendiren bir MessageQueueSelector aracılığıyla işleme sırasını garanti eder.

Sıralı mesajları gönderme

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

class ExampleSelectMessageQueueByHash : public MessageQueueSelector {
public:
    MQMessageQueue select(const std::vector<MQMessageQueue> &mqs,
                          const MQMessage &msg, void *arg) {
        // Route messages to a queue based on the partition key
        int orderId = *static_cast<int *>(arg);
        int index = orderId % mqs.size();
        return mqs[index];
    }
};

int main() {
    cout << "=======Before sending messages=======" << endl;

    DefaultMQProducer producer("<your-group-id>");
    producer.setNamesrvAddr("<your-access-point>");
    producer.setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");
    producer.start();

    auto start = chrono::system_clock::now();
    int count = 32;

    ExampleSelectMessageQueueByHash *pSelector = new ExampleSelectMessageQueueByHash();
    for (int i = 0; i < count; ++i) {
        MQMessage msg("<your-topic>", "HiTAG", "Hello,CPP SDK, Orderly Message.");
        try {
            // Pass the partition key (i) and the queue selector
            SendResult sendResult = producer.send(msg, pSelector, &i, 1, false);
            cout << "SendResult:" << sendResult.getSendStatus()
                 << ", Message ID: " << sendResult.getMsgId()
                 << " MessageQueue:" << sendResult.getMessageQueue().toString()
                 << endl;
            this_thread::sleep_for(chrono::seconds(1));
        } catch (MQException& e) {
            cout << "ErrorCode: " << e.GetError()
                 << " Exception:" << e.what() << endl;
        }
    }

    auto interval = chrono::system_clock::now() - start;
    cout << "Send " << count << " messages OK, costs "
         << chrono::duration_cast<chrono::milliseconds>(interval).count()
         << "ms" << endl;

    producer.shutdown();
    cout << "=======After sending messages=======" << endl;
    return 0;
}

Sıralı mesajlara abone olma

Consumer, her kuyruk içindeki mesajları sırayla işlemek için MessageListenerConcurrently yerine MessageListenerOrderly kullanır.

#include <iostream>
#include <thread>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;

class ExampleOrderlyMessageListener : public MessageListenerOrderly {
public:
    ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {
        for (auto item = msgs.begin(); item != msgs.end(); item++) {
            std::cout << "Received Message Topic:" << item->getTopic()
                      << ", MsgId:" << item->getMsgId() << std::endl;
        }
        return CONSUME_SUCCESS;
    }
};

int main(int argc, char *argv[]) {
    std::cout << "=======Before consuming messages=======" << std::endl;

    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("<your-group-id>");
    consumer->setNamesrvAddr("<your-access-point>");
    consumer->setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

    // Register the orderly listener and subscribe to the topic
    ExampleOrderlyMessageListener *messageListener = new ExampleOrderlyMessageListener();
    consumer->subscribe("<your-topic>", "*");
    consumer->registerMessageListener(messageListener);

    // Start after subscriptions are configured
    consumer->start();

    // Keep the main thread alive
    std::this_thread::sleep_for(std::chrono::seconds(60));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}

Zamanlanmış ve gecikmeli mesajlar

Mesajları belirli bir zamanda veya bir gecikme sonrası teslim etmek için zamanlanmış veya gecikmeli mesajları kullanın. Örnek kullanım alanları arasında zamanlı bildirimler, yeniden deneme mekanizmaları ve ertelenmiş görev yürütme yer alır.

__STARTDELIVERTIME mesaj özelliğini milisaniye cinsinden bir Unix zaman damgasına ayarlayın. Broker, belirtilen zamana kadar mesajı bekletir.

Zaman damgası gelecekteyse broker mesajı o zamanda teslim eder.
Zaman damgası geçmişteyse broker mesajı hemen teslim eder.

Zamanlanmış veya gecikmeli mesajları gönderme

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

int main() {
    cout << "=======Before sending messages=======" << endl;

    DefaultMQProducer producer("<your-group-id>");
    producer.setNamesrvAddr("<your-access-point>");
    producer.setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");
    producer.start();

    auto start = chrono::system_clock::now();
    int count = 32;

    for (int i = 0; i < count; ++i) {
        MQMessage msg("<your-topic>", "HiTAG", "Hello,CPP SDK, Delay Message.");

        // Calculate a delivery time 10 seconds from now
        chrono::system_clock::duration d = chrono::system_clock::now().time_since_epoch();
        chrono::milliseconds mil = chrono::duration_cast<chrono::milliseconds>(d);
        long exp = mil.count() + 10000;  // Delay: 10,000 ms (10 seconds)
        msg.setProperty("__STARTDELIVERTIME", to_string(exp));

        cout << "Now: " << mil.count() << " Exp:" << exp << endl;
        try {
            SendResult sendResult = producer.send(msg);
            cout << "SendResult:" << sendResult.getSendStatus()
                 << ", Message ID: " << sendResult.getMsgId() << endl;
            this_thread::sleep_for(chrono::seconds(1));
        } catch (MQException& e) {
            cout << "ErrorCode: " << e.GetError()
                 << " Exception:" << e.what() << endl;
        }
    }

    auto interval = chrono::system_clock::now() - start;
    cout << "Send " << count << " messages OK, costs "
         << chrono::duration_cast<chrono::milliseconds>(interval).count()
         << "ms" << endl;

    producer.shutdown();
    cout << "=======After sending messages=======" << endl;
    return 0;
}

Zamanlanmış veya gecikmeli mesajlara abone olma

Consumer, normal mesajlarda kullanılan dinleyici türü olan MessageListenerConcurrently kullanır. Zamanlama işlemini broker yönetir; consumer kodu aynıdır.

#include <iostream>
#include <thread>
#include <chrono>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;
using namespace std;

class ExampleDelayMessageListener : public MessageListenerConcurrently {
public:
    ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {
        for (auto item = msgs.begin(); item != msgs.end(); item++) {
            chrono::system_clock::duration d = chrono::system_clock::now().time_since_epoch();
            chrono::milliseconds mil = chrono::duration_cast<chrono::milliseconds>(d);
            cout << "Now: " << mil.count()
                 << " Received Message Topic:" << item->getTopic()
                 << ", MsgId:" << item->getMsgId()
                 << " DelayTime:" << item->getProperty("__STARTDELIVERTIME")
                 << endl;
        }
        return CONSUME_SUCCESS;
    }
};

int main(int argc, char *argv[]) {
    cout << "=======Before consuming messages=======" << endl;

    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("<your-group-id>");
    consumer->setNamesrvAddr("<your-access-point>");
    consumer->setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

    // Register the listener and subscribe to the topic
    ExampleDelayMessageListener *messageListener = new ExampleDelayMessageListener();
    consumer->subscribe("<your-topic>", "*");
    consumer->registerMessageListener(messageListener);

    // Start after subscriptions are configured
    consumer->start();

    // Keep the main thread alive (10 minutes to allow delayed messages to arrive)
    this_thread::sleep_for(chrono::seconds(600));
    consumer->shutdown();
    cout << "=======After consuming messages======" << endl;
    return 0;
}

İşlemsel mesajlar

Yerel iş mantığının ve mesaj teslimatının bir bütün olarak başarılı olması veya başarısız olması gerektiğinde (örneğin hesaplar arası bakiye transferleri) işlemsel mesajları kullanın.

İşlemsel mesajlaşma üç adımda gerçekleşir:

Producer, broker'a yarım mesaj gönderir.
Broker, producer tarafındaki executeLocalTransaction işlevini çağırır. Teslimat için COMMIT_MESSAGE değerini, atmak için ROLLBACK_MESSAGE değerini veya kararı ertelemek için UNKNOWN değerini döndürün.
Sonuç UNKNOWN ise broker, kesin bir sonuç alana kadar periyodik olarak checkLocalTransaction işlevini çağırır.

İşlemsel mesajları gönderme

#include <iostream>
#include <chrono>
#include <thread>
#include "TransactionMQProducer.h"
#include "MQClientException.h"
#include "TransactionListener.h"

using namespace std;
using namespace rocketmq;

class ExampleTransactionListener : public TransactionListener {
public:
    // Called after the half message is sent
    LocalTransactionState executeLocalTransaction(const MQMessage &msg, void *arg) {
        cout << "Execute Local Transaction, Received Message Topic:" << msg.getTopic()
             << ", Body:" << msg.getBody() << endl;

        // Run local business logic here. Return:
        //   COMMIT_MESSAGE  - deliver the message to consumers
        //   ROLLBACK_MESSAGE - discard the message
        //   UNKNOWN         - defer; the broker will call checkLocalTransaction later
        return UNKNOWN;
    }

    // Called by the broker when executeLocalTransaction returned UNKNOWN
    LocalTransactionState checkLocalTransaction(const MQMessageExt &msg) {
        cout << "Check Local Transaction, Received Message Topic:" << msg.getTopic()
             << ", MsgId:" << msg.getMsgId() << endl;

        // Query local transaction status and return the result
        return COMMIT_MESSAGE;
    }
};

int main() {
    cout << "=======Before sending messages=======" << endl;

    // Use TransactionMQProducer instead of DefaultMQProducer
    TransactionMQProducer producer("<your-group-id>");
    producer.setNamesrvAddr("<your-access-point>");
    producer.setSessionCredentials("<instance-username>", "<instance-password>", "ALIYUN");

    // Register the transaction listener before starting the producer
    ExampleTransactionListener *exampleTransactionListener = new ExampleTransactionListener();
    producer.setTransactionListener(exampleTransactionListener);
    producer.start();

    auto start = chrono::system_clock::now();
    int count = 3;

    for (int i = 0; i < count; ++i) {
        MQMessage msg("<your-topic>", "HiTAG", "Hello,CPP SDK, Transaction Message.");
        try {
            SendResult sendResult = producer.sendMessageInTransaction(msg, &i);
            cout << "SendResult:" << sendResult.getSendStatus()
                 << ", Message ID: " << sendResult.getMsgId() << endl;
            this_thread::sleep_for(chrono::seconds(1));
        } catch (MQException& e) {
            cout << "ErrorCode: " << e.GetError()
                 << " Exception:" << e.what() << endl;
        }
    }

    auto interval = chrono::system_clock::now() - start;
    cout << "Send " << count << " messages OK, costs "
         << chrono::duration_cast<chrono::milliseconds>(interval).count()
         << "ms" << endl;

    // Wait 60 seconds for the broker to call checkLocalTransaction
    cout << "Wait for local transaction check..... " << endl;
    for (int i = 0; i < 6; ++i) {
        this_thread::sleep_for(chrono::seconds(10));
        cout << "Running " << i * 10 + 10 << " Seconds......" << endl;
    }

    producer.shutdown();
    cout << "=======After sending messages=======" << endl;
    return 0;
}

İşlemsel mesajlara abone olma

İşlemsel mesajlara normal mesajlarla aynı şekilde abone olun. MessageListenerConcurrently ve DefaultMQPushConsumer kullanın. Tam kod için bkz. Normal mesajlara abone olma.

ApsaraMQ for RocketMQ:Send and receive messages with the RocketMQ 3.x/4.x SDK for C++

Önkoşullar

Ortak yapılandırma

Normal mesajlar

Normal mesajları gönderme

Normal mesajlara abone olma

Sıralı mesajlar

Sıralı mesajları gönderme

Sıralı mesajlara abone olma

Zamanlanmış ve gecikmeli mesajlar

Zamanlanmış veya gecikmeli mesajları gönderme

Zamanlanmış veya gecikmeli mesajlara abone olma

İşlemsel mesajlar

İşlemsel mesajları gönderme

İşlemsel mesajlara abone olma

İlgili konular