基于MQTT.fx的ESP8266主题发布订阅-阿里云开发者社区

发布、订阅和取消订阅

理论知识

==PUBLISH – 发布消息==

这里主要介绍客户端PUBLISH报文中的详细信息：

名称	内容	说明
packetId	4314	报文标识符。报文的ID编号
topicName	”weather“	报文所属主题
qos	1	服务质量等级。有`0 1 2`三个等级
retainFlag	false	保留标志。决定了当有新的客户端订阅该主题时，能否收到该主题中之前保留的内容
payload	"sunny"	有效载荷。即该客户端向服务器端发送的报文内容
dupFlag	false	重发标志。若接收方没接到报文，且`dupFlag==true`时，该报文将被重发

注意：

报文标识符packetId与QoS级别有密不可分的关系。只有QoS级别大于0时，报文标识符才是非零数值。如果QoS等于0，报文标识符为0。
重发标志dupFlag只在QoS级别大于0时使用

==SUBSCRIBE – 订阅主题==

客户端订阅主题时，一个报文中可以包含有单个或者多个订阅主题名
客户端在订阅主题时也可以明确QoS

==UNSUBSCRIBE – 取消订阅==

客户端取消订阅时，一个报文中可以包含有单个或者多个订阅主题名

==SUBACK – 订阅确认==

SUBACK报文是服务器端向客户端发送的报文

SUBACK报文包括两个信息

名称	内容	说明
packetId	4314	报文标识符。报文的ID编号报文标识符
returnCode	0 订阅成功 – QoS 0 1 订阅成功 – QoS 1 2 订阅成功 – QoS 2 128 订阅失败	订阅返回码。

程序

开发板发布MQTT主题

MQTT.fx软件页面介绍

这里以MQTT.fx1.7.1版本作为示例进行介绍
==Publish页面==

至于如何建立这个profile请参考[这个教程]()
==Subscribe页面==

开发板发布MQTT主题

/*
# 程序目的：实现ESP8266向 然也 物联网服务器发布订阅
# 创建时间：2022-12-25
# 函数：
## wifi对象：WiFi.SSID();WiFi.localIP(); WiFi.macAddress(); WiFi.mode(WIFI_STA);
## mqttClient对象：mqttClient.setServer(网站, 端口号); mqttClient.connected(); mqttClient.loop();                             mqttClient.connect(订阅者的ID.c_str()); mqttClient.state();mqttClient.publish(主题名数组,内容                    数组)
## Ticker对象：ticker.attach(即使周期(s),执行函数名);
# 程序思路：串口初始化->连接wifi->设置MQTT服务器和端口号->连接MQTT服务器(自建函数)->开启定时->loop函数中检查是否连接->             发布信息->保持客户端心跳
# 难点：变量类型间的转换
# 注意：需要把WiFi_Connect()函数中的WiFi名称和密码换成自己的。否则连接不上
*/    
#include <ESP8266WiFi.h>
#include <ESP8266WiFiMulti.h>
#include <PubSubClient.h>
#include <Ticker.h>

//创建对象
ESP8266WiFiMulti My_WifiMulti;
WiFiClient My_WiFiClient;
PubSubClient mqttClient(My_WiFiClient);
Ticker My_ticker;

//定义常量
const char* mqttServerSite = "test.ranye-iot.net";

//定义变量
int count = 0;  //计时

//函数申明
int WiFi_Connect();
void timeCounter();
void My_connectMQTTServer();
void pubMQTTmsg();

void setup() {
  //串口初始化
  Serial.begin(9600);

  //WiFi连接
  WiFi_Connect();

  //设置MQTT服务器网站地址和端口号
  mqttClient.setServer(mqttServerSite, 1883);

  //连接MQTT服务器
  My_connectMQTTServer();

  //开启定时
  // ticker.attach(即使周期（单位：s）,执行函数);
  My_ticker.attach(1,timeCounter); //注意：这里不是调用计数函数，只要写函数名即可

}

void loop()
{
  if(mqttClient.connected())//如果连接成功且间隔3秒：发布信息 
  {
    if(count>=3)
    {
      pubMQTTmsg();
      count = 0;
    }
    mqttClient.loop();
  }
  else //不成功重新尝试连接
  {
    My_connectMQTTServer();
  }
}

