:Notes de version

Consignes d'utilisation

Vous pouvez utiliser le SDK client TCP pour C++ V3.x.x uniquement pour accéder aux instances qui contiennent des namespaces. Si l'instance que vous utilisez ne contient pas de namespace, ne mettez pas à niveau le SDK client TCP pour C++ vers la version V3.x.x.

Par défaut, toutes les instances ApsaraMQ for RocketMQ 5.x contiennent des namespaces. Si vous utilisez une instance ApsaraMQ for RocketMQ 4.x, vous pouvez vérifier si l'instance contient un namespace dans la section Basic Information de la page Instance Details dans la console ApsaraMQ for RocketMQ.

Informations sur la version

Exigences environnementales

ONS-Client-CPP est un SDK client open source natif pour Apache RocketMQ 5.0. Apache RocketMQ 5.0 adopte le framework gRPC, développé sur la base de HTTP 2.0 et Protobuf. Par conséquent, gRPC est également requis pour le SDK pour C++ V3.0.0. Le tableau suivant décrit les exigences de version que les dépendances et les chaînes d'outils doivent respecter.

Dépendance

Chaînes d'outils

Standard C++

Le SDK utilise la bibliothèque standard C++ 11. C++ 11 ou une version ultérieure est requis.

Instructions de compilation

Instructions de compilation pour le code source open source

1. Installez Bazel. Pour plus d'informations, consultez Installing Bazel.

   Remarque

   Python 3.x.x est requis pour Bazel 4.x.

2. Téléchargez et décompressez le code source open source. Vous pouvez télécharger le code en utilisant l'une des méthodes suivantes :

   • Exécutez la commande git clone https://github.com/aliyun-mq/ons-client-cpp.git pour cloner le code source.

   • Téléchargez le code sur votre machine locale à partir du lien indiqué dans la section Informations sur la version.

3. Exécutez la commande suivante dans le dossier du projet. Bazel télécharge automatiquement toutes les dépendances tierces.

   bazel -c opt //dist/...

   Le code suivant fournit un exemple de sortie :

   INFO: From Action dist/libons_library.pic.a:
   starting to run shell
   INFO: Elapsed time: 39.480s, Critical Path: 38.89s
   INFO: 2044 processes: 1796 remote cache hit, 241 internal, 7 processwrapper-sandbox.
   INFO: Build completed successfully, 2044 total actions

   Lorsque le code est compilé, la bibliothèque statique contenant tous les objets fusionnés est enregistrée dans le fichier bazel-bin/dist/ons-dist.tar.gz.

   root@a36849cf2f24:~/ons-client-cpp# ls -lah bazel-bin/dist/ons-dist.tar.gz 
   -r-xr-xr-x 1 root root 15M Oct 14 08:03 bazel-bin/dist/ons-dist.tar.gz

Instructions de compilation pour CentOS 7

wget https://copr.fedorainfracloud.org/coprs/vbatts/bazel/repo/epel-7/vbatts-bazel-epel-7.repo
cp vbatts-bazel-epel-7.repo /etc/yum.repos.d/
yum install devtoolset-4-gcc devtoolset-4-gcc-c++ bazel4 python3 git -y
scl enable devtoolset-4 bash
unlink /usr/bin/python && ln -s /usr/bin/python3 /usr/bin/python
git clone git@github.com:aliyun-mq/ons-client-cpp.git
cd ons-client-cpp && bazel build //dist/...

Modifications des fonctionnalités

Messages ordonnés

La valeur par défaut du paramètre MaxReconsumeTimes est passée de Integer.MAX à 16. Ce paramètre spécifie le nombre maximal de tentatives pour les messages ordonnés. Si un consommateur échoue à consommer un message après avoir atteint le nombre maximal de tentatives, le message est envoyé vers une file d'attente de messages morts. Vous pouvez spécifier une valeur personnalisée pour le paramètre MaxReconsumeTimes.

Consommation en mode broadcast

En mode de consommation en diffusion (broadcasting), l'opération offsetStore est prise en charge. Vous pouvez appeler cette opération pour spécifier le décalage du consommateur à partir duquel il commence à consommer les messages. Si vous ne spécifiez pas le décalage du consommateur, celui-ci commence la consommation des messages à partir du dernier décalage enregistré. Ce comportement est cohérent avec les versions précédentes.

#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>

#include "ons/MessageModel.h"
#include "ons/ONSFactory.h"

