TDengine.cpp 7.7 KB

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