关于MQTT

发表于
字数统计 546

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的”轻量级”通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
using MQTTnet;
using MQTTnet.Client;

private async Task Handle_Received_Application_Message()
{
    var mqttFactory = new MqttFactory();
    using (var mqttClient = mqttFactory.CreateMqttClient())
    {
        var mqttClientOptions = new MqttClientOptionsBuilder()
            .WithTcpServer("192.168.106.76")
            .Build();

        mqttClient.ApplicationMessageReceivedAsync += e =>
        {
            string payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
            HandlePayload(payload);
            return Task.CompletedTask;
        };

        var mqttSubscribeOptions = mqttFactory.CreateSubscribeOptionsBuilder()
            .WithTopicFilter(f =>
            {
                f.WithTopic("fe-prod/sso2_burnin/v1/0c:b2:b7:ce:de:73/SSO2/all-data");
            })
            .WithTopicFilter(f =>
            {
                f.WithTopic("fe-prod/sso2_burnin/v1/fc:69:47:b8:3a:54/SSO2/all-data");
            })
            .WithTopicFilter(f => {
                f.WithTopic("fe-prod/sso2_burnin/v1/24:76:25:e9:9b:73/SSO2/all-data");
            })
            .Build();

        mqttClient.DisconnectedAsync += async e =>
        {
            AddLog("Disconnected...");
            await Task.Delay(1000);
            log.Error("连接已断开,尝试重新连接...", e.Exception);
            for (int i = 9; i > 0; i--)
            {
                AddLog($"Reconnect after {i}s...");
                await Task.Delay(1000);
            }
            await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
            await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
            //if (e.ClientWasConnected)
            //{
            //    // Use the current options as the new options.
            //    await mqttClient.ConnectAsync(mqttClient.Options);
            //}
            //else
            //{
            //    await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
            //    await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);
            //}
        };

        await mqttClient.ConnectAsync(mqttClientOptions, CancellationToken.None);
        await mqttClient.SubscribeAsync(mqttSubscribeOptions, CancellationToken.None);

        locks.WaitOne();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
 
import paho.mqtt.client as mqtt

# Successful Connection Callback
def on_connect(client, userdata, flags, rc):
    print('Connected with result code '+str(rc))
    client.subscribe('testtopic/#')

# Message delivery callback
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

client = mqtt.Client()

# Set callback handler
client.on_connect = on_connect
client.on_message = on_message

# Set up connection
client.connect('broker.emqx.io', 1883, 60)
# Publish message
client.publish('emqtt',payload='Hello World',qos=0)

client.loop_forever()