Mqtt

MQTT 中的 retained（保留消息）是一个重要概念，主要作用如下： 基本定义 retained 是 MQTT 消息的一个标志位，当发布者发送消息时可以设置 RETAIN 标志为 true，这样的消息会被 MQTT 代理（broker）保存。 主要特性 消息持久化：Broker 会为每个主题保存最后一条带有 retain 标志的消息 新订阅者立即获取：当新的客户端订阅某个主题时，会立即收到该主题的最后一条保留消息 自动更新：当有新的保留消息发布到同一主题时，旧的保留消息会被替换 清除机制：发布一条空内容（payload 为空）的保留消息可以清除该主题的保留消息 使用场景 设备状态报告：传感器发布温度、湿度等状态信息 配置信息：系统配置参数的发布 在线状态：设备的上下线状态通知 初始数据：确保新订阅者能立即获得最新数据 工作示例 # 客户端A发布保留消息 PUBLISH topic:"home/temperature" payload:"25.6" retain:true # 客户端B稍后订阅该主题 SUBSCRIBE topic:"home/temperature" # 客户端B会立即收到之前发布的保留消息 MESSAGE topic:"home/temperature" payload:"25.6" retain:true 保留消息机制使得 MQTT 非常适合物联网场景，新设备上线时能立即获取到最新的状态信息，而不需要等待下一次数据更新。

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。 MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。 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(); } } 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()