【嵌入式】MQTT

MQTT

安装

安装Paho MQTT C库:

sudo apt-get install libpaho-mqtt3-dev

头文件包含：

#include "MQTTClient.h"

编译选项：

gcc -o $@ $^ -lpaho-mqtt3c

简介

MQTT协议全称是（Message Queuing Telemetry Transport），即消息队列遥测传输协议。

是一种基于发布/订阅（Publish/Subscribe）模式的轻量级通讯协议，并且该协议构建于TCP/IP协议之上，TCP协议本身就具有高可靠性的特点，因此基于其上的MQTT协议同样也是具有高可靠、低开销的特点，之所以低开销，是以为MQTT协议传输的最小的报文也只有两个字节。

在物联网开发中，MQTT不是唯一的选择，与MQTT互相竞争的协议有XMPP和CoAP协议等。

关于发布和订阅的概念我们拿抖音平台来举个例子，我们每一个用户就都是一个客户端，而抖音就是MQTT协议中的服务器，当我们（用户一）关注某一个视频发布者（用户二）时，这样一个关注的行为就可以理解为订阅；同时用户二也可以关注你，那么这就是相互订阅。当用户二发布作品的时候，这个作品是发布到了抖音平台，也就是我们现在的服务器，这个过程就是消息的发布。

在这里需要注意的是：用户二（客户端）发布的消息并不是直接发布给了用户一，而是发布到了抖音平台（服务器），由于用户一订阅了用户二的消息（相当于点了关注），所以抖音平台（服务器）就会向用户一推送这个消息(注意发布和推送的区别)。这就是MQTT协议订阅&发布的一个简单比喻。

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分

• Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）
• payload，可以理解为消息的内容，是指订阅者具体要使用的内容

MQTT客户端代码

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"#define ADDRESS "thirdparty.mqtt.yoplore.com"  //服务器的IP地址
#define CLIENTID "client01"   //发布者的姓名(唯一的，如果发布者和订阅者用同一个姓名，就会出现顶号的现象)
#define TOPIC1 "MQTT topic/topic1"              //订阅主题
#define TOPIC2 "MQTT topic/topic2"              //订阅主题
#define QOS 1                              //服务登记（0.最多一次，1.最少一次，2.确保一次）
#define TIMEOUT 10000L                     //响应时间//定义一个传递令牌
volatile MQTTClient_deliveryToken deliveredtoken;//令牌交付回调函数，当消息成功交付给 MQTT 服务器时调用
/**context：用户自定义的上下文指针，此处未使用dt：消息交付的令牌
*/
void delivered(void *context, MQTTClient_deliveryToken dt)
{printf("Message with token value %d delivery confirmed\n", dt);deliveredtoken = dt;
}//接受订阅信息的回调函数，当接收到订阅主题的消息时调用
/**context：用户自定义的上下文指针，此处未使用topicName：接收到消息的主题名称topicLen：主题名称的长度message：接收到的 MQTT 消息结构体指针
*/
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{int i;char *payloadptr;printf("Message arrived\n");printf("topic: %s\n", topicName);printf("message: ");payloadptr = message->payload;for (i = 0; i < message->payloadlen; i++){putchar(*payloadptr);payloadptr++;}putchar('\n');MQTTClient_freeMessage(&message);MQTTClient_free(topicName);return 1;
}
//断开链接的回调函数
void connlost(void *context, char *cause)
{printf("\nConnection lost\n");printf("cause: %s\n", cause);
}int main(int argc, char *argv[])
{printf("\nCreating MQTTClient\n");// 消息缓冲区char buf[1024];// 1、定义一个MQTT客户端结构体指针MQTTClient client;// 2、创建一个MQTT客户端MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;MQTTClient_create(&client, ADDRESS, CLIENTID,MQTTCLIENT_PERSISTENCE_NONE, NULL);conn_opts.keepAliveInterval = 20;// 连接保活时间conn_opts.cleansession = 1; // 设置是否清除会话，1为清除// 定义一个 MQTT 消息结构体，用于存储要发布的消息MQTTClient_message publish_msg=MQTTClient_message_initializer;// 令牌tokenMQTTClient_deliveryToken token;// 设置回调MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);// 链接int rc = MQTTClient_connect(client, &conn_opts);if (rc != MQTTCLIENT_SUCCESS) {printf("Failed to connect, return code %d\n", rc);return EXIT_FAILURE;}// 订阅多个主题rc = MQTTClient_subscribe(client, TOPIC1, QOS);if (rc != MQTTCLIENT_SUCCESS) {printf("Failed to subscribe to %s, return code %d\n", TOPIC1, rc);return EXIT_FAILURE;}rc = MQTTClient_subscribe(client, TOPIC2, QOS);if (rc != MQTTCLIENT_SUCCESS) {printf("Failed to subscribe to %s, return code %d\n", TOPIC2, rc);return EXIT_FAILURE;}//用户退出char ch;while (1) {// 发送信息printf("请输入要发布的内容（输入 'q' 或 'Q' 退出）：\n");if (fgets(buf, sizeof(buf), stdin) == NULL) {printf("读取输入失败\n");continue;}// 去除换行符(fgets会将换行符一并读取)size_t len = strlen(buf);if (len > 0 && buf[len - 1] == '\n') {buf[len - 1] = '\0';}// 检查是否退出if (buf[0] == 'q' || buf[0] == 'Q') {break;}publish_msg.payload = (void *)buf;publish_msg.payloadlen = strlen(buf);rc = MQTTClient_publishMessage(client, TOPIC2, &publish_msg, &token);//用于将消息发布到指定的主题if (rc != MQTTCLIENT_SUCCESS) {printf("Failed to publish message, return code %d\n", rc);continue;}rc = MQTTClient_waitForCompletion(client, token, 1000); //用于等待指定的消息交付完成if (rc != MQTTCLIENT_SUCCESS) {printf("Failed to wait for message completion, return code %d\n", rc);continue;}printf("buf中的内容: %s\n", buf);}MQTTClient_disconnect(client,10000);MQTTClient_destroy(&client);printf("\nExiting\n");return 0;
}