1. 概述


在《用C语言实现mosquitto MQTT订阅消息》


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


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


一文中我们用C语言实现了mosquitto MQTT同步订阅消息。mosquitto的同步函数是以阻塞方式工作的，也就是订阅程序一直等待接收消息，阻塞了其他程序的运行，效率很低。


下面我们用异步mosquitto的函数实现MQTT消息订阅，异步是非阻塞的方式，比同步方式性能更好。


2. 修改例子subscribe


2.1 异步与同步程序的差异


  • 异步方式连接服务器的函数mosquitto_connect_async

该函数连接MQTT代理。这是一个非阻塞调用。如果使用mosquitto_connect_async，则客户端必须使用线程接口mosquitto_loop_start。


注意：如果代码中直接使用mosquitto_loop循环，则必须使用同步mosquitto_connect函数连接代理。


例如用下面的循环代替mosquitto_loop_forever函数时


        while(running) {


                mosquitto_loop(mosq, -1, 1);


        }


  • 异步循环函数mosquitto_loop_start

此函数开启一个新线程，在线程里循环调用 mosquitto_loop。


而同步循环函数mosquitto_loop_forever在无限阻塞循环中调用mosquitto_loop。


mosquitto_loop是客户端的主循环函数，必须经常调用它以保持客户机和代理之间的通信正常工作。mosquitto_loop_forever和mosquitto_loop_start都是通过调用mosquitto_loop来实现的。


你也可以直接使用此函数，但不能在回调中调用它。


例如上面用循环mosquitto_loop代替mosquitto_loop_forever函数的例子。


2.2 重写C程序


  1. #include <mosquitto.h>
  2. #include <stdio.h>
  3. #include <stdlib.h>
  4. #include <string.h>
  5. #include <unistd.h>
  8. // 定义运行标志决定是否需要结束
  9. static int running = 1;
  11. // 当客户端从代理接收到CONNACK消息时调用回调
  12. void on_connect(struct mosquitto _mosq, void _obj, int reason_code)
  13. {
  14. int rc;
  16. printf(“on_connect: %s\n”, mosquitto_connack_string(reason_code));
  17. if(reason_code != 0){
  18. mosquitto_disconnect(mosq);
  19. }
  21. rc = mosquitto_subscribe(mosq, NULL, “example/temperature”, 1);
  22. if(rc != MOSQ_ERR_SUCCESS){
  23. fprintf(stderr, “Error subscribing: %s\n”, mosquitto_strerror(rc));
  24. mosquitto_disconnect(mosq);
  25. }
  26. }
  28. // 当代理在响应订阅发送SUBACK时调用回调
  29. void on_subscribe(struct mosquitto _mosq, void _obj, int mid, int qos_count, const int _granted_qos)
  30. {
  31. int i;
  32. bool have_subscription = false;
  34. for(i=0; i<qos_count; i++){
  35. printf(“on_subscribe: %d:granted qos = %d\n”, i, granted_qos[i]);
  36. if(granted_qos[i] <= 2){
  37. have_subscription = true;
  38. }
  39. }
  40. if(have_subscription == false){
  41. fprintf(stderr, “Error: All subscriptions rejected.\n”);
  42. mosquitto_disconnect(mosq);
  43. }
  44. }
  46. // 当客户端收到消息时调用回调该函数
  47. void on_message(struct mosquitto _mosq, void _obj, const struct mosquitto_message _msg)
  48. {
  49. // 打印有效载荷
  50. printf(“%s %d %s\n”, msg->topic, msg->qos, (char _)msg->payload);
  51. }
  53. // 当断开连接时调用回调该函数
  54. void on_disconnect(struct mosquitto _mosq, void _obj, int rc)
  55. {
  56. printf(“Call the function: my_disconnect_callback\n”);
  57. running = 0;
  58. }
  60. int main(int argc, char _argv[])
  61. {
  62. struct <span class=”hljs-title class_“>mosquitto *mosq;
  63. int rc;
  65. // 初始化mosquitto库
  66. mosquitto_lib_init();
  68. // 创建新的客户端实例。
  69. mosq = mosquitto_new(NULL, true, NULL);
  70. if(mosq == NULL){
  71. fprintf(stderr, “Error: Out of memory.\n”);
  72. return 1;
  73. }
  75. // 配置回调函数
  76. mosquitto_connect_callback_set(mosq, on_connect);
  77. mosquitto_subscribe_callback_set(mosq, on_subscribe);
  78. mosquitto_message_callback_set(mosq, on_message);
  79. mosquitto_disconnect_callback_set(mosq, on_disconnect);
  81. // 连接服务器
  82. mosquitto_username_pw_set(mosq, “ct”, “1qaz2wsx”);
  83. rc = mosquitto_connect_async (mosq, “raspberrypi”, 1883, 60);
  84. if(rc != MOSQ_ERR_SUCCESS){
  85. mosquitto_destroy(mosq);
  86. fprintf(stderr, “Error: %s\n”, mosquitto_strerror(rc));
  87. return 1;
  88. }
  90. // 异步循环
  91. rc = mosquitto_loop_start(mosq);
  92. if(rc != MOSQ_ERR_SUCCESS)
  93. {
  94. mosquitto_destroy(mosq);
  95. fprintf(stderr, “Error: %s\n”, mosquitto_strerror(rc));
  96. return 1;
  97. }
  99. // 开始循环
  100. printf(“Start!\n”);
  101. while(running)
  102. {
  103. sleep(1);
  104. }
  106. // 结束后的清理工作
  107. mosquitto_destroy(mosq);
  108. mosquitto_lib_cleanup();
  109. printf(“End!\n”);
  111. return 0;
  112. }








 




2.3 修改例子




    

  • 进入例子目录
    • 





cd mosquitto-2.0.9/examples/subscribe/








    

  • 备份一下原来的代码
    • 





cp basic-1.c basic-1.c-bak




    

  • 编辑basic-1.c
    • 





nano basic-1.c




用重写的C程序替换原来的程序。




3.3 编译




    

  • 编译
    • 





gcc -o basic-1 basic-1.c -lmosquitto








3.4 测试




    

  • 运行basic-1
    • 





./basic-1








从rc = mosquitto_subscribe(mosq, NULL, “example/temperature”, 1);可知，订阅的消息主题为example/temperature。




    

  • 再开一个窗口发布消息
    • 





mosquitto_pub -p 1883 -u ct -P xxxxxxxx -t example/temperature -m “26.6”








    

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









    

  • 用MQTT.fx远程测试
    • 





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




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




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








点击Publish：




订阅测试窗口收到消息








 




参考文档




    

  1. MQTT通信协议(mosquitto)发布订阅例子C语言实现
    https://blog.csdn.net/qq_33406883/article/details/107466430
    2.