TDengine.cpp 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243
  1. #include "TDengine.h"
  2. #include "taos.h"
  3. #include <QtCore/QDebug>
  4. #include <QtConcurrent/QtConcurrent>
  5. #include <assert.h>
  6. #include <stdio.h>
  7. #include <stdlib.h>
  8. #include <string.h>
  9. #include <time.h>
  10. #include "taos.h"
  11. TAOS* pConn = NULL;
  12. EventSubInterface* g_pSubCB; // 订阅回调
  13. std::list<std::string> g_lstTopics;
  14. int32_t TDengine::msgProcess(TAOS_RES* msg)
  15. {
  16. char buf[1024];
  17. int32_t rows = 0;
  18. const char* topicName = tmq_get_topic_name(msg);
  19. const char* dbName = tmq_get_db_name(msg);
  20. int32_t vgroupId = tmq_get_vgroup_id(msg);
  21. //qDebug() << __FILE__ << __LINE__ << "db: " << dbName;
  22. //qDebug() << __FILE__ << __LINE__ << "topic: " << topicName;
  23. //qDebug() << __FILE__ << __LINE__ << "vgroup id:" << vgroupId;
  24. while (1)
  25. {
  26. TAOS_ROW row = taos_fetch_row(msg);
  27. if (row == NULL)
  28. break;
  29. TAOS_FIELD* fields = taos_fetch_fields(msg);
  30. int32_t numOfFields = taos_field_count(msg);
  31. // int32_t* length = taos_fetch_lengths(msg);
  32. // int32_t precision = taos_result_precision(msg);
  33. rows++;
  34. taos_print_row(buf, row, fields, numOfFields);
  35. if( g_pSubCB != nullptr )
  36. {
  37. std::string topic = topicName;
  38. std::string content = buf;
  39. g_pSubCB->SubCB((char*)topic.c_str(),(char*)content.c_str());
  40. }
  41. //qDebug() << __FILE__ << __LINE__ << "row content: " << buf;
  42. }
  43. return rows;
  44. }
  45. //构建消费者
  46. tmq_t* TDengine::buildConsumer()
  47. {
  48. tmq_conf_res_t code;
  49. tmq_conf_t* conf = tmq_conf_new();
  50. code = tmq_conf_set(conf, "td.connect.ip", host.toStdString().c_str());
  51. if (TMQ_CONF_OK != code) {
  52. tmq_conf_destroy(conf);
  53. return NULL;
  54. }
  55. //code = tmq_conf_set(conf, "enable.auto.commit", "true");
  56. code = tmq_conf_set(conf, "enable.auto.commit", "false"); //禁用提交回调
  57. if (TMQ_CONF_OK != code) {
  58. tmq_conf_destroy(conf);
  59. return NULL;
  60. }
  61. // code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
  62. // if (TMQ_CONF_OK != code) {
  63. // tmq_conf_destroy(conf);
  64. // return NULL;
  65. // }
  66. code = tmq_conf_set(conf, "group.id", "cgrpName");
  67. if (TMQ_CONF_OK != code) {
  68. tmq_conf_destroy(conf);
  69. return NULL;
  70. }
  71. code = tmq_conf_set(conf, "client.id", "user defined name");
  72. if (TMQ_CONF_OK != code) {
  73. tmq_conf_destroy(conf);
  74. return NULL;
  75. }
  76. code = tmq_conf_set(conf, "td.connect.user", user.toStdString().c_str());
  77. if (TMQ_CONF_OK != code) {
  78. tmq_conf_destroy(conf);
  79. return NULL;
  80. }
  81. code = tmq_conf_set(conf, "td.connect.pass", password.toStdString().c_str());
  82. if (TMQ_CONF_OK != code) {
  83. tmq_conf_destroy(conf);
  84. return NULL;
  85. }
  86. //v3.2以后默认为最新的latest
  87. // code = tmq_conf_set(conf, "auto.offset.reset", "latest");
  88. // if (TMQ_CONF_OK != code) {
  89. // tmq_conf_destroy(conf);
  90. // return NULL;
  91. // }
  92. //tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); //禁用提交回调
  93. tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
  94. //qDebug()<< __FILE__ << __LINE__ << "tmq" << tmq;
  95. tmq_conf_destroy(conf);
  96. return tmq;
  97. }
  98. //构建主题
  99. tmq_list_t* TDengine::buildTopicList()
  100. {
  101. tmq_list_t* topicList = tmq_list_new();
  102. std::list<std::string>::iterator itr;
  103. for( itr = g_lstTopics.begin(); itr != g_lstTopics.end(); ++itr )
  104. {
  105. std::string szTopic = *itr;
  106. int32_t code = tmq_list_append(topicList, szTopic.c_str());
  107. if (code) {
  108. return NULL;
  109. }
  110. }
  111. return topicList;
  112. }
  113. //轮询主题
  114. void TDengine::topicLoop()
  115. {
  116. int32_t code;
  117. tmq_t* tmq = buildConsumer();
  118. if (NULL == tmq) {
  119. qDebug()<< __FILE__ << __LINE__ << "err" << stderr;
  120. return;
  121. }
  122. tmq_list_t* topic_list = buildTopicList();
  123. code = tmq_subscribe(tmq, topic_list);
  124. if ( code ) {
  125. qDebug() << __FILE__ << __LINE__ << "Failed to tmq_subscribe(): " << tmq_err2str(code);
  126. }
  127. tmq_list_destroy(topic_list);
  128. int32_t totalRows = 0;
  129. int32_t msgCnt = 0;
  130. int32_t timeout = 100; //多个主题共用1个线程轮询,超时时间太长会影响其它主题实时性
  131. while (true) {
  132. TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout);
  133. if (tmqmsg) {
  134. msgCnt++;
  135. totalRows += msgProcess(tmqmsg);
  136. taos_free_result(tmqmsg);
  137. } else {
  138. //超时
  139. //break;
  140. }
  141. }
  142. code = tmq_consumer_close(tmq);
  143. if (code) {
  144. fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(code));
  145. } else {
  146. fprintf(stderr, "%% Consumer closed\n");
  147. }
  148. qDebug()<< __FILE__ << __LINE__ << "stderr" << stderr;
  149. }
  150. TDengine::TDengine(QObject *parent)
  151. : QObject(parent)
  152. {
  153. pConn = nullptr;
  154. g_pSubCB = nullptr;
  155. g_lstTopics.clear();
  156. }
  157. TDengine::~TDengine()
  158. {
  159. taos_close(pConn);
  160. }
  161. void TDengine::exec(QString sql)
  162. {
  163. if (pConn)
  164. {
  165. TAOS_RES* pRes = taos_query(pConn, sql.toStdString().c_str());
  166. if (taos_errno(pRes) != 0) {
  167. qDebug() << __FILE__ << __LINE__ << "failed to insert into tab, reason:" << taos_errstr(pRes);
  168. qDebug()<< __FILE__ << __LINE__ << "err sql" << sql;
  169. }
  170. else
  171. {
  172. //qDebug()<< "exec ok";
  173. }
  174. taos_free_result(pRes);
  175. }
  176. }
  177. void TDengine::subscribe(QString ch, EventSubInterface *fn)
  178. {
  179. g_pSubCB = fn;
  180. g_lstTopics.push_back(ch.toLocal8Bit().toStdString());
  181. }
  182. void TDengine::psubscribe(QString ch, EventSubInterface *fn)
  183. {
  184. g_pSubCB = fn;
  185. g_lstTopics.push_back(ch.toLocal8Bit().toStdString());
  186. }
  187. void TDengine::Setup(tagSetup ts)
  188. {
  189. host = ts.addr.c_str();
  190. user = ts.user.c_str();
  191. dbName = ts.db.c_str();
  192. password = ts.password.c_str();
  193. m_nPort = ts.port;
  194. }
  195. void TDengine::start()
  196. {
  197. pConn = taos_connect(host.toStdString().c_str(),
  198. user.toStdString().c_str(),
  199. password.toStdString().c_str(),
  200. dbName.toStdString().c_str(),
  201. m_nPort);
  202. if (pConn == NULL) {
  203. qDebug()<< __FILE__ << __LINE__ << "td conn err.";
  204. return;
  205. }
  206. //开启一个线程轮询订阅的主题
  207. QtConcurrent::run(this, &TDengine::topicLoop);
  208. }