in example/AsyncProducer.cpp [86:135]
int main(int argc, char* argv[]) {
RocketmqSendAndConsumerArgs info;
if (!ParseArgs(argc, argv, &info)) {
exit(-1);
}
DefaultMQProducer producer("please_rename_unique_group_name");
if (!info.IsAutoDeleteSendCallback) {
g_callback = new MySendCallback();
}
PrintRocketmqSendAndConsumerArgs(info);
if (!info.namesrv.empty())
producer.setNamesrvAddr(info.namesrv);
producer.setGroupName(info.groupname);
producer.setInstanceName(info.groupname);
producer.setNamesrvDomain(info.namesrv_domain);
producer.start();
g_tps.start();
std::vector<std::shared_ptr<std::thread>> work_pool;
auto start = std::chrono::system_clock::now();
int msgcount = g_msgCount.load();
for (int j = 0; j < info.thread_count; j++) {
std::shared_ptr<std::thread> th = std::make_shared<std::thread>(AsyncProducerWorker, &info, &producer);
work_pool.push_back(th);
}
{
std::unique_lock<std::mutex> lck(g_mtx);
g_finished.wait(lck);
g_quit.store(true);
}
auto end = std::chrono::system_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
std::cout << "per msg time: " << duration.count() / (double)msgcount << "ms \n"
<< "========================finished==============================\n";
producer.shutdown();
for (size_t th = 0; th != work_pool.size(); ++th) {
work_pool[th]->join();
}
if (!info.IsAutoDeleteSendCallback) {
delete g_callback;
}
return 0;
}