#include "rocketmq/Logger.h"

using namespace std;
using namespace ons;

std::mutex console_mtx;

class ExampleMessageListener : public MessageListener {
public:
  Action consume(const Message& message, ConsumeContext& context) noexcept override {
    std::lock_guard<std::mutex> lk(console_mtx);
    auto latency = std::chrono::system_clock::now() - message.getStoreTimestamp();
    auto latency2 = std::chrono::system_clock::now() - message.getBornTimestamp();
    std::cout << "Received a message. Topic: " << message.getTopic() << ", MsgId: " << message.getMsgID()
              << ", Body-size: " << message.getBody().size()
              << ", Current - Store-Time: " << std::chrono::duration_cast<std::chrono::milliseconds>(latency).count()
              << "ms, Current - Born-Time: " << std::chrono::duration_cast<std::chrono::milliseconds>(latency2).count()
              << "ms" << std::endl;
    return Action::CommitMessage;
  }
};

int main(int argc, char* argv[]) {
  auto& logger = rocketmq::getLogger();
  logger.setLevel(rocketmq::Level::Debug);
  logger.init();

  std::cout << "=======Before consuming messages=======" << std::endl;
  ONSFactoryProperty factory_property;
  
  // Obtain the consumer offset by calling the offsetStore operation. This operation is supported only in broadcasting consumption mode. 
  factory_property.setMessageModel(ONS_NAMESPACE::MessageModel::BROADCASTING);

  factory_property.setFactoryProperty(ons::ONSFactoryProperty::GroupId, "GID_cpp_sdk_standard");

  PushConsumer* consumer = ONSFactory::getInstance()->createPushConsumer(factory_property);

  const char* topic = "cpp_sdk_standard";
  const char* tag = "*";

  // register your own listener here to handle the messages received.
  auto* messageListener = new ExampleMessageListener();
  consumer->subscribe(topic, tag);
  consumer->registerMessageListener(messageListener);

  // Start this consumer
  consumer->start();

  // Keep main thread running until process finished.
  std::this_thread::sleep_for(std::chrono::minutes(15));

  consumer->shutdown();
  std::cout << "=======After consuming messages======" << std::endl;
  return 0;
}

Mode Push

• Si le nombre spécifié de threads de consommation n'est pas compris dans la plage valide de 1 à 1000, une exception est levée lorsque le système tente de créer un consommateur. Ce type d'exception n'est pas levé lorsque le système tente de démarrer un consommateur.

• La fonctionnalité de limitation du débit de consommation est prise en charge. Vous pouvez configurer cette fonctionnalité pour limiter le taux de consommation des messages. Cela permet d'éviter les exceptions applicatives causées par des pics soudains de messages sur les clients consommateurs.

  Remarque

  La fonctionnalité de limitation du débit de consommation ne s'applique pas aux nouvelles tentatives de messages ordonnés.

  L'exemple de code suivant illustre comment configurer la fonctionnalité de limitation du débit de consommation :

  #include <chrono>
  #include <iostream>
  #include <mutex>
  #include <thread>
  
  #include "ons/MessageModel.h"
  #include "ons/ONSFactory.h"
  
  #include "rocketmq/Logger.h"
  
  using namespace std;
  using namespace ons;
  
  std::mutex console_mtx;
  
  class ExampleMessageListener : public MessageListener {
  public:
    Action consume(const Message& message, ConsumeContext& context) noexcept override {
      std::lock_guard<std::mutex> lk(console_mtx);
      auto latency = std::chrono::system_clock::now() - message.getStoreTimestamp();
      auto latency2 = std::chrono::system_clock::now() - message.getBornTimestamp();
      std::cout << "Received a message. Topic: " << message.getTopic() << ", MsgId: " << message.getMsgID()
                << ", Body-size: " << message.getBody().size() << ", Tag: " << message.getTag()
                << ", Current - Store-Time: " << std::chrono::duration_cast<std::chrono::milliseconds>(latency).count()
                << "ms, Current - Born-Time: " << std::chrono::duration_cast<std::chrono::milliseconds>(latency2).count()
                << "ms" << std::endl;
      return Action::CommitMessage;
    }
  };
  
  int main(int argc, char* argv[]) {
    auto& logger = rocketmq::getLogger();
    logger.setLevel(rocketmq::Level::Debug);
    logger.init();
  
    const char* topic = "cpp_sdk_standard";
    const char* tag = "*";
  
    std::cout << "=======Before consuming messages=======" << std::endl;
    ONSFactoryProperty factory_property;
    factory_property.setFactoryProperty(ons::ONSFactoryProperty::GroupId, "GID_cpp_sdk_standard");
  
    // Client-side throttling.
    factory_property.throttle(topic, 16);
  
    PushConsumer* consumer = ONSFactory::getInstance()->createPushConsumer(factory_property);
  
    // register your own listener here to handle the messages received.
    auto* messageListener = new ExampleMessageListener();
    consumer->subscribe(topic, tag);
    consumer->registerMessageListener(messageListener);
  
    // Start this consumer.
    consumer->start();
  
    // Keep main thread running until process finished.
    std::this_thread::sleep_for(std::chrono::minutes(15));
  
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
  }