/*
wifi连接函数
需引 ESP8266WiFiMulti.h 库  并建立ESP8266WiFiMulti对象
*/
int WiFi_Connect() {
  My_WifiMulti.addAP("TPLINK2.4G", "@@@@@@@@");  // Wifi1
  My_WifiMulti.addAP("username2", "password");   // Wifi2
  My_WifiMulti.addAP("username3", "password");   // Wifi3

  int i = 0;
  Serial.print("\n-------------Connected Time:-------------\n");
  while (My_WifiMulti.run() != WL_CONNECTED) {
    i += 1;
    Serial.print(i);
    Serial.println("->");
    delay(1000);
    if (i > 15) {
      Serial.print("\n-------------WIFI connected failed!-------------\n");
      return 0;
    }
  }

  Serial.println("\n-------------WIFI connected successful!-------------\n");
  Serial.println("\n-------------WIFI Name:-------------");
  Serial.println(WiFi.SSID());
  Serial.println("\n-------------ESP8266 IP address:-------------");
  Serial.println(WiFi.localIP());
  Serial.println("\n-------------------------------------------");
  return 1;
}

/*
客户端连接服务器端函数
注意：需要提前建立mqttClient对象
*/
void My_connectMQTTServer()
{
  // 根据ESP8266的MAC地址生成客户端ID（避免与其它ESP8266的客户端ID重名）
  String clientID="ESP8266-" + WiFi.macAddress();//获取ESP8266的MAC地址
  //连接MQTT服务器
  if(mqttClient.connect(clientID.c_str()))//注意：这里的String.c_str是对字符串的处理
  {
    Serial.println("MQTT Server Connected.");
    Serial.print("Server Address: "); Serial.println(mqttServerSite);
    Serial.print("ClientId:"); Serial.println(clientID);
    Serial.print("Client State:"); Serial.println(mqttClient.state());
  }
  else
  {
    Serial.print("MQTT Server Connect Failed. Client State:");
    Serial.println(mqttClient.state());
    delay(3000);
  }
}

//发布信息
void pubMQTTmsg()
{
/* —----------------------主题名称编辑操作------------------------ */
  //发布主题的名称 字符串型
  String topicString = "PUBLIC-topicName-" + WiFi.macAddress();

  //转换成char数组类型（mqttClient.public传参需要）
  //创建数组
  char topicArr[topicString.length() + 1];
  //将信息拷贝到数组内
  strcpy(topicArr,topicString.c_str()); // strcpy(目标变量,拷贝源);

/* —----------------------发布内容编辑操作------------------------ */
  static int value = 0; //一定要设成静态的
  String messageString = "PUBLIC-messageame-" + String(value++);

  //转换成char数组类型（mqttClient.public传参需要）
  //创建数组
  char messageArr[messageString.length() + 1];
  //将信息拷贝到数组内
  strcpy(messageArr,messageString.c_str()); // strcpy(目标变量,拷贝源);

/* —----------------------向主题发布信息------------------------ */
  if(mqttClient.publish(topicArr,messageArr)) //mqttClient.publish(主题名数组,内容数组)
  {
    Serial.print("Topic:");Serial.println(topicArr);
    Serial.print("Message:");Serial.println(messageArr);
  }
  else
  {
    Serial.println("Public Failed!");
  }
}

//计时函数
void timeCounter(){
  count++;
}

若程序正确，且成功上传开发板之后。可以通过如下操作验证已发布订阅消息：
1. 打开ArduinoIDE的串口监视器，找到自己发布的主题名称，并复制
2. 打开MQTT.fx软件，进入Subscribe页面，输入自己的订阅名称，点击订阅
3. 如果这里出现不断滚动的消息的话。那么恭喜你，实验成功！

开发板订阅主题

程序

