MQTT是机器对机器(M2M)/物联网(IoT)连接协议。它被设计为一个极其轻量级的发布/订阅消息传输协议。对于需要较小代码占用空间和/或网络带宽非常宝贵的远程连接非常有用,是专为受限设备和低带宽、高延迟或不可靠的网络而设计。这些原则也使该协议成为新兴的“机器到机器”(M2M)或物联网(IoT)世界的连接设备,以及带宽和电池功率非常高的移动应用的理想选择。例如,它已被用于通过卫星链路与代理通信的传感器、与医疗服务提供者的拨号连接,以及一系列家庭自动化和小型设备场景。它也是移动应用的理想选择,因为它体积小,功耗低,数据包最小,并且可以有效地将信息分配给一个或多个接收器。
添加MQTT依赖库
3.1的库:https://pub.flutter-io.cn/packages/mqtt_client 5.0的库:https://pub.flutter-io.cn/packages/mqtt5_client
我们使用3.1的库
dependencies:
flutter:
sdk: flutter
flutter_localizations:
sdk: flutter
...
mqtt_client: ^9.6.3
使用MQTT前,最好先熟悉一下MQTT相关的概念,比如Topic、clean session、qos、retain。
配置
创建
/// 服务器地址,这里使用EMQ提供的免费mqtt测试服务器。
/// https://www.emqx.com/zh/mqtt/public-mqtt5-broker
final String brokerUrl = "broker-cn.emqx.io";
/// 设备id
final String clientId = "deviced_111";
/// 端口号
final int port = 1883;
/// 创建Mqtt实例
var _client = MqttServerClient.withPort(brokerUrl, clientId, port);
参数配置
/// 是否打印mqtt日志信息
_client.logging(on: true);
/// 设置端口号。创建时已经指定端口号就不需要设置。
_client.port = port;
/// 设置协议版本,默认是3.1,根据服务器需要的版本来设置
/// _client.setProtocolV31();
_client.setProtocolV311();
/// 保持连接ping-pong周期。默认不设置时关闭。
_client.keepAlivePeriod = 60;
/// 连接成功回调
_client.onConnected = onConnected;
/// 连接断开回调
_client.onDisconnected = onDisconnected;
/// 取消订阅回调
_client.onUnsubscribed = onUnsubscribed;
/// 订阅成功回调
_client.onSubscribed = onSubscribed;
/// 订阅失败回调
_client.onSubscribeFail = onSubscribeFail;
/// ping pong响应回调
_client.pongCallback = pong;
void onConnected() {
print("连接成功....");
}
void onDisconnected() {
print("连接断开");
}
void onUnsubscribed(String topic) {
print("取消订阅 $topic");
}
void onSubscribed(String topic) {
print("订阅 $topic 成功");
}
void onSubscribeFail(String topic) {
print("订阅主题: $topic 失败");
}
void pong() {
print("Ping的响应");
}
SSL证书配置
如果服务器需要使用ssl,我们需要提前下载好证书文件,放在assets目录中。
_configSSL() async{
/// 证书路径
var certPath = "assets/xxxx.cer";
/// 开启安全设置
_client.secure = true;
/// 创建SecurityContext
final mqttContext = SecurityContext.defaultContext;
/// 加载SSL证书
var byteData = await rootBundle.load(certPath);
/// 将证书的buffer数据添加到context中
mqttContext.setClientAuthoritiesBytes(byteData.buffer.asUint8List());
/// client指定context.
_client.securityContext = mqttContext;
}
连接设置
/// 创建连接配置
final connMessage = MqttConnectMessage()
/// 设置wiil遗嘱主题
.withWillTopic("willTopic")
/// 设置will遗嘱消息
.withWillMessage("willMessage")
/// 开启Clean Session.每次连接都是一个新的Session.
.startClean()
/// 设置will消息的qos模式。
.withWillQos(MqttQos.atLeastOnce);
_client.connectionMessage = connMessage;
连接
_connect() async {
try{
_client.connect();
}catch (e) {
print("连接异常: $e");
_client.disconnect();
}
}
/// 检查连接状态
if (_client.connectionStatus.state == MqttConnectionState.connected) {
print("已经连接....");
} else {
print("连接失败啦 状态===${_client.connectionStatus}");
_client.disconnect();
return;
}
/// 连接成功,可以继续操作
print("连接成功.....");
发送和接收消息
接收主题消息
订阅
想要接收消息,必须先订阅对应的主题
/// 需要订阅的主题
var scribeTopic = "topic/test";
订阅主题,并设置qos
_client.subscribe(scribeTopic, MqttQos.atLeastOnce);
监听消息
MqttClient.updates
_client.updates.listen((event) {
var recvMessage = event[0].payload as MqttPublishMessage;
/// 二进制格式的消息
bytes = recvMessage.payload.message;
/// 字符串格式的消息
message = Utf8Decoder().convert(recvMessage.payload.message);
print("原始数据-----:${recvMessage.payload.message}");
/// 转换成字符串
print(
"接收到了主题${event[0].topic}的消息: ${Utf8Decoder().convert(recvMessage.payload.message)}");
});
发送消息
使用MqttClientPayloadBuilder 构造消息内容,然后使用MqttServerClient.publishMessage() 发送消息。
发送明文
var builder = MqttClientPayloadBuilder();
builder.addUTF8String("This is a message");
_client.publishMessage(topic, MqttQos.atLeastOnce, builder.payload);
发送字节数组
var builder = MqttClientPayloadBuilder();
var data = [11,12,103,33];
builder.payload.addAll(data);
_client.publishMessage("消息的主题", MqttQos.atLeastOnce, builder.payload);
发送JSON数据
var builder = MqttClientPayloadBuilder();
var data = [11,12,103,33];
var jsonData = json.encode({"temperature": 23.5, "humidity": 32});
builder.addUTF8String(jsonData);
_client.publishMessage("消息的主题", MqttQos.atLeastOnce, builder.payload);
关闭
使用disconnect 关闭连接,一般在dispose中关闭。
@override
void dispose() {
super.dispose();
if (_client != null) {
_client.disconnect();
_client = null;
}
}
|