in example/TransactionProducer.cpp [87:135]
int main(int argc, char* argv[]) {
RocketmqSendAndConsumerArgs info;
if (!ParseArgs(argc, argv, &info)) {
exit(-1);
}
PrintRocketmqSendAndConsumerArgs(info);
TransactionMQProducer producer("please_rename_unique_group_name");
producer.setNamesrvAddr(info.namesrv);
producer.setNamesrvDomain(info.namesrv_domain);
producer.setGroupName(info.groupname);
producer.setInstanceName(info.groupname);
producer.setSessionCredentials("mq acesskey", "mq secretkey", "ALIYUN");
producer.setSendMsgTimeout(500);
producer.setTcpTransportTryLockTimeout(1000);
producer.setTcpTransportConnectTimeout(400);
producer.setLogLevel(eLOG_LEVEL_DEBUG);
producer.setTransactionListener(new MyTransactionListener());
producer.start();
std::vector<std::shared_ptr<std::thread>> work_pool;
auto start = std::chrono::system_clock::now();
int msgcount = g_msgCount.load();
g_tps.start();
int threadCount = info.thread_count;
for (int j = 0; j < threadCount; j++) {
std::shared_ptr<std::thread> th = std::make_shared<std::thread>(SyncProducerWorker, &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";
for (size_t th = 0; th != work_pool.size(); ++th) {
work_pool[th]->join();
}
producer.shutdown();
return 0;
}