:Notas de la versión

Instrucciones de uso

El SDK de cliente TCP para C++ V3.x.x solo permite acceder a instancias que contienen namespaces. Si la instancia que utiliza no tiene un namespace, no actualice el SDK de cliente TCP para C++ a la versión V3.x.x.

De forma predeterminada, todas las instancias de ApsaraMQ for RocketMQ 5.x incluyen namespaces. Si usa una instancia de ApsaraMQ for RocketMQ 4.x, puede comprobar si la instancia contiene un namespace en la sección Basic Information de la página Instance Details en ApsaraMQ for RocketMQ console.

Información de la versión

Requisitos del entorno

ONS-Client-CPP es un SDK de cliente de código abierto nativo de Apache RocketMQ 5.0. Apache RocketMQ 5.0 adopta el framework gRPC, desarrollado con base en HTTP 2.0 y Protobuf. Por lo tanto, gRPC también es obligatorio para el SDK para C++ V3.0.0. La siguiente tabla describe los requisitos de versión que deben cumplir las dependencias y las cadenas de herramientas.

Dependencias

Cadenas de herramientas

Estándar de C++

El SDK utiliza la biblioteca estándar de C++ 11. Se requiere C++ 11 o una versión posterior.

Instrucciones de compilación

Instrucciones de compilación para código abierto

1. Instale Bazel. Para obtener más información, consulte Installing Bazel.

   Nota

   Python 3.x.x es obligatorio para Bazel 4.x.

2. Descargue y descomprima el código abierto. Puede descargar el código mediante uno de los siguientes métodos:

   • Ejecute el comando git clone https://github.com/aliyun-mq/ons-client-cpp.git para clonar el código fuente.

   • Descargue el código en su equipo local desde el enlace en Version information.

3. Ejecute el siguiente comando en la carpeta del proyecto. Bazel descarga automáticamente todas las dependencias de terceros.

   bazel -c opt //dist/...

   El siguiente código muestra un ejemplo de salida:

   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

   Una vez compilado el código, la biblioteca estática que contiene todos los objetos fusionados se guarda en el archivo 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

Instrucciones de compilación para 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/...

Cambios en las funciones

Mensajes ordenados

El valor predeterminado del parámetro MaxReconsumeTimes cambió de Integer.MAX a 16. Este parámetro especifica el número máximo de reintentos para mensajes ordenados. Si un consumidor no logra procesar un mensaje después de alcanzar el número máximo de reintentos, el mensaje se envía a una cola de mensajes muertos. Puede especificar un valor personalizado para el parámetro MaxReconsumeTimes.

Consumo por difusión

En el modo de consumo por difusión, se admite la operación offsetStore. Puede llamar a esta operación para especificar el desplazamiento del consumidor desde el cual este comienza a consumir mensajes. Si no especifica el desplazamiento del consumidor, este inicia el consumo de mensajes desde el desplazamiento más reciente. Este comportamiento es consistente con versiones anteriores.

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

Modo push

• Si el número especificado de hilos de consumo no se encuentra dentro del rango válido de 1 a 1000, se genera una excepción cuando el sistema intenta crear un consumidor. Este tipo de excepción no se genera cuando el sistema intenta iniciar un consumidor.

• Se admite la función de limitación de consumo. Puede configurar esta función para limitar la tasa de consumo de mensajes. Esto ayuda a prevenir excepciones en la aplicación causadas por aumentos repentinos de mensajes en los clientes consumidores.

  Nota

  La función de limitación de consumo no aplica a los reintentos de mensajes ordenados.

  El siguiente código de ejemplo muestra cómo configurar la función de limitación de consumo:

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

Rastros de mensajes

Cambios en las operaciones de API

• La ruta de registro predeterminada cambió de ~/logs/rocketmqlogs/ons.log a ~/logs/rocketmqlogs/ons.log.

• La clase enum Action se movió del namespace global al namespace ons.

• Los archivos de encabezado se almacenan en la ruta /ons.

• El valor de retorno de Message#getStartDeliverTime cambió de int64_t a std::chrono::system_clock::timepoint o std::chrono::milliseconds.

• Se eliminó la declaración throws de las funciones porque ya no es compatible con C++ 11.

• La clase Producer proporciona una operación noexcept. Puede llamar a esta operación para evitar que se generen excepciones.

• Los tipos de enumeración cambiaron a namespace enum, es decir, enum class Type.

Preguntas frecuentes

• ¿Puedo usar un SDK de la versión más reciente junto con un SDK de una versión anterior en el mismo proceso? ¿Existe conflicto de símbolos?

  Los símbolos de la biblioteca estática precompilada del SDK más reciente existen en el namespace ons, que es el namespace predeterminado. Estos símbolos entran en conflicto con los símbolos utilizados en versiones anteriores del SDK. Puede compilar su SDK desde el código fuente y asegurarse de que el valor de la macro ONS_NAMESPACE no sea ons. De esta manera, puede usar su SDK junto con una versión anterior en el mismo proceso.

  Bazel ofrece varios métodos para definir macros. Por ejemplo, puede usar .bazelrc, la propiedad defines en reglas cc_library y la propiedad cc_library#copts para definir una macro.

• ¿Cómo compilo una biblioteca estática que contenga una tabla de símbolos al depurar el código?

  Ejecute el siguiente comando:

  bazel -c dbg //dist/...

  Para obtener información sobre las opciones de compilación, consulte la guía de usuario de Bazel.

• ¿Cómo resuelvo conflictos de dependencias si mi dependencia de Protobuf no cumple con el requisito de versión?

  ONS-Client-CPP utiliza el código fuente de dependencias de terceros. Solo necesita especificar las versiones de las dependencias de ONS-Client-CPP que sean consistentes con su dependencia.

  ONS-Client-CPP depende de RocketMQ-Client-CPP. Debe hacer un fork del repositorio apache/rocketmq-client-cpp y cambiar la URL de la dependencia en ons-cilent-cpp/bazel/deps.bzl por la URL del repositorio bifurcado.

• ¿Por qué se agota el tiempo de espera de una gran cantidad de mensajes salientes después de habilitar un proxy HTTP en mi equipo local y declarar variables de entorno como http_proxy y grpc_proxy?

  El SDK basado en gRPC admite proxies como http_proxy, https_proxy y grpc_proxy. Si no se requiere ningún proxy, puede configurar la variable de entorno no_grpc_proxy o no_proxy para omitir el sitio del proxy. Para obtener más información, consulte variables de entorno de gRPC.

• ¿El SDK admite los estándares C++ 98 y C++ 03?

  No, el SDK no admite el estándar C++ 98 ni C++ 03. El SDK utiliza gRPC como protocolo principal y Protobuf como dependencia clave. Los protocolos gRPC y Protobuf no son compatibles con el estándar C++ 98 ni C++ 03.

Proyecto de ejemplo

ons-client-cpp-demo