文章目录
😏1. MQTT介绍
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。
MQTT (Message Queuing Telemetry Transport) 是一种轻量级的消息传输协议，通常用于物联网设备和应用程序之间进行通信。它是基于发布/订阅模式设计的，其中消息发布者将消息发布到特定主题（Topic），然后订阅该主题的客户端将收到这些消息。MQTT 特别适合在网络带宽有限的情况下进行通信，因为它使用的数据包非常小。此外，它还提供了多种 QoS (Quality of Service) 级别来确保消息的可靠性和有效性。
MQTT 协议具有以下特点：
MQTT 支持三种不同级别的服务质量（Quality of Service，QoS），分别为 QoS0、QoS1 和 QoS2。
选择哪种服务质量级别取决于应用场景和对通信安全性的要求。需要注意的是，在选择高级别的服务质量时，会增加通信延迟和网络带宽的消耗。
MQTT数据包结构如下：
😊2. MQTT（Mosquitto）安装与测试
目前mqtt的代理平台主要有：Mosquitto、VerneMQ、EMQTT。
下面在Ubuntu安装Mosquitto来体验mqtt的消息传递过程：
sudo apt-get install mosquitto # 服务端
sudo apt install mosquitto-clients # 客户端
启动/关闭mqtt服务：
mosquitto -v # 启用所有日志记录类型
# 启动和关闭服务
sudo service mosquitto start
sudo service mosquitto stop
# 查看运行状态
sudo systemctl status mosquitto
# 查看帮助
mosquitto --help
#查看运行进程号：
ps -aux | grep mosquitto
#执行命令杀死进程：
kill -9 进程号
消息传输测试：
1、启动代理服务：mosquitto -v
-v 详细模式 打印调试信息（启动一次就好）
2、订阅主题：mosquitto_sub -v -t hello
-t 指定订阅的主题，主题为：hello
-v 详细模式 打印调试信息
3、发布内容：mosquitto_pub -t hello -m world
-t 指定订阅的主题，主题为：hello
-m 指定发布的消息的内容
此外，还可以去官网（https://mosquitto.org/）安装最新版源码并编译，编译后的版本适合用于开发。
MQTT在线测试工具：https://mqttx.app/zh
😆3. MQTT（Eclipse Paho）库安装与实例
Eclipse Paho C++是一个跨平台的MQTT客户端实现，可以用于在C++应用程序中发送和接收MQTT消息。它支持许多不同的传输协议（如TCP、WebSocket、SSL等），并提供了多种QoS级别。开发中常使用这个。
github地址：
https://github.com/eclipse/paho.mqtt.c
https://github.com/eclipse/paho.mqtt.cpp
安装步骤：
# 依赖
sudo apt-get install build-essential gcc make cmake libssl-dev doxygen graphviz
# c
git clone https://github.com/eclipse/paho.mqtt.c
cd paho.mqtt.c
git checkout v1.3.8
cmake -Bbuild -H. -DPAHO_ENABLE_TESTING=OFF -DPAHO_BUILD_STATIC=ON \
-DPAHO_WITH_SSL=ON -DPAHO_HIGH_PERFORMANCE=ON
sudo cmake --build build/ --target install
sudo ldconfig
# c++
git clone https://github.com/eclipse/paho.mqtt.cpp
cd paho.mqtt.cpp
cmake -Bbuild -H. -DPAHO_BUILD_STATIC=ON \
-DPAHO_BUILD_DOCUMENTATION=TRUE -DPAHO_BUILD_SAMPLES=TRUE
sudo cmake --build build/ --target install
sudo ldconfig
客户端代码示例：
mqtt_client.hpp
#ifndef __MQTT_CLIENT_TO_CLOUD_HPP__
#define __MQTT_CLIENT_TO_CLOUD_HPP__
#include <mqtt/async_client.h> // mqtt库头文件
#include <mqtt/topic.h>
namespace cloud {
// Handler on cloud message
using message_handler = std::function<void(const std::string&)>;
class mqtt_client
{
public:
mqtt_client();
~mqtt_client();
void send(const std::string& message);
void set_message_handler(message_handler cb);
private:
// static constexpr const char* BROKER_HOST = "localhost:1883"; //本地测试：mosquitto
static constexpr const char* BROKER_HOST = "broker.emqx.io:1883"; //公共mqtt broker：MQTTX
// static constexpr const char* BROKER_HOST = "124.XXX.XXX.XXX:1883"; //云端测试：mosquitto
private:
mqtt::async_client cli_;
mqtt::topic topic_;
};
} // namespace cloud
#endif
mqtt_client.cpp
#include <iostream>
#include <string>
#include "mqtt_client.hpp"
using namespace std;
namespace cloud {
// mqtt_client类构造函数实现
mqtt_client::mqtt_client()
// 1.The server URI string 2.The client ID string that we provided to the server 3.The MQTT protocol version we're connected at
: cli_(BROKER_HOST, "client", mqtt::create_options(MQTTVERSION_5))
// 1.The client to which this topic is connected 2.The topic name(pub & sub) 3.The default QoS
, topic_(cli_,"haojuhu", 1)
{
//! Handler on connection lost, do reconnect here
cli_.set_connection_lost_handler([this](const string& info) {
std::cout<<"mqtt connection lost <" << info << ">, reconnting"<<std::endl;
cli_.reconnect();
});
//! Handler on connected, it'll subscribe the topic and publish online info
cli_.set_connected_handler([this](const string& info) {
std::cout << "mqtt connected <" << info << ">"<<std::endl;
topic_.subscribe(mqtt::subscribe_options(true)); // client订阅topic[haojuhu]
topic_.publish("online"); // client发布消息"online"至topic[haojuhu]
});
}
//2.mqtt_client类析构函数实现
mqtt_client::~mqtt_client()
{
cli_.disconnect();
cli_.disable_callbacks();
}
/**
* @brief Publish message to topic，发布消息给topic
* @param[in] message The message payload
*/
void mqtt_client::send(const string& message)
{
topic_.publish(message);
}
/**
* @brief Set mqtt message handler，设置mqtt消息处理句柄（也就是函数对象cb）
* @note The mqtt connection will established here
* @param[in] cb The message handler
*/
void mqtt_client::set_message_handler(message_handler cb)
{
//! Set message callback here
cli_.set_message_callback([cb](mqtt::const_message_ptr message)
{
cb(message->get_payload_str()); //执行函数cb
});
//! Set connect options and do connect
auto opts = mqtt::connect_options_builder()
.mqtt_version(MQTTVERSION_5)
.clean_start(true)
.finalize();
cli_.connect(opts);
}
} // namespace cloud
main.cpp
#include <iostream>
#include <string>
#include <map>
#include "mqtt_client.hpp"
using namespace std;
/**
* @brief 消息处理
* @param[in] data The message payload from cloud
*/
void on_cloud_message(const string& data)
{
// 接收数据
std::cout<<"received data is: "<<data<<std::endl;
}
int main(int argc,char **argv)
{
cloud::mqtt_client g_client; // 定义一个mqtt客户端
std::cout << "[CLOUD] listen starting"<<std::endl;
g_client.set_message_handler(on_cloud_message); // 开启mqtt clinet监听消息，消息处理函数为on_cloud_message
while(1)
{
std::cout << "运行中..."<<std::endl;
this_thread::sleep_for(10s);
g_client.send("online..."); // 确保与mqtt broker server建立连接之后再publish!!!
}
return 0;
}
编译运行：
g++ main.cpp mqtt_client.cpp -o main -lpaho-mqttpp3 -lpaho-mqtt3a
./main
运行如下：
[CLOUD] listen starting
运行中...
mqtt connected <connect onSuccess called>
以上。
