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

 1 #include <stdio.h>
 2 #include <mosquitto.h>
 3  void on_connect(struct mosquitto *mosq, void *userdata, int rc) {
 4     if(rc == 0) {
 5         printf("Connected to MQTT broker!\n");
 6         mosquitto_subscribe(mosq, NULL, "test/topic", 0);
 7     } else {
 8         printf("Connection failed: %s\n", mosquitto_strerror(rc));
 9     }
10 }
11  void on_message(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message) {
12     printf("Received message: %s\n", (char *)message->payload);
13 }
14  int main() {
15     struct mosquitto *mosq = NULL;
16     mosquitto_lib_init();
17      mosq = mosquitto_new(NULL, true, NULL);
18     if(mosq) {
19         mosquitto_connect_callback_set(mosq, on_connect);
20         mosquitto_message_callback_set(mosq, on_message);
21         mosquitto_connect(mosq, "localhost", 1883, 60);
22          mosquitto_loop_start(mosq);
23          mosquitto_publish(mosq, NULL, "test/topic", strlen("Hello, MQTT!"), "Hello, MQTT!", 0, false);
24          getchar();  // 按下Enter键退出程序
25          mosquitto_disconnect(mosq);
26         mosquitto_destroy(mosq);
27     }
28      mosquitto_lib_cleanup();
29     return 0;
30 }

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

