You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2022/08/01 03:04:34 UTC
[rocketmq-clients] branch master updated: Rate limit async producer example (#117)
This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new acfb3ca Rate limit async producer example (#117)
acfb3ca is described below
commit acfb3cab8f411e0f62fa7c6aecd99226155485a0
Author: Zhanhui Li <li...@gmail.com>
AuthorDate: Mon Aug 1 11:04:30 2022 +0800
Rate limit async producer example (#117)
* Rate limit async producer example
* Set log level of examples to Info
---
cpp/examples/ExampleProducer.cpp | 4 +-
cpp/examples/ExampleProducerWithAsync.cpp | 103 ++++++++++++++-------
cpp/examples/ExampleProducerWithFifoMessage.cpp | 4 +-
cpp/examples/ExampleProducerWithTimedMessage.cpp | 6 +-
.../ExampleProducerWithTransactionalMessage.cpp | 4 +-
cpp/examples/ExamplePushConsumer.cpp | 4 +-
cpp/examples/ExampleSimpleConsumer.cpp | 4 +-
7 files changed, 85 insertions(+), 44 deletions(-)
diff --git a/cpp/examples/ExampleProducer.cpp b/cpp/examples/ExampleProducer.cpp
index 33e10ee..e7a2bc6 100644
--- a/cpp/examples/ExampleProducer.cpp
+++ b/cpp/examples/ExampleProducer.cpp
@@ -62,8 +62,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
CredentialsProviderPtr credentials_provider;
diff --git a/cpp/examples/ExampleProducerWithAsync.cpp b/cpp/examples/ExampleProducerWithAsync.cpp
index c8f9dfb..3a453a9 100644
--- a/cpp/examples/ExampleProducerWithAsync.cpp
+++ b/cpp/examples/ExampleProducerWithAsync.cpp
@@ -17,18 +17,60 @@
#include <algorithm>
#include <atomic>
#include <condition_variable>
+#include <cstdint>
#include <iostream>
#include <mutex>
#include <random>
#include <string>
#include <system_error>
+#include "absl/base/thread_annotations.h"
+#include "absl/synchronization/mutex.h"
#include "gflags/gflags.h"
#include "rocketmq/Message.h"
#include "rocketmq/Producer.h"
using namespace ROCKETMQ_NAMESPACE;
+/**
+ * @brief A simple Semaphore to limit request concurrency.
+ */
+class Semaphore {
+public:
+ Semaphore(std::size_t permits) : permits_(permits) {
+ }
+
+ /**
+ * @brief Acquire a permit.
+ */
+ void acquire() LOCKS_EXCLUDED(mtx_) {
+ while (true) {
+ absl::MutexLock lk(&mtx_);
+ if (permits_ > 0) {
+ permits_--;
+ return;
+ }
+ cv_.Wait(&mtx_);
+ }
+ }
+
+ /**
+ * @brief Release the permit back to semaphore.
+ */
+ void release() LOCKS_EXCLUDED(mtx_) {
+ absl::MutexLock lk(&mtx_);
+ permits_++;
+ if (1 == permits_) {
+ cv_.Signal();
+ }
+ }
+
+private:
+ std::size_t permits_{0};
+ absl::Mutex mtx_;
+ absl::CondVar cv_;
+};
+
const std::string& alphaNumeric() {
static std::string alpha_numeric("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ");
return alpha_numeric;
@@ -50,10 +92,11 @@ std::string randomString(std::string::size_type len) {
return result;
}
-DEFINE_string(topic, "lingchu_normal_topic", "Topic to which messages are published");
+DEFINE_string(topic, "standard_topic_sample", "Topic to which messages are published");
DEFINE_string(access_point, "121.196.167.124:8081", "Service access URL, provided by your service provider");
DEFINE_int32(message_body_size, 4096, "Message body size");
DEFINE_uint32(total, 256, "Number of sample messages to publish");
+DEFINE_uint32(concurrency, 128, "Concurrency of async send");
DEFINE_string(access_key, "", "Your access key ID");
DEFINE_string(access_secret, "", "Your access secret");
@@ -61,8 +104,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
CredentialsProviderPtr credentials_provider;
@@ -94,46 +137,44 @@ int main(int argc, char* argv[]) {
std::thread stats_thread(stats_lambda);
std::string body = randomString(FLAGS_message_body_size);
- std::cout << "Message body size: " << body.length() << std::endl;
std::size_t completed = 0;
std::mutex mtx;
std::condition_variable cv;
- try {
- auto send_callback = [&](const std::error_code& ec, const SendReceipt& receipt) {
- std::unique_lock<std::mutex> lk(mtx);
- completed++;
- count++;
- std::cout << "Message[id=" << receipt.message_id << "] sent" << std::endl;
- if (completed >= FLAGS_total) {
- cv.notify_all();
- }
- };
-
- for (std::size_t i = 0; i < FLAGS_total; ++i) {
- auto message = Message::newBuilder()
- .withTopic(FLAGS_topic)
- .withTag("TagA")
- .withKeys({"Key-" + std::to_string(i)})
- .withBody(body)
- .build();
- producer.send(std::move(message), send_callback);
- }
+ auto semaphore = absl::make_unique<Semaphore>(FLAGS_concurrency);
- {
- std::unique_lock<std::mutex> lk(mtx);
- cv.wait(lk, [&]() { return completed >= FLAGS_total; });
+ auto send_callback = [&](const std::error_code& ec, const SendReceipt& receipt) {
+ std::unique_lock<std::mutex> lk(mtx);
+ semaphore->release();
+ completed++;
+ count++;
+ if (completed >= FLAGS_total) {
+ cv.notify_all();
}
- } catch (...) {
- std::cerr << "Ah...No!!!" << std::endl;
+ };
+
+ for (std::size_t i = 0; i < FLAGS_total; ++i) {
+ auto message = Message::newBuilder()
+ .withTopic(FLAGS_topic)
+ .withTag("TagA")
+ .withKeys({"Key-" + std::to_string(i)})
+ .withBody(body)
+ .build();
+ semaphore->acquire();
+ producer.send(std::move(message), send_callback);
}
+
+ {
+ std::unique_lock<std::mutex> lk(mtx);
+ cv.wait(lk, [&]() { return completed >= FLAGS_total; });
+ std::cout << "Completed: " << completed << ", total: " << FLAGS_total << std::endl;
+ }
+
stopped.store(true, std::memory_order_relaxed);
if (stats_thread.joinable()) {
stats_thread.join();
}
- std::this_thread::sleep_for(std::chrono::seconds(1));
-
return EXIT_SUCCESS;
}
\ No newline at end of file
diff --git a/cpp/examples/ExampleProducerWithFifoMessage.cpp b/cpp/examples/ExampleProducerWithFifoMessage.cpp
index 52a377b..134f9df 100644
--- a/cpp/examples/ExampleProducerWithFifoMessage.cpp
+++ b/cpp/examples/ExampleProducerWithFifoMessage.cpp
@@ -60,8 +60,8 @@ int main(int argc, char* argv[]) {
// Adjust log level for file/console sinks
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
CredentialsProviderPtr credentials_provider;
diff --git a/cpp/examples/ExampleProducerWithTimedMessage.cpp b/cpp/examples/ExampleProducerWithTimedMessage.cpp
index 30d3ea3..8eaf94b 100644
--- a/cpp/examples/ExampleProducerWithTimedMessage.cpp
+++ b/cpp/examples/ExampleProducerWithTimedMessage.cpp
@@ -50,7 +50,7 @@ std::string randomString(std::string::size_type len) {
return result;
}
-DEFINE_string(topic, "lingchu_normal_topic", "Topic to which messages are published");
+DEFINE_string(topic, "standard_topic_sample", "Topic to which messages are published");
DEFINE_string(access_point, "121.196.167.124:8081", "Service access URL, provided by your service provider");
DEFINE_int32(message_body_size, 4096, "Message body size");
DEFINE_uint32(total, 256, "Number of sample messages to publish");
@@ -61,8 +61,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
CredentialsProviderPtr credentials_provider;
diff --git a/cpp/examples/ExampleProducerWithTransactionalMessage.cpp b/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
index f32490b..e6c0aaf 100644
--- a/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
+++ b/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
@@ -59,8 +59,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
auto checker = [](const Message& message) -> TransactionState {
diff --git a/cpp/examples/ExamplePushConsumer.cpp b/cpp/examples/ExamplePushConsumer.cpp
index a10a1e2..365a84d 100644
--- a/cpp/examples/ExamplePushConsumer.cpp
+++ b/cpp/examples/ExamplePushConsumer.cpp
@@ -35,8 +35,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
std::string tag = "*";
diff --git a/cpp/examples/ExampleSimpleConsumer.cpp b/cpp/examples/ExampleSimpleConsumer.cpp
index 94c8d6c..b1294cf 100644
--- a/cpp/examples/ExampleSimpleConsumer.cpp
+++ b/cpp/examples/ExampleSimpleConsumer.cpp
@@ -34,8 +34,8 @@ int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
auto& logger = getLogger();
- logger.setConsoleLevel(Level::Debug);
- logger.setLevel(Level::Debug);
+ logger.setConsoleLevel(Level::Info);
+ logger.setLevel(Level::Info);
logger.init();
std::string tag = "*";