/*
# 程序目的：实现ESP8266订阅 然也 物联网服务器的主题。服务端发送1，灯亮；发送0，灯灭
# 创建时间：2022-12-26
# 函数：
## wifi对象： WiFi.SSID();WiFi.localIP(); WiFi.macAddress(); WiFi.mode(WIFI_STA);
## mqttClient对象：mqttClient.setServer(网站, 端口号); mqttClient.connected(); mqttClient.loop(); mqttClient.connect(订阅者的ID.c_str()); mqttClient.state();mqttClient.subscribe(主题名数组); mqttClient.setCallback(回调函数名称)
## Ticker对象：ticker.attach(即使周期(s),执行函数名);
# 程序思路：串口初始化->配置引脚高低电平 -> 连接wifi->设置MQTT服务器和端口号->重写接收回调函数 -> 连接MQTT服务器(自建函数)->开启定时->loop函数中检查是否连接->订阅信息->保持客户端心跳
# 难点：变量类型间的转换、回调函数的书写
# 注意：需要把WiFi_Connect()函数中的WiFi名称和密码换成自己的。否则连接不上
*/    
#include <ESP8266WiFi.h>
#include <ESP8266WiFiMulti.h>
#include <PubSubClient.h>
#include <Ticker.h>

//创建对象
ESP8266WiFiMulti My_WifiMulti;
WiFiClient My_WifiClient;
PubSubClient mqttClient(My_WifiClient);
Ticker My_ticker;

//定义常量
const char* mqttServerSite = "test.ranye-iot.net";

//定义变量
unsigned int worktime_s;

//函数申明
int WiFi_Connect();
void My_connectMQTTServer();
void My_receiveCallback(char* topic, uint8_t* payload, unsigned int length);//注意：回调函数的参数需参考PubSubClient.头文件
void My_subscribeTopic();
void calcTheWorkTime();

void setup()
{
  //串口
  Serial.begin(9600);

  //引脚配置
  pinMode(LED_BUILTIN,OUTPUT); // 设置板上LED引脚为输出模式
  digitalWrite(LED_BUILTIN, HIGH); // 启动后关闭板上LED

  //连接WiFi
  WiFi_Connect();

  //记录工作时间
  My_ticker.attach(1,calcTheWorkTime);

  //建立服务
  mqttClient.setServer(mqttServerSite,1883);

  //重写回调函数(注意：回调函数的重写要在连接之前)
  mqttClient.setCallback(My_receiveCallback);

  //连接mqtt服务器
  My_connectMQTTServer();

  //订阅指定主题
  My_subscribeTopic();

}

void loop()
{
  if (mqttClient.connected()) // 连接成功
  {
    mqttClient.loop(); //保持客户端心跳
  }
  else //失败——重新尝试连接
  {
    My_connectMQTTServer();
  }
}

/*
wifi连接函数
需引 ESP8266WiFiMulti.h 库  并建立ESP8266WiFiMulti对象
*/
int WiFi_Connect() {
  My_WifiMulti.addAP("TPLINK2.4G", "@@@@@@@@");  // Wifi1
  My_WifiMulti.addAP("username2", "password");   // Wifi2
  My_WifiMulti.addAP("username3", "password");   // Wifi3

  int i = 0;
  Serial.print("\n-------------Connected Time:-------------\n");
  while (My_WifiMulti.run() != WL_CONNECTED) {
    i += 1;
    Serial.print(i);
    Serial.print("->");
    delay(1000);
    if (i > 15) {
      Serial.print("\n-------------WIFI connected failed!-------------\n");
      return 0;
    }
  }

  Serial.println("\n-------------WIFI connected successful!-------------\n");
  Serial.println("\n-------------WIFI Name:-------------");
  Serial.println(WiFi.SSID());
  Serial.println("\n-------------ESP8266 IP address:-------------");
  Serial.println(WiFi.localIP());
  Serial.println("\n-------------------------------------------");
  return 1;
}


/*
客户端连接服务器端函数
注意：需要提前建立mqttClient对象
*/
void My_connectMQTTServer()
{
  // 根据ESP8266的MAC地址生成客户端ID（避免与其它ESP8266的客户端ID重名）
  String clientId = "ESP8266_clientId_" + String(WiFi.macAddress()); //注意：这里的String.c_str是对字符串的处理
  //连接MQTT服务器
  if (mqttClient.connect(clientId.c_str()))
  {
    Serial.print("Connect to "); Serial.print(mqttServerSite); Serial.println(" successful!");
    Serial.print("clientId = ");Serial.println(clientId);
    Serial.print("Server status: ");Serial.println(mqttClient.state());
  }
  else
  {
    Serial.print("Connect to "); Serial.print(mqttServerSite); Serial.println(" failed!");
    Serial.print("Server status: ");Serial.println(mqttClient.state());
    delay(3000);
  }
}


