MQTT 常用客户端库介绍 (全面涵盖c,c++,java,c#,python)

2023-09-08 10:04

短信预约 -IT技能 免费直播动态提醒

MQTT（Message Queuing Telemetry Transport）是一种轻量级的通信协议，被广泛应用于物联网和分布式系统中。它以其简单、可靠和高效的特性而备受推崇，成为连接设备和应用程序的首选协议。MQTT的重要性不言而喻，它为实时通信提供了可靠的解决方案，使得设备之间的数据交换和控制变得更加便捷。

本文介绍几种常用的MQTT客户端库，帮助您快速上手使用MQTT协议，构建稳定可靠的通信系统。无论您是开发物联网设备、构建分布式系统，还是进行实时数据传输，了解和掌握MQTT客户端库是至关重要的。

MQTT协议简介

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的即时通讯协议，物联网通信的主流协议之一。该协议支持所有平台，几乎可以把所有联网物品和外部连接起来，被用来当做传感器和制动器的通信协议。

MQTT协议的底层是基于TCP/IP协议。TCP/IP是一组用于在网络上进行通信的协议，包括TCP（传输控制协议）和IP（Internet协议）等。MQTT使用TCP作为底层传输协议，利用TCP的可靠性和连接性来确保消息的可靠传输。MQTT客户端通过TCP连接到MQTT代理服务器，并在TCP连接上传输MQTT消息。TCP协议提供了可靠的字节流传输，确保消息的顺序和完整性。因此，MQTT协议在TCP/IP协议栈上运行，利用TCP/IP提供的网络通信能力来实现消息的传输。

MQTT协议是为大量计算能力有限，且工作在低带宽、不可靠的网络的远程传感器和控制设备通讯而设计的协议，它具有以下主要特性：

发布/订阅模式：MQTT采用发布/订阅模式，其中消息发布者（Publisher）将消息发布到特定的主题（Topic），而订阅者（Subscriber）通过订阅特定的主题来接收消息。

主题（Topic）：主题是MQTT消息的分类标识，类似于消息的目的地。发布者将消息发布到特定的主题，而订阅者通过订阅特定的主题来接收与该主题相关的消息。

消息质量等级（QoS）：MQTT定义了三个消息质量等级：QoS 0（最多一次）、QoS 1（至少一次）和QoS 2（只有一次）。发布者和订阅者可以根据需求选择适当的QoS级别。

保留消息（Retained Message）：发布者可以选择发布一个保留消息，该消息将被存储在MQTT代理服务器上，并在有订阅者订阅该主题时发送给订阅者。

遗嘱消息（Will Message）：发布者可以设置一个遗嘱消息，当发布者异常断开连接时，MQTT代理服务器会将遗嘱消息发送给订阅者。

连接和会话：MQTT客户端通过与MQTT代理服务器建立连接来进行通信。客户端可以选择保持连接以维护会话状态，并在重新连接时恢复之前的会话。

低带宽和资源消耗：MQTT协议设计为轻量级，具有较低的带宽和资源消耗。它适用于低带宽和不稳定的网络环境，如物联网设备和传感器网络。

MQTT协议，终于有人讲清楚了

MQTT Version 3.1.1协议介绍

：MQTT Version 3.1.1

MQTT 3.1.1 协议中文版 | MQTT中文网

MQTT常用代理服务器

MQTT常用的代理服务器有多种选择，以下是几个常见的MQTT代理服务器的介绍：

Mosquitto： Mosquitto是一个轻量级的MQTT代理服务器，它是开源的，支持MQTT 3.1和MQTT 3.1.1协议。Mosquitto易于安装和配置，适用于各种规模的应用。它提供了可靠的消息传递和订阅功能，并支持安全认证、访问控制等特性。

HiveMQ： HiveMQ是一个功能丰富且高度可扩展的MQTT代理服务器，适用于大规模的MQTT通信。HiveMQ支持MQTT 3.1和MQTT 3.1.1协议，提供了可靠的消息传递、集群支持、安全认证、访问控制、实时监控等特性。它还提供了扩展插件机制，允许用户根据需要自定义功能。

EMQ X： EMQ X是一个开源的分布式MQTT代理服务器，支持MQTT 3.1、MQTT 3.1.1和MQTT 5.0协议。EMQ X具有高可用性和高性能的特点，适用于大规模的物联网应用场景。它提供了集群支持、安全认证、访问控制、数据持久化、实时监控等功能。EMQX: 大规模分布式 MQTT 消息服务器

RabbitMQ： RabbitMQ是一个通用的消息代理服务器，它支持多种消息传递协议，包括AMQP、MQTT、STOMP等。RabbitMQ提供了可靠的消息传递和订阅功能，并具有高可用性、可扩展性和灵活性。它支持多种操作系统和编程语言，适用于各种不同的应用场景。

5.NanoMQ: NanoMQ (opens new window)是EMQX于 2021 年初发布的边缘计算开源项目，是面向物联网边缘计算场景的下一代轻量级高性能 MQTT 消息服务器。Github 仓库地址: GitHub - emqx/nanomq: An ultra-lightweight and blazing-fast MQTT broker for IoT edge

以上是一些常用的MQTT代理服务器，每个代理服务器都有其特点和适用场景。选择合适的代理服务器取决于具体需求，如规模、性能要求、安全性等。在选择代理服务器时，还应考虑其可靠性、社区支持以及易于使用和配置的程度。

MQTT常用客户端库

MQTT的常用客户端库有很多，下面分别介绍C语言、C++、Java和Python的常用客户端库：

c语言客户端库

自己实现：tcp连接，发包，收包解析。物联网MQTT协议报文解析(简单的c语音客户端实现)_特立独行的猫a

Paho MQTT C：Eclipse Paho项目的C语言实现，提供了基本的MQTT功能，支持多种平台。

libmosquitto：一个轻量级的MQTT客户端库，用C语言编写，支持多种平台。

paho.mqtt.c介绍

paho.mqtt.c 是Eclipse编写的开源mqtt c库，支持Posix标准操作系统(如Linux，Android，Mac)和windows操作系统。

Paho MQTT C客户端支持全部MQTT协议客户端特性，它使用ANSI标准C编写。

实际上这个库提供两套API，分别是"同步"的MQTTClient和“异步”的MQTTAsync,同步API目的是更加简单、更加有用的。为了达到这个目的，部分操作将被阻塞，直到这个操作完成，这样程序的框架更加简单。

相反，在异步模式中只有一个调用会被阻塞——waitForCompletion。异步模式都是通过回调函数返回结果，这样更适合那些不是控制主线程的应用。

源码下载：https://github.com/eclipse/paho.mqtt.c (速度很慢)

国内的源码镜像：https://gitee.com/mirrors/paho.mqtt.c (速度起飞)

官方文档：https://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/index.html

实际上如果用在资源受限的嵌入式平台上，常用的是另外的一个库 paho.mqtt.embedded-c。

它们之间的区别是：

Paho MQTT C是一个通用的C语言MQTT客户端库，它提供了一套API和功能，可以在多种平台上使用，包括Windows、Linux、macOS等。它的设计目标是提供一个跨平台的MQTT客户端库，以便在不同的环境中实现MQTT通信。 paho.mqtt.embedded-c则是专门为嵌入式系统和资源受限设备设计的MQTT C语言实现。它专注于在嵌入式系统中使用，提供了一套轻量级的API和功能，以适应资源有限的设备。

paho.mqtt.embedded-c开源地址：GitHub - eclipse/paho.mqtt.embedded-c: Paho MQTT C client library for embedded systems. Paho is an Eclipse IoT project (https://iot.eclipse.org/)

libmosquitto介绍

libmosquitto是一个轻量级的MQTT客户端库，用于在C语言环境下实现MQTT通信。它提供了一套简单易用的API，可以用于连接MQTT代理服务器、发布和订阅消息等操作。以下是关于libmosquitto的一些介绍和使用示例： 1. 官方地址：

libmosquitto的官方地址为：Eclipse Mosquitto

安装libmosquitto：可以通过源代码编译安装libmosquitto，也可以使用包管理工具进行安装。具体安装方式可以参考官方文档。 3. 使用libmosquitto：下面是一个简单的libmosquitto使用示例，展示了如何连接MQTT代理服务器、发布消息和订阅消息：

#include #include  void on_connect(struct mosquitto *mosq, void *userdata, int rc) {    if(rc == 0) {        printf("Connected to MQTT broker!\n");        mosquitto_subscribe(mosq, NULL, "test/topic", 0);    } else {        printf("Connection failed: %s\n", mosquitto_strerror(rc));    }} void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message) {    printf("Received message: %s\n", (char *)message->payload);} int main() {    struct mosquitto *mosq = NULL;    mosquitto_lib_init();     mosq = mosquitto_new(NULL, true, NULL);    if(mosq) {        mosquitto_connect_callback_set(mosq, on_connect);        mosquitto_message_callback_set(mosq, on_message);        mosquitto_connect(mosq, "localhost", 1883, 60);         mosquitto_loop_start(mosq);         mosquitto_publish(mosq, NULL, "test/topic", strlen("Hello, MQTT!"), "Hello, MQTT!", 0, false);         getchar();  // 按下Enter键退出程序         mosquitto_disconnect(mosq);        mosquitto_destroy(mosq);    }     mosquitto_lib_cleanup();    return 0;}

上述示例代码创建了一个MQTT客户端，连接到本地的MQTT代理服务器（localhost:1883），发布了一条消息到"test/topic"主题，并订阅了该主题。当接收到消息时，会打印出来。通过按下Enter键，可以退出程序。这只是一个简单的示例，libmosquitto还提供了更多的功能和选项，如设置用户名密码、TLS/SSL加密、设置遗嘱消息等。更详细的使用方法可以参考libmosquitto的官方文档和示例代码。

注意事项

在MQTT中，心跳是用来维持客户端与MQTT代理服务器之间的连接活跃状态的机制。通过定期发送心跳包，客户端可以告知服务器自己的在线状态，如果长时间没有收到心跳包，服务器会认为客户端已经断开连接并关闭连接。在使用MQTT客户端库时，通常会有心跳机制的支持，可以通过设置心跳间隔来维持连接的活跃状态。心跳间隔的设置取决于具体的MQTT客户端库和应用场景。

以下是一些常见MQTT客户端库的心跳设置示例：

   mosquitto_connect(mosq, "localhost", 1883, 60);   mosquitto_loop_start(mosq);   mosquitto_publish(mosq, NULL, "topic", strlen("message"), "message", 0, false);   mosquitto_loop_stop(mosq, true);

在上述示例中，通过设置 60 作为心跳间隔，客户端会每60秒发送一次心跳包来维持连接。需要注意的是，心跳间隔的设置应该根据具体的应用场景和MQTT代理服务器的要求进行调整。如果长时间不操作，连接可能会被服务器断开，因此需要根据需求设置合适的心跳间隔来保持连接的稳定性。

c++客户端库

Paho MQTT C++：Eclipse Paho项目的C++语言实现，提供了面向对象的API，支持多种平台。

https://github.com/eclipse/paho.mqtt.cpp

mirrors / eclipse / paho.mqtt.cpp · GitCode

MQTTClient++：一个开源的C++ MQTT客户端库，提供了简单易用的API，支持多种平台。

mosquitto

GitHub - kurdybacha/mqttcpp: MQTT Client C++ library based on libmosquitto

mosqpp::mosquittopp是一个基于Paho MQTT C++的MQTT C++客户端库，提供了面向对象的API，用于在C++环境下实现MQTT通信。

windows下可以使用vcpkg轻松简单的安装。带x64-的为64位版本的，不带的默认为32位版本库。

./vcpkg.exe install  mosquitto

./vcpkg.exe install acl:x64-windows acl asio:x64-windows asio breakpad:x64-windows breakpad crashrpt:x64-windows crashrpt cxxopts:x64-windows cxxopts fast-cpp-csv-parser:x64-windows fast-cpp-csv-parser g3log:x64-windows g3log gtest:x64-windows gtest iniparser:x64-windows iniparser libconfig:x64-windows libconfig linenoise-ng:x64-windows linenoise-ng mosquitto rapidjson:x64-windows rapidjson inih simpleini:x64-windows simpleini rpclib rpclib:x64-windows opencv3 --no-binarycaching

下面是mosqpp::mosquittopp的简单使用示例：

#include #include #include  class MyMqttClient : public mosqpp::mosquittopp {public:    MyMqttClient(const char* id) : mosquittopp(id) {}     void on_connect(int rc) {        if (rc == 0) {            std::cout << "Connected to MQTT broker" << std::endl;            subscribe(NULL, "test/topic");        } else {            std::cout << "Failed to connect to MQTT broker" << std::endl;        }    }     void on_message(const struct mosquitto_message* message) {        std::cout << "Received message: " << static_cast(message->payload) << std::endl;    }}; int main() {    MyMqttClient client("client1");     client.connect("broker.hivemq.com");    client.loop_start();     while (true) {        client.publish(NULL, "test/topic", strlen("Hello, MQTT!"), "Hello, MQTT!");        std::this_thread::sleep_for(std::chrono::seconds(1));    }     client.loop_stop();    client.disconnect();     return 0;}

上述示例中，首先定义了一个继承自mosquittopp的MyMqttClient类。在这个类中，重写了on_connect()和on_message()方法，分别处理连接成功和接收到消息的回调。

在on_connect()方法中，打印连接成功的消息并订阅了"test/topic"主题。在on_message()方法中，打印接收到的消息内容。在main()函数中，我们创建了一个MyMqttClient对象，并通过connect()方法连接到MQTT代理服务器。然后通过loop_start()方法启动MQTT循环，使客户端可以接收和处理消息。在循环中，我们每秒钟发布一条消息到"test/topic"主题。最后，通过loop_stop()和disconnect()方法停止MQTT循环并断开连接。

使用简单封装

#include "mosquittopp.h"#include #include #include #include #include namespace mqtt_client{class MQTTClient final : public mosqpp::mosquittopp{public:  MQTTClient(const std::string& id, std::string host, int port, std::string title);  ~MQTTClient() override;public:  using onMessage = std::function;  void setReceiveCallback(const onMessage& f) { onMessage_ = f; }  bool connectToServer();  void connectToServer(std::error_code& errorCode);  const std::string& title() const { return title_; }private:  void on_connect(int rc) override;  void on_disconnect(int rc) override;  void on_error() override;  void on_subscribe(int mid, int qos_count, const int* granted_qos) override;  void on_message(const struct mosquitto_message* message) override;  void on_unsubscribe(int rc) override;private:  std::string title_;  std::string host_;  int port_;  onMessage onMessage_;};class MQTTClientPtr{public:  std::vector> ptr_;};}  // namespace mqtt_client

namespace mqtt_client{MQTTClient::MQTTClient(const std::string& id, std::string host, int port, std::string title)    :  title_(std::move(title)),      host_(std::move(host)),      port_(port),      onMessage_(nullptr){}MQTTClient::~MQTTClient(){  disconnect();  loop_stop();}void MQTTClient::on_connect(int rc){  if (rc == 0)  {    subscribe(NULL, title_.c_str());  }  else  {    LOGGING_ERROR("Connect - Code - %d", rc);  }}void MQTTClient::on_disconnect(int rc){  if (rc != 0)  {    LOGGING_ERROR("Disconnect - Code - %d", rc);  }}void MQTTClient::on_error(){}void MQTTClient::on_subscribe(int mid, int qos_count, const int* granted_qos){  LOGGING_WARN("Subscription succeeded.");}void MQTTClient::on_message(const struct mosquitto_message* message){#define MAX_PAYLOAD_SIZE (1024U)  char buf[MAX_PAYLOAD_SIZE] = {0};  if (!strcmp(message->topic, title_.c_str()))  {        memcpy(buf, message->payload, message->payloadlen);    auto message_str = std::string(buf, message->payloadlen);    if (onMessage_)    {      onMessage_(title_, message_str);    }  }}void MQTTClient::on_unsubscribe(int rc){}bool MQTTClient::connectToServer(){  username_pw_set("loadpoint", "gl@loadpoint.cn");  auto result = connect(host_.c_str(), port_, 60);  if (result == MOSQ_ERR_SUCCESS)    return true;  return false;}void MQTTClient::connectToServer(std::error_code& errorCode){  username_pw_set("loadpoint", "gl@loadpoint.cn");  auto result = connect(host_.c_str(), port_, 60);  errorCode = MakeErrorCode(result);}}  // namespace mqtt_client

QT的qmqtt,Qt官方在github上提供了源代码，地址：https://github.com/qt/qtmqtt

此外还有一个emqx维护的一个qt的客户端库qmqtt，也不错。地址：https://github.com/emqx/qmqtt

.net语言客户端库

MQTTnet是一个开源的MQTT库，用于在.NET平台上实现MQTT通信。它提供了一套简单易用的API，支持MQTT协议的连接、发布和订阅等功能。同时提供了 MQTT client and a MQTT server (broker)。

mqttnet地址：

https://github.com/dotnet/MQTTnet

以下是MQTTnet的一些特点和功能：

支持多种MQTT协议版本：MQTTnet支持MQTT 3.1.1和MQTT 5.0版本，可以根据需求选择合适的协议版本进行通信。
异步支持：MQTTnet使用异步操作，可以在高并发场景下提供更好的性能和响应能力。
完全可定制化：MQTTnet提供了丰富的配置选项，可以根据具体需求进行定制。可以设置连接参数、消息处理方式、保持活动时间等。
支持TLS/SSL加密：MQTTnet支持通过TLS/SSL对通信进行加密，保证数据的安全性。
支持QoS（服务质量）：MQTTnet支持消息的三个不同的服务质量等级：QoS 0（最多一次）、QoS 1（至少一次）和QoS 2（只有一次）。
支持遗嘱消息：MQTTnet支持设置遗嘱消息，当客户端异常断开连接时，服务器可以发送遗嘱消息给其他订阅者。
支持保留消息：MQTTnet支持发送和接收保留消息，保留消息会被服务器保留并发送给新的订阅者。
支持会话保持：MQTTnet支持会话保持功能，可以在客户端重新连接时恢复之前的会话状态。
跨平台支持：MQTTnet可以在多个.NET平台上使用，包括.NET Framework、.NET Core和Xamarin等。

MQTTnet是一个功能强大且易于使用的MQTT库，提供了服务端(Broker)和客户端，可以帮助开发人员在.NET平台上实现MQTT通信。它提供了丰富的功能和灵活的配置选项，适用于各种不同的MQTT场景和需求。

using System;using System.Text;using MQTTnet;using MQTTnet.Client;using MQTTnet.Client.Options;using MQTTnet.Extensions.ManagedClient; class Program{    static async Task Main(string[] args)    {        var factory = new MqttFactory();        var mqttClient = factory.CreateMqttClient();         var options = new MqttClientOptionsBuilder()            .WithTcpServer("broker.hivemq.com", 1883)            .Build();         mqttClient.UseConnectedHandler(async e =>        {            Console.WriteLine("Connected to MQTT broker");             await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("test/topic").Build());        });         mqttClient.UseDisconnectedHandler(async e =>        {            Console.WriteLine("Disconnected from MQTT broker");            await Task.Delay(TimeSpan.FromSeconds(5));            await mqttClient.ConnectAsync(options);        });         mqttClient.UseApplicationMessageReceivedHandler(e =>        {            Console.WriteLine($"Received message: {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");        });         await mqttClient.ConnectAsync(options);         while (true)        {            var message = new MqttApplicationMessageBuilder()                .WithTopic("test/topic")                .WithPayload("Hello, MQTT!")                .WithExactlyOnceQoS()                .WithRetainFlag()                .Build();             await mqttClient.PublishAsync(message);            await Task.Delay(TimeSpan.FromSeconds(1));        }    }}

java语言客户端库

GitHub - eclipse/paho.mqtt.java: Eclipse Paho Java MQTT client library. Paho is an Eclipse IoT project.

在Java中，常用的MQTT客户端库是 Eclipse Paho Java。下面是使用 Eclipse Paho Java 的简单示例代码：

import org.eclipse.paho.client.mqttv3.*;import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class MqttClientExample {    public static void main(String[] args) {        String broker = "tcp://broker.hivemq.com:1883";        String clientId = "JavaMqttClient";        MemoryPersistence persistence = new MemoryPersistence();         try {            MqttClient mqttClient = new MqttClient(broker, clientId, persistence);            MqttConnectOptions connectOptions = new MqttConnectOptions();            connectOptions.setCleanSession(true);             mqttClient.setCallback(new MqttCallback() {                public void connectionLost(Throwable cause) {                    System.out.println("Connection lost!");                }                 public void messageArrived(String topic, MqttMessage message) throws Exception {                    System.out.println("Received message: " + new String(message.getPayload()));                }                 public void deliveryComplete(IMqttDeliveryToken token) {                    System.out.println("Message delivered!");                }            });             mqttClient.connect(connectOptions);            mqttClient.subscribe("test/topic");            mqttClient.publish("test/topic", new MqttMessage("Hello, MQTT!".getBytes()));             // Keep the program running            while (true) {                Thread.sleep(1000);            }        } catch (MqttException e) {            e.printStackTrace();        } catch (InterruptedException e) {            e.printStackTrace();        }    }}

python语言客户端库

在Python下，常用的MQTT客户端库是Paho MQTT Python。

下面是安装和使用Paho MQTT Python的示例：

1. 安装Paho MQTT Python库：
使用pip命令安装Paho MQTT Python库：

pip install paho-mqtt

使用Paho MQTT Python库发送和接收MQTT消息的示例代码：

import paho.mqtt.client as mqtt    # MQTT连接成功回调函数   def on_connect(client, userdata, flags, rc):       print("Connected with result code "+str(rc))       # 订阅主题       client.subscribe("test/topic")    # MQTT接收消息回调函数   def on_message(client, userdata, msg):       print("Received message: "+str(msg.payload.decode()))    # 创建MQTT客户端   client = mqtt.Client()    # 设置连接成功和接收消息的回调函数   client.on_connect = on_connect   client.on_message = on_message    # 连接MQTT代理服务器   client.connect("broker.hivemq.com", 1883, 60)    # 循环处理网络流量，保持MQTT连接   client.loop_start()    # 发布消息   client.publish("test/topic", "Hello, MQTT!")    # 持续运行，直到手动停止   while True:       pass

以上示例代码创建了一个MQTT客户端，连接到HiveMQ公共MQTT代理服务器，并订阅了"test/topic"主题。当接收到消息时，会打印出来。然后发布一条消息到"test/topic"主题。最后，通过循环保持MQTT连接。

注意示例中使用的MQTT代理服务器是HiveMQ的公共服务器，实际使用时应根据需要设置为自己的MQTT代理服务器。

运行示例代码后，将能够发送和接收MQTT消息。

Android平台

在Android开发中，常用的MQTT客户端库有以下几个：

Eclipse Paho Android Service：这是Eclipse Paho项目的Android版本，提供了一个基于Service的MQTT客户端库。它提供了简单易用的API，支持MQTT 3.1和MQTT 3.1.1协议。通过将MQTT连接放在Service中，可以在后台持续运行MQTT连接，即使应用程序处于后台或设备休眠状态。

Eclipse Paho Android MQTT Client：这也是Eclipse Paho项目的Android版本，提供了一个基于Android库的MQTT客户端库。它提供了与Paho Java库相似的API，支持MQTT 3.1和MQTT 3.1.1协议。它可以在Android应用程序中使用，但不会在后台持续运行。

MQTT-Client-Framework：这是一个开源的Android MQTT客户端库，提供了基本的MQTT功能。它支持MQTT 3.1和MQTT 3.1.1协议，并提供了简单易用的API。它可以在Android应用程序中使用，但不会在后台持续运行。

HiveMQ MQTT Client：这是HiveMQ提供的Android MQTT客户端库，支持MQTT 3.1和MQTT 3.1.1协议。它提供了简单易用的API，可以在Android应用程序中使用。

这些MQTT客户端库都是开源的，可以根据具体需求选择合适的库进行Android应用程序的开发。它们提供了丰富的功能和易于使用的API，可以帮助你在Android平台上实现MQTT通信。