This is an automated email from the ASF dual-hosted git repository.
lizhanhui pushed a commit to branch develop-cpp
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/develop-cpp by this push:
new e11d745b fix: example publish message QPS stats lambda
e11d745b is described below
commit e11d745b41eaf71ee4be4e93a382bbf24c3de0ee
Author: Li Zhanhui <lizhan...@gmail.com>
AuthorDate: Fri Mar 29 23:01:32 2024 +0800
fix: example publish message QPS stats lambda
---
cpp/examples/ExampleProducer.cpp | 4 ++--
cpp/examples/ExampleProducerWithAsync.cpp | 4 ++--
cpp/examples/ExampleProducerWithFifoMessage.cpp | 7 ++++---
cpp/examples/ExampleProducerWithTimedMessage.cpp | 4 ++--
cpp/examples/ExampleProducerWithTransactionalMessage.cpp | 4 ++--
5 files changed, 12 insertions(+), 11 deletions(-)
diff --git a/cpp/examples/ExampleProducer.cpp b/cpp/examples/ExampleProducer.cpp
index 2e170ce5..5e20cc12 100644
--- a/cpp/examples/ExampleProducer.cpp
+++ b/cpp/examples/ExampleProducer.cpp
@@ -89,8 +89,8 @@ int main(int argc, char* argv[]) {
auto stats_lambda = [&] {
while (!stopped.load(std::memory_order_relaxed)) {
long cnt = count.load(std::memory_order_relaxed);
- while (count.compare_exchange_weak(cnt, 0)) {
- break;
+ while (!count.compare_exchange_weak(cnt, 0)) {
+ cnt = count.load(std::memory_order_relaxed);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "QPS: " << cnt << std::endl;
diff --git a/cpp/examples/ExampleProducerWithAsync.cpp
b/cpp/examples/ExampleProducerWithAsync.cpp
index 5e9cc12d..63b7611c 100644
--- a/cpp/examples/ExampleProducerWithAsync.cpp
+++ b/cpp/examples/ExampleProducerWithAsync.cpp
@@ -128,8 +128,8 @@ int main(int argc, char* argv[]) {
auto stats_lambda = [&] {
while (!stopped.load(std::memory_order_relaxed)) {
long cnt = count.load(std::memory_order_relaxed);
- while (count.compare_exchange_weak(cnt, 0)) {
- break;
+ while (!count.compare_exchange_weak(cnt, 0)) {
+ cnt = count.load(std::memory_order_relaxed);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "QPS: " << cnt << std::endl;
diff --git a/cpp/examples/ExampleProducerWithFifoMessage.cpp
b/cpp/examples/ExampleProducerWithFifoMessage.cpp
index f45b2d12..4fa34f9d 100644
--- a/cpp/examples/ExampleProducerWithFifoMessage.cpp
+++ b/cpp/examples/ExampleProducerWithFifoMessage.cpp
@@ -84,10 +84,11 @@ int main(int argc, char* argv[]) {
std::atomic_long count(0);
auto stats_lambda = [&] {
+ std::cout << "Stats thread starts" << std::endl;
while (!stopped.load(std::memory_order_relaxed)) {
long cnt = count.load(std::memory_order_relaxed);
- while (count.compare_exchange_weak(cnt, 0)) {
- break;
+ while (!count.compare_exchange_weak(cnt, 0)) {
+ cnt = count.load(std::memory_order_relaxed);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "QPS: " << cnt << std::endl;
@@ -110,7 +111,7 @@ int main(int argc, char* argv[]) {
.build();
std::error_code ec;
SendReceipt send_receipt = producer.send(std::move(message), ec);
- std::cout << "Message-ID: " << send_receipt.message_id << std::endl;
+ // std::cout << "Message-ID: " << send_receipt.message_id << std::endl;
count++;
}
} catch (...) {
diff --git a/cpp/examples/ExampleProducerWithTimedMessage.cpp
b/cpp/examples/ExampleProducerWithTimedMessage.cpp
index 62b81385..d6237459 100644
--- a/cpp/examples/ExampleProducerWithTimedMessage.cpp
+++ b/cpp/examples/ExampleProducerWithTimedMessage.cpp
@@ -87,8 +87,8 @@ int main(int argc, char* argv[]) {
auto stats_lambda = [&] {
while (!stopped.load(std::memory_order_relaxed)) {
long cnt = count.load(std::memory_order_relaxed);
- while (count.compare_exchange_weak(cnt, 0)) {
- break;
+ while (!count.compare_exchange_weak(cnt, 0)) {
+ cnt = count.load(std::memory_order_relaxed);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "QPS: " << cnt << std::endl;
diff --git a/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
b/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
index 13d7f046..50620c5a 100644
--- a/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
+++ b/cpp/examples/ExampleProducerWithTransactionalMessage.cpp
@@ -92,8 +92,8 @@ int main(int argc, char* argv[]) {
auto stats_lambda = [&] {
while (!stopped.load(std::memory_order_relaxed)) {
long cnt = count.load(std::memory_order_relaxed);
- while (count.compare_exchange_weak(cnt, 0)) {
- break;
+ while (!count.compare_exchange_weak(cnt, 0)) {
+ cnt = count.load(std::memory_order_relaxed);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "QPS: " << cnt << std::endl;
