:Notas de versão

Observações de uso

O SDK de cliente TCP para C++ V3.x.x permite acessar apenas instâncias que contêm namespaces. Caso a instância em uso não possua um namespace, não atualize o SDK de cliente TCP para C++ para a versão V3.x.x.

Por padrão, todas as instâncias do ApsaraMQ for RocketMQ 5.x incluem namespaces. Se você utiliza uma instância do ApsaraMQ for RocketMQ 4.x, verifique se ela contém um namespace na seção Basic Information da página Instance Details no console do ApsaraMQ for RocketMQ.

Informações de versão

Requisitos de ambiente

O ONS-Client-CPP é um SDK de cliente open source nativo do Apache RocketMQ 5.0. Como o Apache RocketMQ 5.0 adota o framework gRPC, desenvolvido com base em HTTP 2.0 e Protobuf, o gRPC também é obrigatório para o SDK para C++ V3.0.0. A tabela a seguir detalha os requisitos de versão para dependências e toolchains.

Dependências

Toolchains

Padrão C++

Este SDK utiliza a biblioteca padrão C++ 11. É necessário usar C++ 11 ou superior.

Instruções de compilação

Instruções para compilar o código open source

1. Instale o Bazel. Para mais detalhes, consulte Installing Bazel.

   Nota

   O Bazel 4.x exige o Python 3.x.x.

2. Baixe e descompacte o código open source. Utilize um dos métodos abaixo para obter o código:

   • Execute o comando git clone https://github.com/aliyun-mq/ons-client-cpp.git para clonar o código-fonte.

   • Baixe o código para sua máquina local pelo link disponível em Informações de versão.

3. Execute o comando a seguir na pasta do projeto. O Bazel baixa automaticamente todas as dependências de terceiros.

   bazel -c opt //dist/...

   O exemplo abaixo mostra uma saída típica:

   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

   Após a compilação, a biblioteca estática contendo todos os objetos mesclados será salva no arquivo 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

Instruções de compilação 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/...

Alterações em recursos

Mensagens ordenadas

O valor padrão do parâmetro MaxReconsumeTimes mudou de Integer.MAX para 16. Esse parâmetro define o número máximo de tentativas de reconsumo para mensagens ordenadas. Quando o consumidor atinge esse limite sem conseguir processar a mensagem, ela vai para uma fila de mensagens mortas (dead-letter queue). É possível definir um valor personalizado para MaxReconsumeTimes.

Consumo por broadcast

No modo de consumo por broadcast, há suporte à operação offsetStore. Essa operação permite especificar o offset a partir do qual o consumidor iniciará o processamento das mensagens. Caso nenhum offset seja definido, o consumo começa pelo offset mais recente, mantendo a compatibilidade com versões 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

• Se o número de threads de consumo especificado estiver fora do intervalo válido de 1 a 1000, uma exceção será lançada durante a criação do consumidor, e não apenas na tentativa de iniciá-lo.

• Há suporte ao recurso de limitação de consumo (throttling), que restringe a taxa de processamento de mensagens. Essa funcionalidade evita falhas na aplicação causadas por picos repentinos de mensagens nos clientes consumidores.

  Nota

  A limitação de consumo não se aplica às novas tentativas de mensagens ordenadas.

  O trecho a seguir demonstra como configurar a limitação 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;
  }

Rastreamento de mensagens

Alterações nas operações de API

• O caminho padrão de log mudou de ~/logs/rocketmqlogs/ons.log para ~/logs/rocketmqlogs/ons.log.

• A classe enum Action foi movida do namespace global para o namespace ons.

• Os arquivos de cabeçalho agora residem no caminho /ons.

• O retorno de Message#getStartDeliverTime passou de int64_t para std::chrono::system_clock::timepoint ou std::chrono::milliseconds.

• A declaração throws foi removida das funções, pois não há mais suporte para ela no C++ 11.

• A classe Producer agora oferece uma operação noexcept, permitindo desativar o lançamento de exceções.

• Os tipos de enumeração foram convertidos para namespace enum, assumindo o formato enum class Type.

Perguntas frequentes

• É possível usar um SDK da versão mais recente junto com um SDK de versão anterior no mesmo processo? Os símbolos entram em conflito?

  Os símbolos da biblioteca estática pré-compilada do SDK mais recente residem no namespace ons (padrão), o que gera conflitos com os símbolos das versões anteriores. Para evitar isso, compile seu SDK a partir do código-fonte e garanta que o valor da macro ONS_NAMESPACE seja diferente de ons. Assim, ambas as versões poderão coexistir no mesmo processo.

  O Bazel oferece diversas formas de definir macros. Você pode utilizar o arquivo .bazelrc, a propriedade defines nas regras cc_library ou a propriedade cc_library#copts para essa finalidade.

• Como compilar uma biblioteca estática com tabela de símbolos para depuração?

  Execute o seguinte comando:

  bazel -c dbg //dist/...

  Para detalhes sobre opções de compilação, consulte o guia do usuário do Bazel.

• Como resolver conflitos se minha dependência do Protobuf não atender ao requisito de versão?

  O ONS-Client-CPP consome o código-fonte das dependências de terceiros. Basta ajustar as versões das dependências do ONS-Client-CPP para que correspondam às suas.

  Como o ONS-Client-CPP depende do RocketMQ-Client-CPP, faça um fork do repositório apache/rocketmq-client-cpp e atualize a URL da dependência no arquivo ons-cilent-cpp/bazel/deps.bzl para apontar para o repositório bifurcado.

• Por que muitas mensagens de saída apresentam timeout após eu ativar um proxy HTTP na minha máquina local e declarar variáveis de ambiente como http_proxy e grpc_proxy?

  O SDK baseado em gRPC aceita proxies como http_proxy, https_proxy e grpc_proxy. Se o proxy não for necessário, configure a variável de ambiente no_grpc_proxy ou no_proxy para ignorá-lo. Saiba mais em variáveis de ambiente do gRPC.

• O SDK suporta os padrões C++ 98 e C++ 03?

  Não. O SDK utiliza o gRPC como protocolo principal e o Protobuf como dependência essencial, e nenhum desses dois componentes oferece suporte aos padrões C++ 98 ou C++ 03.

Projeto de exemplo

ons-client-cpp-demo