AI 助理
备案控制台
开发者社区 云计算 文章 正文

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

2022-12-27 410
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 本篇文章主要以ESP8266-12E作为开发板，带你了解MQTT发布、订阅、取消订阅的基础知识。

发布、订阅和取消订阅

理论知识

==PUBLISH – 发布消息==

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

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

    注意：

    1. 报文标识符packetId与QoS级别有密不可分的关系。只有QoS级别大于0时，报文标识符才是非零数值。如果QoS等于0，报文标识符为0。
    2. 重发标志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页面==

    image-20221224173621939
    至于如何建立这个profile请参考[这个教程]()

  • ==Subscribe页面==
    image-20221224174014661

  • 开发板发布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的串口监视器，找到自己发布的主题名称，并复制
      image-20221224231749009
    2. 打开MQTT.fx软件，进入Subscribe页面，输入自己的订阅名称，点击订阅

      image-20221224232455000

    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 ++;
}

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

  1. 打开ArduinoIDE的串口监视器，找到自己订阅的主题名称，并复制
    image-20221227195927750
  2. 打开MQTT.fx软件，进入publish页面，输入自己的订阅名称，点击发布
    注意：这一步粘贴过去之后，主题名后面会自动带一个空格，务必把它删掉（踩坑半小时）

    image-20221227200109069

  3. 如果输入1，开发板灯亮的话。那么恭喜你，实验成功！

    image-20221227200319069
    串口输出内容

文章标签：
云消息队列 MQ
网络性能优化
关键词：
云消息队列 MQ发布订阅
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
体温正常
目录
相关文章
薛伟同学
|
6月前
|
消息中间件
RabbitMQ消息模型之发布订阅Publish-Subscribe
RabbitMQ消息模型之发布订阅Publish-Subscribe
薛伟同学
86 0
RabbitMQ消息模型之发布订阅Publish-Subscribe
失散多年的哥哥
|
数据采集 传感器 存储
ESP32+MQTT+MySQL实现发布订阅【气味数据收集】
ESP32+MQTT+MySQL实现发布订阅【气味数据收集】
失散多年的哥哥
1455 0
ESP32+MQTT+MySQL实现发布订阅【气味数据收集】
云内容小助手
|
存储 网络协议 物联网
NB-IoT 通信之 MQTT 发布订阅 | 学习笔记
快速学习 NB-IoT 通信之 MQTT 发布订阅
云内容小助手
806 0
NB-IoT 通信之 MQTT 发布订阅 | 学习笔记
云内容小助手
|
网络协议 安全 物联网
MQTT- 基于 mosquitto 开源 SDK 实现发布订阅 | 学习笔记
快速学习 MQTT- 基于 mosquitto 开源 SDK 实现发布订阅
云内容小助手
638 0
MQTT- 基于 mosquitto 开源 SDK 实现发布订阅 | 学习笔记
mylxsw
|
消息中间件 Java PHP
RabbitMQ发布订阅实战-实现延时重试队列
本文将会讲解如何使用RabbitMQ实现延时重试和失败消息队列，实现可靠的消息消费，消费失败后，自动延时将消息重新投递，当达到一定的重试次数后，将消息投递到失败消息队列，等待人工介入处理。
mylxsw
2693 0
Esofar
|
消息中间件 C# 存储
[译]RabbitMQ教程C#版 - 发布订阅
先决条件 本教程假定RabbitMQ已经安装，并运行在localhost标准端口（5672）。如果你使用不同的主机、端口或证书，则需要调整连接设置。 从哪里获得帮助 如果您在阅读本教程时遇到困难，可以通过邮件列表 联系我们。
Esofar
1208 0
DreamSpark
|
28天前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
DreamSpark
70 5
mikechen的互联网架构
|
23天前
|
消息中间件 存储 Kafka
MQ 消息队列核心原理，12 条最全面总结！
本文总结了消息队列的12个核心原理，涵盖消息顺序性、ACK机制、持久化及高可用性等内容。关注【mikechen的互联网架构】，10年+BAT架构经验倾囊相授。
mikechen的互联网架构
56 3
DreamSpark
|
1月前
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
DreamSpark
71 7
开发者评测
|
27天前
|
消息中间件
解决方案 | 云消息队列RabbitMQ实践获奖名单公布！
云消息队列RabbitMQ实践获奖名单公布！
开发者评测
53 1

热门文章

最新文章

  • 1
    信用算力基于 RocketMQ 实现金融级数据服务的实践
  • 2
    MQTT协议
  • 3
    【解决方案】MQTT 强劲代理服务 EMQX 安装报Unable to load emulator DLL错误
  • 4
    MQTT的心跳保活机制是通过`setKeepAliveInterval()`方法设置的
  • 5
    消息中间件的技术选型心得－RabbitMQ、ActiveMQ和ZeroMQ
  • 6
    【Spring常见错误】Rocketmq 闪退
  • 7
    解决微信小程序MQTT真机连接问题与合法域名配置SSL问题
  • 8
    rabbitmq消息件在idc机房监控中的应用
  • 9
    RocketMQ 消息发送system busy、broker busy原因分析与解决方案
  • 10
    RabbitMQ入门指南(十一)：延迟消息-延迟消息插件
  • 1
    win10 安装RabbitMQ的步骤--和报错解决
    89
  • 2
    在 Windows 平台搭建 MQTT 服务
    116
  • 3
    mq报错abbit@syld36: * connected to epmd (port 4369) on syld36 * epmd reports node ‘rabbit‘ uses po
    76
  • 4
    Spring Boot与RabbitMQ的集成
    89
  • 5
    Docker部署RabbitMQ消息中间件
    277
  • 6
    JAVA面试之MQ
    70
  • 7
    EMQX 与 MQTT: AI 大模型时代的分布式数据中枢
    240
  • 8
    如何在Java中使用RabbitMQ
    249
  • 9
    【RocketMQ系列十四】RocketMQ中消息堆积如何处理
    1378
  • 10
    【RocketMQ系列十三】RocketMQ的集群核心概念之消费重试&死信队列&幂等消息的出现以及处理
    131

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 基于ROS的消息发布与订阅
  • 部署RabbitMQ
  • RocketMQ运维自我实践
  • RocketMQ的常规运维实践应用
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)