./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

 1 #include <iostream>
 2 #include <cstring>
 3 #include <mosquittopp.h>
 4  class MyMqttClient : public mosqpp::mosquittopp {
 5 public:
 6     MyMqttClient(const char* id) : mosquittopp(id) {}
 7      void on_connect(int rc) {
 8         if (rc == 0) {
 9             std::cout << "Connected to MQTT broker" << std::endl;
10             subscribe(NULL, "test/topic");
11         } else {
12             std::cout << "Failed to connect to MQTT broker" << std::endl;
13         }
14     }
15      void on_message(const struct mosquitto_message* message) {
16         std::cout << "Received message: " << static_cast<char*>(message->payload) << std::endl;
17     }
18 };
19  int main() {
20     MyMqttClient client("client1");
21      client.connect("broker.hivemq.com");
22     client.loop_start();
23      while (true) {
24         client.publish(NULL, "test/topic", strlen("Hello, MQTT!"), "Hello, MQTT!");
25         std::this_thread::sleep_for(std::chrono::seconds(1));
26     }
27      client.loop_stop();
28     client.disconnect();
29      return 0;
30 }

 1 #include "mosquittopp.h"
 2  
 3 #include <functional>
 4 #include <memory>
 5 #include <string>
 6 #include <utility>
 7 #include <vector>
 8 namespace mqtt_client
 9 {
10 class MQTTClient final : public mosqpp::mosquittopp
11 {
12 public:
13   MQTTClient(const std::string& id, std::string host, int port, std::string title);
14   ~MQTTClient() override;
15  
16 public:
17   using onMessage = std::function<void(const std::string& title, const std::string& msg)>;
18   void setReceiveCallback(const onMessage& f) { onMessage_ = f; }
19  
20   bool connectToServer();
21   void connectToServer(std::error_code& errorCode);
22   const std::string& title() const { return title_; }
23  
24 private:
25   void on_connect(int rc) override;
26   void on_disconnect(int rc) override;
27   void on_error() override;
28   void on_subscribe(int mid, int qos_count, const int* granted_qos) override;
29   void on_message(const struct mosquitto_message* message) override;
30   void on_unsubscribe(int rc) override;
31  
32 private:
33   std::string title_;
34   std::string host_;
35   int port_;
36   onMessage onMessage_;
37 };
38  
39 class MQTTClientPtr
40 {
41 public:
42   std::vector<std::shared_ptr<MQTTClient>> ptr_;
43 };
44  
45 }  // namespace mqtt_client

 1 namespace mqtt_client
 2 {
 3 MQTTClient::MQTTClient(const std::string& id, std::string host, int port, std::string title)
 4     : /*mosqpp::mosquittopp(id.c_str()),*/ title_(std::move(title)),
 5       host_(std::move(host)),
 6       port_(port),
 7       onMessage_(nullptr)
 8 {
 9 }
10  
11 MQTTClient::~MQTTClient()
12 {
13   disconnect();
14   loop_stop();
15 }
16 void MQTTClient::on_connect(int rc)
17 {
18   if (rc == 0)
19   {
20     subscribe(NULL, title_.c_str());
21   }
22   else
23   {
24     LOGGING_ERROR("Connect - Code - %d", rc);
25   }
26 }
27 void MQTTClient::on_disconnect(int rc)
28 {
29   if (rc != 0)
30   {
31     LOGGING_ERROR("Disconnect - Code - %d", rc);
32   }
33 }
34 void MQTTClient::on_error()
35 {
36 }
37 void MQTTClient::on_subscribe(int mid, int qos_count, const int* granted_qos)
38 {
39   LOGGING_WARN("Subscription succeeded.");
40 }
41 void MQTTClient::on_message(const struct mosquitto_message* message)
42 {
43 #define MAX_PAYLOAD_SIZE (1024U)
44   char buf[MAX_PAYLOAD_SIZE] = {0};
45  
46   if (!strcmp(message->topic, title_.c_str()))
47   {
48     /* Copy N-1 bytes to ensure always 0 terminated. */
49     memcpy(buf, message->payload, message->payloadlen);
50  
51     auto message_str = std::string(buf, message->payloadlen);
52  
53     if (onMessage_)
54     {
55       onMessage_(title_, message_str);
56     }
57   }
58 }
59 void MQTTClient::on_unsubscribe(int rc)
60 {
61 }
62 bool MQTTClient::connectToServer()
63 {
64   username_pw_set("loadpoint", "gl@loadpoint.cn");
65   auto result = connect(host_.c_str(), port_, 60);
66   if (result == MOSQ_ERR_SUCCESS)
67     return true;
68   return false;
69 }
70 void MQTTClient::connectToServer(std::error_code& errorCode)
71 {
72   username_pw_set("loadpoint", "gl@loadpoint.cn");
73   auto result = connect(host_.c_str(), port_, 60);
74   errorCode = MakeErrorCode(result);
75 }
76  
77 }  // namespace mqtt_client

 1 using System;
 2 using System.Text;
 3 using MQTTnet;
 4 using MQTTnet.Client;
 5 using MQTTnet.Client.Options;
 6 using MQTTnet.Extensions.ManagedClient;
 7  class Program
 8 {
 9     static async Task Main(string[] args)
10     {
11         var factory = new MqttFactory();
12         var mqttClient = factory.CreateMqttClient();
13          var options = new MqttClientOptionsBuilder()
14             .WithTcpServer("broker.hivemq.com", 1883)
15             .Build();
16          mqttClient.UseConnectedHandler(async e =>
17         {
18             Console.WriteLine("Connected to MQTT broker");
19              await mqttClient.SubscribeAsync(new MqttTopicFilterBuilder().WithTopic("test/topic").Build());
20         });
21          mqttClient.UseDisconnectedHandler(async e =>
22         {
23             Console.WriteLine("Disconnected from MQTT broker");
24             await Task.Delay(TimeSpan.FromSeconds(5));
25             await mqttClient.ConnectAsync(options);
26         });
27          mqttClient.UseApplicationMessageReceivedHandler(e =>
28         {
29             Console.WriteLine($"Received message: {Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
30         });
31          await mqttClient.ConnectAsync(options);
32          while (true)
33         {
34             var message = new MqttApplicationMessageBuilder()
35                 .WithTopic("test/topic")
36                 .WithPayload("Hello, MQTT!")
37                 .WithExactlyOnceQoS()
38                 .WithRetainFlag()
39                 .Build();
40              await mqttClient.PublishAsync(message);
41             await Task.Delay(TimeSpan.FromSeconds(1));
42         }
43     }
44 }

 1 import org.eclipse.paho.client.mqttv3.*;
 2 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 3  public class MqttClientExample {
 4     public static void main(String[] args) {
 5         String broker = "tcp://broker.hivemq.com:1883";
 6         String clientId = "JavaMqttClient";
 7         MemoryPersistence persistence = new MemoryPersistence();
 8          try {
 9             MqttClient mqttClient = new MqttClient(broker, clientId, persistence);
10             MqttConnectOptions connectOptions = new MqttConnectOptions();
11             connectOptions.setCleanSession(true);
12              mqttClient.setCallback(new MqttCallback() {
13                 public void connectionLost(Throwable cause) {
14                     System.out.println("Connection lost!");
15                 }
16                  public void messageArrived(String topic, MqttMessage message) throws Exception {
17                     System.out.println("Received message: " + new String(message.getPayload()));
18                 }
19                  public void deliveryComplete(IMqttDeliveryToken token) {
20                     System.out.println("Message delivered!");
21                 }
22             });
23              mqttClient.connect(connectOptions);
24             mqttClient.subscribe("test/topic");
25             mqttClient.publish("test/topic", new MqttMessage("Hello, MQTT!".getBytes()));
26              // Keep the program running
27             while (true) {
28                 Thread.sleep(1000);
29             }
30         } catch (MqttException e) {
31             e.printStackTrace();
32         } catch (InterruptedException e) {
33             e.printStackTrace();
34         }
35     }
36 }

pip install paho-mqtt

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