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: Central de Documentação do Alibaba Cloud

ApsaraMQ for RocketMQ 5.x é compatível com clientes dos SDKs RocketMQ 3.x e 4.x. Este tópico fornece códigos de exemplo em C++ para enviar e receber mensagens normais, ordenadas, agendadas/atrasadas e transacionais.

Importante

Recomendamos o uso dos SDKs mais recentes do RocketMQ 5.x. Eles são totalmente compatíveis com os brokers do ApsaraMQ for RocketMQ 5.x e oferecem mais funções e recursos aprimorados. Para obter mais informações, consulte Informações da versão do SDK para C++.
A Alibaba Cloud mantém apenas os SDKs cliente do RocketMQ 4.x, 3.x e TCP. Recomendamos utilizá-los exclusivamente para negócios existentes.

Pré-requisitos

Antes de começar, verifique se você possui:

A biblioteca dinâmica C++ instalada. Para mais detalhes, consulte Instale a biblioteca dinâmica C++
Um tópico e um Group ID criados no console do ApsaraMQ for RocketMQ
O endpoint da instância obtido no console do ApsaraMQ for RocketMQ

Configuração comum

Todos os exemplos deste tópico compartilham a mesma configuração de conexão. Substitua os placeholders abaixo pelos seus valores antes de executar o código.

Placeholder	Descrição	Exemplo
`<your-group-id>`	Group ID criado no console	GID_example
`<your-access-point>`	Endpoint da instância no formato `host:port`. Não inclua `http://` ou `https://` e não utilize um endereço IP resolvido.	rmq-cn-xxx.rmq.aliyuncs.com:8080
`<your-topic>`	Tópico criado no console	normal_topic_01
`<instance-username>`	Nome de usuário da instância na aba Intelligent Authentication da página Access Control	N/A
`<instance-password>`	Senha da instância na aba Intelligent Authentication da página Access Control	N/A

Autenticação:

Acesso pela Internet: Requer nome de usuário e senha.
Acesso via VPC: Não requer nome de usuário nem senha.
Instância Serverless pela Internet: Requer nome de usuário e senha.
Instância Serverless em uma VPC com acesso sem autenticação ativado: Não requer nome de usuário nem senha.

Importante

Não especifique o ID da instância ao usar o SDK do RocketMQ 3.x ou 4.x para C++ com uma instância do ApsaraMQ for RocketMQ 5.0. Essa especificação causa falhas de conexão.

Configure comum do producer:

#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();

Configure comum do consumer:

#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");

Importante

Antes de iniciar um consumer:

Configure todas as assinaturas antes de chamar start().
Garanta que todos os consumers do mesmo grupo usem assinaturas idênticas.

Mensagens normais

Utilize mensagens normais para troca de mensagens de uso geral que não exigem ordenação, entrega atrasada ou suporte a transações.

Enviar mensagens normais

#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;
}

Assinar mensagens normais

#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;
}

Mensagens ordenadas

Use mensagens ordenadas quando as mensagens de uma mesma entidade de negócio precisarem ser processadas em sequência, como atualizações de status de pedidos ou alterações de estoque.

Esse tipo de mensagem encaminha aquelas com a mesma chave de partição para a mesma fila por meio de um MessageQueueSelector, garantindo a ordem de processamento.

Enviar mensagens ordenadas

#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;
}

Assinar mensagens ordenadas

O consumer utiliza MessageListenerOrderly em vez de MessageListenerConcurrently para processar as mensagens em ordem dentro de cada fila.

#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;
}

Mensagens agendadas e atrasadas

Utilize mensagens agendadas ou atrasadas para entregar mensagens em um horário específico ou após um atraso, como notificações cronometradas, mecanismos de nova tentativa ou execução adiada de tarefas.

Defina a propriedade da mensagem __STARTDELIVERTIME com um timestamp Unix em milissegundos. O broker retém a mensagem até o horário especificado.

Se o timestamp estiver no futuro, o broker entrega a mensagem nesse momento.
Caso o timestamp esteja no passado, o broker entrega a mensagem imediatamente.

Enviar mensagens agendadas ou atrasadas

#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;
}

Assinar mensagens agendadas ou atrasadas

O consumer usa MessageListenerConcurrently, o mesmo tipo de listener das mensagens normais. O broker gerencia o tempo; o código do consumer permanece idêntico.

#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;
}

Mensagens transacionais

Adote mensagens transacionais quando a lógica de negócio local e a entrega da mensagem precisarem ter sucesso ou falhar como uma unidade, como transferências de saldo entre contas.

O envio de mensagens transacionais funciona em três etapas:

O producer envia uma half message para o broker.
O broker invoca executeLocalTransaction no lado do producer. Retorne COMMIT_MESSAGE para entregar, ROLLBACK_MESSAGE para descartar ou UNKNOWN para adiar a decisão.
Se o resultado for UNKNOWN, o broker invoca periodicamente checkLocalTransaction até receber um resultado definitivo.

Enviar mensagens transacionais

#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;
}

Assinar mensagens transacionais

Assine mensagens transacionais da mesma forma que mensagens normais. Utilize MessageListenerConcurrently e DefaultMQPushConsumer. Para o código completo, consulte Assinar mensagens normais.

Tópicos relacionados

SDKs para outras linguagens de programação: Visão geral
Conceitos de tipos de mensagem: