:Sürüm notları

Kullanım notları

C++ için TCP istemci SDK'sı V3.x.x sürümünü yalnızca ad alanı içeren örneklere erişmek için kullanabilirsiniz. Kullandığınız örnek bir ad alanı içermiyorsa C++ için TCP istemci SDK'sını V3.x.x sürümüne yükseltmeyin.

Varsayılan olarak tüm ApsaraMQ for RocketMQ 5.x örnekleri ad alanı içerir. Bir ApsaraMQ for RocketMQ 4.x örneği kullanıyorsanız örneğin bir ad alanı içerip içermediğini Instance Details sayfasının Basic Information bölümünden ApsaraMQ for RocketMQ konsolu nda kontrol edebilirsiniz.

Sürüm bilgileri

Ortam gereksinimleri

ONS-Client-CPP, Apache RocketMQ 5.0 ile yerel uyumlu açık kaynaklı bir istemci SDK'sıdır. Apache RocketMQ 5.0, HTTP 2.0 ve Protobuf tabanlı geliştirilen gRPC çerçevesini benimser. Bu nedenle C++ SDK'sı V3.0.0 için ayrıca gRPC gerekir. Aşağıdaki tabloda bağımlılıkların ve araç zincirlerinin karşılaması gereken sürüm gereksinimleri açıklanmaktadır.

Bağımlılık

Araç zincirleri

C++ Standardı

SDK, C++ 11 standart kitaplığını kullanır. C++ 11 veya üstü bir sürüm gerekir.

Derleme talimatları

Açık kaynak kodu için derleme talimatları

1. Bazel'i yükleyin. Daha fazla bilgi için bkz. Installing Bazel.

   Not

   Bazel 4.x için Python 3.x.x gerekir.

2. Açık kaynak kodunu indirin ve sıkıştırılmış dosyayı açın. Kodu aşağıdaki yöntemlerden birini kullanarak indirebilirsiniz:

   • Kaynak kodunu klonlamak için git clone https://github.com/aliyun-mq/ons-client-cpp.git komutunu çalıştırın.

   • Kodu Sürüm bilgileri bölümündeki bağlantıdan yerel makinenize indirin.

3. Proje klasöründe aşağıdaki komutu çalıştırın. Bazel tüm üçüncü taraf bağımlılıkları otomatik olarak indirir.

   bazel -c opt //dist/...

   Aşağıdaki kod örnek bir çıktıyı göstermektedir:

   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

   Kod derlendiğinde birleştirilmiş tüm nesneleri içeren statik kitaplık bazel-bin/dist/ons-dist.tar.gz dosyasına kaydedilir.

   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

CentOS 7 için derleme talimatları

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/...

Özellik değişiklikleri

Sıralı mesajlar

MaxReconsumeTimes parametresinin varsayılan değeri Integer.MAX değerinden 16'ya değiştirilmiştir. Bu parametre sıralı mesajlar için maksimum yeniden deneme sayısını belirtir. Bir tüketici maksimum yeniden deneme sayısına ulaştıktan sonra bir mesajı tüketemezse mesaj dead-letter queue'a iletilir. MaxReconsumeTimes parametresi için özel bir değer belirtebilirsiniz.

Yayın yoluyla tüketim

Yayın yoluyla tüketim modunda offsetStore işlemi desteklenir. Bu işlemi çağırarak bir tüketicinin mesajları tüketmeye başlayacağı tüketici ofsetini belirtebilirsiniz. Tüketici ofsetini belirtmezseniz tüketici mesaj tüketimine en son tüketici ofsetinden başlar. Bu davranış önceki sürümlerle tutarlıdır.

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

Push modu

• Belirtilen tüketim iş parçacığı sayısı 1 ile 1000 arasındaki geçerli aralıkta değilse sistem bir tüketici oluşturmaya çalıştığında bir özel durum oluşur. Bu tür bir özel durum sistem bir tüketiciyi başlatmaya çalıştığında oluşmaz.