Traces de messages

Modifications des opérations API

• Le chemin d'accès par défaut des journaux est passé de ~/logs/rocketmqlogs/ons.log à ~/logs/rocketmqlogs/ons.log.

• La classe enum Action a été déplacée de l'espace de noms global vers l'espace de noms ons.

• Les fichiers d'en-tête sont stockés dans le chemin /ons.

• La valeur de retour de Message#getStartDeliverTime est passée de int64_t à std::chrono::system_clock::timepoint ou std::chrono::milliseconds.

• La déclaration throws pour les fonctions a été supprimée car elle n'est plus prise en charge en C++ 11.

• La classe Producer fournit une opération noexcept. Vous pouvez appeler cette opération pour désactiver le lancement d'exceptions.

• Les types d'énumération sont passés à namespace enum, à savoir enum class Type.

FAQ

• Puis-je utiliser un SDK de la dernière version conjointement avec un SDK d'une version antérieure dans le même processus ? Les symboles entrent-ils en conflit ?

  Les symboles de la bibliothèque statique précompilée du dernier SDK existent dans l'espace de noms ons, qui est l'espace de noms par défaut. Ces symboles entrent en conflit avec ceux utilisés dans les versions antérieures du SDK. Vous pouvez compiler votre SDK à partir du code source et vous assurer que la valeur de la macro ONS_NAMESPACE n'est pas ons. De cette manière, vous pouvez utiliser votre SDK conjointement avec une version antérieure dans le même processus.

  Bazel propose plusieurs méthodes pour définir des macros. Par exemple, vous pouvez utiliser .bazelrc, la propriété defines dans les règles cc_library, ainsi que la propriété cc_library#copts pour définir une macro.

• Comment compiler une bibliothèque statique contenant une table des symboles lors du débogage du code ?

  Exécutez la commande suivante :

  bazel -c dbg //dist/...

  Pour plus d'informations sur les options de compilation, consultez le guide utilisateur Bazel.

• Comment résoudre les conflits de dépendances si ma dépendance Protobuf ne respecte pas l'exigence de version ?

  ONS-Client-CPP utilise le code source des dépendances tierces. Vous devez uniquement spécifier les versions des dépendances ONS-Client-CPP qui sont cohérentes avec votre propre dépendance.

  ONS-Client-CPP dépend de RocketMQ-Client-CPP. Vous devez fork le dépôt apache/rocketmq-client-cpp, et modifier l'URL de la dépendance dans ons-cilent-cpp/bazel/deps.bzl pour qu'elle pointe vers l'URL du dépôt forké.

• Pourquoi un grand nombre de messages sortants expirent-ils après avoir activé un proxy HTTP sur ma machine locale et déclaré des variables d'environnement telles que http_proxy et grpc_proxy ?

  Le SDK basé sur gRPC prend en charge les proxies tels que http_proxy, https_proxy et grpc_proxy. Si aucun proxy n'est requis, vous pouvez configurer la variable d'environnement no_grpc_proxy ou no_proxy pour ignorer le site proxy. Pour plus d'informations, consultez les variables d'environnement gRPC.

• Le SDK prend-il en charge les normes C++ 98 et C++ 03 ?

  Non, le SDK ne prend pas en charge la norme C++ 98 ni C++ 03. Le SDK utilise gRPC comme protocole principal et Protobuf comme dépendance clé. Les protocoles gRPC et Protobuf ne prennent pas en charge les normes C++ 98 ou C++ 03.

Exemple de projet

ons-client-cpp-demo