陈拓 2021/04/19-2021/04/20

1. 概述

在《将MQTT收到的数据保存到MySQL数据库》


https://zhuanlan.zhihu.com/p/365940502


https://blog.csdn.net/chentuo2000/article/details/115862060


一文中我们用同步的方法将MQTT订阅消息中的数据存储到了MySQL数据库中。


本文的程序用异步方式订阅消息。异步订阅消息的说明见：


《用C语言实现mosquitto MQTT订阅消息(异步)》


https://zhuanlan.zhihu.com/p/365483724


https://blog.csdn.net/chentuo2000/article/details/115786111


2. 异步mysql_log.c程序代码


  1. #include <stdio.h>
  2. #include <stdlib.h>
  3. #include <string.h>
  4. #include <unistd.h>
  5. #include <signal.h>
  6. #include <mosquitto.h>
  7. #include <mysql.h>
  9. #define db_host “localhost”
  10. #define db_username “ct”
  11. #define db_password “ct”
  12. #define db_database “smarthome”
  13. #define db_port 3306
  15. #define db_query “INSERT INTO temperature (deviceid, celsius_temp) VALUES (?,?)”
  17. #define mqtt_host “localhost”
  18. #define mqtt_port 1883
  20. static int run = 1;
  21. static MYSQL_STMT _stmt = NULL;
  23. void handle_signal(int s)
  24. {
  25. printf(“\n Capture sign no:%d\n”, s);
  26. run = 0;
  27. }
  29. void connect_callback(struct mosquitto _mosq, void _obj, int reason_code)
  30. {
  31. }
  33. void message_callback(struct mosquitto _mosq, void _obj, const struct mosquitto_message _message)
  34. {
  35. printf(“%s %d %s\n”, message->topic, message->qos, (char _)message->payload);
  37. MYSQL_BIND bind[2];
  39. memset(bind, 0, sizeof(bind));
  41. bind[0].buffer_type = MYSQL_TYPE_STRING;
  42. bind[0].buffer = message->topic;
  43. bind[0].buffer_length = strlen(message->topic);
  44. // Note: payload is normally a binary blob and could contains
  45. // NULL byte. This sample does not handle it and assume payload is a
  46. // string.
  47. bind[1].buffer_type = MYSQL_TYPE_STRING;
  48. bind[1].buffer = message->payload;
  49. bind[1].buffer_length = message->payloadlen;
  51. mysql_stmt_bind_param(stmt, bind);
  52. mysql_stmt_execute(stmt);
  53. }
  55. int main(int argc, char _argv[])
  56. {
  57. MYSQL *connection;
  58. my_bool reconnect = true;
  59. char clientid[24];
  60. struct <span class=”hljs-title class_“>mosquitto _mosq;
  61. int rc = 0;
  63. signal(SIGINT, handle_signal); // 捕捉终端CTRL+c产生的SIGINT信号
  64. signal(SIGTERM, handle_signal); // 程序结束(terminate)信号
  66. mysql_library_init(0, NULL, NULL);
  67. mosquitto_lib_init();
  69. connection = mysql_init(NULL);
  71. if(connection){
  72. mysql_options(connection, MYSQL_OPT_RECONNECT, &reconnect);
  74. connection = mysql_real_connect(connection, db_host, db_username, db_password, db_database, db_port, NULL, 0);
  76. if(connection){
  77. stmt = mysql_stmt_init(connection);
  79. mysql_stmt_prepare(stmt, db_query, strlen(db_query));
  81. memset(clientid, 0, 24);
  82. snprintf(clientid, 23, “mysql_log_%d”, getpid());
  83. mosq = mosquitto_new(clientid, true, connection);
  84. if(mosq){
  85. mosquitto_connect_callback_set(mosq, connect_callback);
  86. mosquitto_message_callback_set(mosq, message_callback);
  88. mosquitto_username_pw_set(mosq, “ct”, “1qaz2wsx”);
  89. rc = mosquitto_connect_async(mosq, mqtt_host, mqtt_port, 60);
  90. if(rc != MOSQ_ERR_SUCCESS) {
  91. mosquitto_destroy(mosq);
  92. fprintf(stderr, “Error: %s\n”, mosquitto_strerror(rc));
  93. return 1;
  94. }
  96. mosquitto_subscribe(mosq, NULL, “#”, 0);
  98. rc = mosquitto_loop_start(mosq);
  99. if(rc != MOSQ_ERR_SUCCESS) {
  100. mosquitto_destroy(mosq);
  101. fprintf(stderr, “Error: %s\n”, mosquitto_strerror(rc));
  102. return 1;
  103. }
  105. printf(“Start!\n”);
  106. while(run) {
  107. sleep(20);
  108. }
  109. }
  110. mysql_stmt_close(stmt);
  111. mysql_close(connection);
  112. }else{
  113. fprintf(stderr, “Error: Unable to connect to database.\n”);
  114. printf(“%s\n”, mysql_error(connection));
  115. rc = 1;
  116. }
  117. }else{
  118. fprintf(stderr, “Error: Unable to start mysql.\n”);
  119. rc = 1;
  120. }
  122. mysql_library_end();
  123. mosquitto_destroy(mosq);
  124. mosquitto_lib_cleanup();
  125. printf(“End!\n”);
  127. return rc;
  128. }








说明：




(1) 通配符#的使用




mosquitto_subscribe(mosq, NULL, “#”, 0);




在订阅中使用通配符可以收到多个主题的消息，通配符的详细用法可以看MQTT文档。




(2) CTRL+c正常结束程序




signal(SIGINT, handle_signal); // 捕捉终端CTRL+c产生的SIGINT信号




通过捕捉终端操作CTRL+c使程序能够正常退出以释放资源，而不是用CTRL+z强行终止程序的运行。




(3) 使用异步连接mosquitto_connect_async




(4) 使用异步循环mosquitto_loop_start






    

  1. 

    

    

    

    

    while(run) {
    

    

    2. 

  2. 

    

    

    

    

    sleep(20);
    

    

    3. 

  3. 

    

    

    

    

    }
    

    

    4. 









    

  • 编辑mysql_log.c
    • 





nano mysql_log.c








3. 编译、测试




编译




make








    

  • 本地测试
    • 





./mosquitto_mysql_log








    

  • 发布消息
    • 





再打开一个终端窗口，发送：




mosquitto_pub -p 1883 -u ct -P 1qaz2wsx -t temperature001 -m “25.1”








    

  • 查看订阅窗口
    • 









    

  • 远程测试
    • 





详细说明见《树莓派MQTT服务远程测试MQTT.fx》




https://zhuanlan.zhihu.com/p/363373024




https://blog.csdn.net/chentuo2000/article/details/115539377








点击Publish。




    

  • 订阅测试窗口收到消息
    • 









    

  • 退出程序
    • 





CTRL+c正常退出：








    

  • 查看数据库
    • 





select _ from temperature where deviceid like ‘temperature%’;