/*
重写收到订阅消息的回调函数（难点）
注意：参数需参考PubSubClient.头文件  而且即使有多个主题，也只能有一个回调函数
*/
void My_receiveCallback(char* topic, uint8_t* payload, unsigned int length) //((char*)主题,()有效载荷,(uint)长度)
{
  static int number = 0;
  Serial.println(" ");
  Serial.print("This is the ["); Serial.print(number++); Serial.println("] message"); 
  Serial.print("WorkTime: ["); Serial.print(worktime_s); Serial.println("s]");

  /* —----------------------打印主题名------------------------ */
  Serial.print("Message is from the topic of: [");
  Serial.print(topic);
  Serial.println("] ");

/* —----------------------打印主题长度------------------------ */
  Serial.print("Message Length: ["); Serial.print(length); Serial.println("(Bytes) ] ");

/* —----------------------内容及逻辑控制------------------------ */
  Serial.print("Payload: [");
  for (int i = 0; i < length; i++ )
  {
    Serial.print((char)payload[i]); //强制类型转换语法：(类型)表达式
  }
  Serial.println("] ");

  if( (char)payload[0] == '1')
  {
    digitalWrite(LED_BUILTIN,LOW);
  }
  else
  {
    digitalWrite(LED_BUILTIN,HIGH);
  }
}

/*
主题订阅函数
注意：与主题发布函数差不多
*/
void My_subscribeTopic()
{
/* —----------------------第一个主题变量加工------------------------ */
  //订阅主题的名称 字符串型
  String topicString_1 = "WuKaiXiong's topic_1";

  //转换成char数组类型（mqttClient.public传参需要）
  //创建数组
  char topicArr_1[topicString_1.length() + 1];
  //将信息拷贝到数组内
  strcpy(topicArr_1,topicString_1.c_str());// strcpy(目标变量,拷贝源);

/* —----------------------第二个主题变量加工------------------------ */
  //订阅主题的名称 字符串型
  String topicString_2 = "WuKaiXiong_topic_2";

  //转换成char数组类型（mqttClient.public传参需要）
  //创建数组
  char topicArr_2[topicString_2.length() + 1];
  //将信息拷贝到数组内
  strcpy(topicArr_2,topicString_2.c_str());// strcpy(目标变量,拷贝源);

/* —----------------------向服务器发布订阅请求------------------------ */
  if(mqttClient.subscribe(topicArr_1) && mqttClient.subscribe(topicArr_2))
  {
    Serial.print("topic: "); Serial.print(topicArr_1); Serial.print(" and "); Serial.print(topicArr_2); Serial.println(" subscribe success!");
  }
  else 
  {
    Serial.print("topic: "); Serial.print(topicArr_1); Serial.print(" and "); Serial.print(topicArr_2); Serial.println("failed!...");
  }
}

void calcTheWorkTime()
{
  worktime_s ++;
}

若程序正确，且成功上传开发板之后。可以通过如下操作验证已发布订阅消息：

打开ArduinoIDE的串口监视器，找到自己订阅的主题名称，并复制
打开MQTT.fx软件，进入publish页面，输入自己的订阅名称，点击发布
注意：这一步粘贴过去之后，主题名后面会自动带一个空格，务必把它删掉（踩坑半小时）
如果输入1，开发板灯亮的话。那么恭喜你，实验成功！

串口输出内容

基于MQTT.fx的ESP8266主题发布订阅

发布、订阅和取消订阅

理论知识

程序

开发板发布MQTT主题

开发板订阅主题

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

基于MQTT.fx的ESP8266主题发布订阅

发布、订阅和取消订阅

理论知识

程序

开发板发布MQTT主题

开发板订阅主题

热门文章

最新文章

相关课程

相关电子书

相关实验场景