• Tüketim kısıtlama özelliği desteklenir. Tüketim kısıtlama özelliğini yapılandırarak mesajların tüketim hızını sınırlayabilirsiniz. Bu sayede tüketici istemcilerinde ani mesaj artışlarından kaynaklanan uygulama özel durumlarının önüne geçebilirsiniz.

  Not

  Tüketim kısıtlama özelliği sıralı mesajların yeniden denenmesine uygulanmaz.

  Aşağıdaki örnek kod tüketim kısıtlama özelliğinin nasıl yapılandırılacağını göstermektedir:

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

Mesaj izlemeleri

API işlemlerindeki değişiklikler

• Varsayılan günlük yolu ~/logs/rocketmqlogs/ons.log konumundan ~/logs/rocketmqlogs/ons.log konumuna değiştirilmiştir.

• enum Action sınıfı genel ad alanından ons ad alanına taşınmıştır.

• Üstbilgi dosyaları /ons yolunda depolanır.

• Message#getStartDeliverTime dönüş değeri int64_t türünden std::chrono::system_clock::timepoint veya std::chrono::milliseconds türüne değiştirilmiştir.

• Fonksiyonlar için throws bildirimi silinmiştir çünkü throws bildirimi artık C++ 11'de desteklenmemektedir.

• Producer sınıfı özel durumların oluşmasını devre dışı bırakmak için kullanabileceğiniz bir noexcept işlemi sağlar.

• Numaralandırma türleri namespace enum, yani enum class Type olarak değiştirilmiştir.

SSS

• En son sürüm bir SDK'yı daha eski bir sürüm SDK ile aynı işlemde birlikte kullanabilir miyim? Semboller çakışır mı?

  En yeni SDK'nın önceden derlenmiş statik kitaplığına ait semboller varsayılan ad alanı olan ons ad alanında bulunur. Bu semboller SDK'nın önceki sürümlerinde kullanılan sembollerle çakışır. SDK'nızı kaynak koddan derleyebilir ve ONS_NAMESPACE makrosunun değerinin ons olmadığından emin olabilirsiniz. Bu şekilde SDK'nızı aynı işlemde daha eski bir sürümle birlikte kullanabilirsiniz.

  Bazel makro tanımlamak için birden çok yöntem sunar. Örneğin .bazelrc dosyasını, cc_library kurallarındaki defines property özelliğini ve cc_library#copts özelliğini kullanarak bir makro tanımlayabilirsiniz.

• Kodu hata ayıklarken sembol tablosu içeren bir statik kitaplığı nasıl derlerim?

  Aşağıdaki komutu çalıştırın:

  bazel -c dbg //dist/...

  Derleme seçenekleri hakkında bilgi için bkz. Bazel user guide.

• Protobuf bağımlılığım sürüm gereksinimini karşılamıyorsa bağımlılık çakışmalarını nasıl çözebilirim?

  ONS-Client-CPP üçüncü taraf bağımlılıkların kaynak kodunu kullanır. Yalnızca ONS-Client-CPP bağımlılıklarının sürümlerinin kendi bağımlılığınızla tutarlı olduğunu belirtmeniz gerekir.

  ONS-Client-CPP RocketMQ-Client-CPP'ye bağlıdır. apache/rocketmq-client-cpp repository, deposunu fork'lamanız ve ons-cilent-cpp/bazel/deps.bzl içindeki bağımlılık URL'sini fork'ladığınız deponun URL'siyle değiştirmeniz gerekir.

• Yerel makinemde bir HTTP proxy'sini etkinleştirip http_proxy ve grpc_proxy gibi ortam değişkenlerini bildirdikten sonra neden çok sayıda giden mesaj zaman aşımına uğruyor?

  gRPC tabanlı SDK http_proxy, https_proxy ve grpc_proxy gibi proxy'leri destekler. Proxy gerekmiyorsa proxy sitesini yok saymak için no_grpc_proxy veya no_proxy ortam değişkenini yapılandırabilirsiniz. Daha fazla bilgi için bkz. gRPC environment variables.

• SDK C++ 98 ve C++ 03 standartlarını destekliyor mu?

  Hayır, SDK C++ 98 veya C++ 03 standardını desteklemez. SDK temel protokol olarak gRPC'yi ve ana bağımlılık olarak Protobuf'u kullanır. gRPC ve Protobuf protokolleri C++ 98 veya C++ 03 standardını desteklemez.

Örnek proje

ons-client-cpp-demo