一、简介
MQTT(Message Queuing Telemetry Transport)是一种轻量级的、基于发布/订阅模式的通信协议,旨在实现物联网设备之间的低带宽、高延迟的通信。MQTT协议设计简洁,使用TCP/IP协议进行通信,适用于各种网络环境,尤其适合在有限的网络带宽和不稳定的网络连接条件下进行通信。
MQTT的工作原理是基于发布/订阅模式的消息传递,它包括两个主要角色:发布者(Publisher)和订阅者(Subscriber)。发布者负责发布消息到主题(Topic),而订阅者则可以订阅感兴趣的主题,从而接收相应的消息。这种模式使得设备之间可以实现异步通信,同时降低了对网络和系统资源的占用。MQTT的两个常用的开源库,mosquitto和emqx。本文服务端使用的是emqx,而客户端则用mosquitto开源库。
二、通信时序框图
三、服务器搭建
- 创建MQTT服务
可以进入emqx官网参考下载过程。
按照官网执行以下三个步骤,即可打开emqx服务。
curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash
sudo apt-get install emqx
sudo systemctl start emqx
如果要查看服务开启状态,则使用以下命令。
sudo systemctl status emqx
如果需要关闭服务,则使用以下命令。
sudo systemctl stop emqx
- 服务测试
首先安装mosquitto客户端。
sudo apt install mosquitto-clients
先打开一个终端,输入以下命令。
mosquitto_sub -h localhost -t "topic"
再打开另外一个终端,输入以下命令,如果可以在前一个终端界面上看到发送的信息,则说明服务正常开启。
mosquitto_pub -h localhost -t "topic" -m "Hello, MQTT"
四、客户端代码实现
这里我们实现两个客户端用来相互通信。
客户端1代码
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <mosquitto.h>
#define MQTT_SERVER_ADDRESS "127.0.0.1"
#define MQTT_SERVER_PORT 1883
#define MQTT_TOPIC "topic"
struct mosquitto *mosq = NULL;
void on_connect(struct mosquitto *mosq, void *obj, int rc)
{
if (rc == 0) {
printf("Connected to MQTT broker\n");
} else {
fprintf(stderr, "Failed to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
exit(EXIT_FAILURE);
}
}
void on_publish(struct mosquitto *mosq, void *obj, int mid)
{
printf("Message published\n");
}
int main()
{
int rc;
int protocol_version = MQTT_PROTOCOL_V311;
mosquitto_lib_init(); // 初始化Mosquitto库
mosq = mosquitto_new(NULL, true, NULL);
if (!mosq) {
fprintf(stderr, "Error: Out of memory.\n");
return EXIT_FAILURE;
}
mosquitto_connect_callback_set(mosq, on_connect);
mosquitto_publish_callback_set(mosq, on_publish);
// 设置MQTT协议版本
mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);
rc = mosquitto_connect(mosq, MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60); // 连接到MQTT broker
if (rc != MOSQ_ERR_SUCCESS) {
fprintf(stderr, "Unable to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
return EXIT_FAILURE;
}
printf("Connected to MQTT broker\n");
while (1) {
char payload[100];
printf("Enter message: ");
fgets(payload, sizeof(payload), stdin);
payload[strcspn(payload, "\n")] = '\0';
rc = mosquitto_publish(mosq, NULL, MQTT_TOPIC, strlen(payload), payload, 0, false);
if (rc != MOSQ_ERR_SUCCESS) {
fprintf(stderr, "Failed to publish message: %s\n", mosquitto_strerror(rc));
}
}
mosquitto_destroy(mosq); // 销毁Mosquitto对象
mosquitto_lib_cleanup(); // 清理Mosquitto库
return 0;
}
客户端2代码
#include <stdio.h>
#include <stdlib.h>
#include <mosquitto.h>
#define MQTT_SERVER_ADDRESS "localhost"
#define MQTT_SERVER_PORT 1883
#define MQTT_TOPIC "topic"
struct mosquitto *mosq = NULL;
void on_connect(struct mosquitto *mosq, void *obj, int rc)
{
if (rc == 0) {
printf("Connected to MQTT broker\n");
mosquitto_subscribe(mosq, NULL, MQTT_TOPIC, 0); // 订阅主题
} else {
fprintf(stderr, "Failed to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
exit(EXIT_FAILURE);
}
}
void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
printf("Received message: %.*s\n", (int)msg->payloadlen, (char *)msg->payload);
}
int main()
{
int rc;
int protocol_version = MQTT_PROTOCOL_V311;
mosquitto_lib_init(); // 初始化Mosquitto库
mosq = mosquitto_new(NULL, true, NULL);
if (!mosq) {
fprintf(stderr, "Error: Out of memory.\n");
return EXIT_FAILURE;
}
mosquitto_connect_callback_set(mosq, on_connect);
mosquitto_message_callback_set(mosq, on_message);
// 设置MQTT协议版本
printf("protocol version is:%d\n", protocol_version);
mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);
rc = mosquitto_connect(mosq, MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60); // 连接到MQTT broker
if (rc != MOSQ_ERR_SUCCESS) {
fprintf(stderr, "Unable to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
return EXIT_FAILURE;
}
printf("Connected to MQTT broker\n");
mosquitto_loop_start(mosq); // 启动MQTT客户端循环
while (1) {
// 保持客户端运行
}
mosquitto_disconnect(mosq); // 断开连接
mosquitto_destroy(mosq); // 销毁Mosquitto对象
mosquitto_lib_cleanup(); // 清理Mosquitto库
return 0;
}
- 本地编译测试
gcc -o MyMQTT *.c -lpthread -lmosquitto
五、交叉编译
先上mosquitto官网下载mosquitto
源码,这里下载的是当前最新版本2.0.18。
解压,并进入文件夹。
tar xvf mosquitto-2.0.18.tar.gz
cd mosquitto-2.0.18
打开config.mk
配置文件,将CJSON
注释掉,并且把编译静态库功能打开。
如果不用SSL
加密,把WITH_TLS:=yes
也注释掉,不然就得再装个openssl
。
执行make
命令进行编译,注意这里要使用交叉编译工具,即把CC替换成交叉编译gcc,把CXX替换成交叉编译g++。
make CC=/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-gcc CXX=/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-g++
然后再执行安装命令,这里注意加上sudo
权限。
sudo make install
安装后库会在/usr/local/lib
中,头文件会在/usr/local/include
中。
然后回到上面的例子工程中,使用交叉编译工具进行编译,编译的时候指定链接静态库mosquitto
。
/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-gcc -static -o MyMQTT *.c -lpthread -L/usr/local/lib -lmosquitto -I/usr/local/include
六、注意事项
1、使用mosquitto_connect
接口进行MQTT连接时,第4个参数是保持心跳持续时间,这个值不能设为0,不然一直连接不上。
2、注意协议版本号,需要保持通信双方协议版本号一致,正常默认是V3.11。
3、如果要使用交叉编译时,需要安装宿主机对应架构的mosquitto
库,比如需要交叉编译arm架构的,那需要手动下载mosquitto
的源码包,编译生成对应的库,编译时链接到该库。