4、MQTT客户端实现：PubSubClient库安装、连接MQTT Broker、订阅主题、发布消息、心跳保活机制

好，咱们接着往下走。上一节我们把MQTT Broker搭起来了，相当于有了一个“消息中转站”。那这一节，我们就来搞定客户端——也就是ESP32这边，怎么连上去、怎么说话、怎么听话。

说白了，就是让ESP32变成一个合格的MQTT“会员”。

4.1 PubSubClient库：MQTT客户端的“身份证”

ESP32要跟MQTT Broker通信，不能自己瞎写协议栈吧？那太累了。Arduino生态里有个现成的库，叫PubSubClient。我个人习惯用这个，轻量、稳定，社区活跃度高。

安装方式很简单：

  1. 打开Arduino IDE，点「工具」→「管理库」
  2. 搜索框输入 PubSubClient
  3. 找到作者是 Nick O'Leary 的那个，点安装
小提示： 我建议安装最新稳定版。早期版本（比如2.6之前）对MQTT 5.0支持不太好，但咱们用3.1.1协议，其实无所谓。我项目里一直用2.8.0，稳得很。

4.2 连接MQTT Broker：第一次“握手”

库装好了，接下来就是写代码连Broker。先看一个最基础的连接示例：

#include <WiFi.h>
#include <PubSubClient.h>

// WiFi配置
const char* ssid = "你的WiFi名";
const char* password = "你的WiFi密码";

// MQTT Broker配置
const char* mqtt_server = "192.168.1.100"; // 换成你的Broker IP
const int mqtt_port = 1883;

WiFiClient espClient;
PubSubClient client(espClient);

void setup() {
  Serial.begin(115200);
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
}

void setup_wifi() {
  delay(10);
  Serial.println();
  Serial.print("连接WiFi中...");
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println(" 已连接！");
}

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("尝试连接MQTT...");
    if (client.connect("ESP32_Client_001")) {
      Serial.println(" 成功！");
    } else {
      Serial.print(" 失败，错误码：");
      Serial.print(client.state());
      delay(5000);
    }
  }
}

嗯，这里要注意几个点：

  • 客户端ID必须唯一。我见过有人直接用 "ESP32" 当ID，结果两个设备同时在线，后一个把前一个踢下线了。我建议加个随机数或MAC地址后缀。
  • 连接失败要重试。Broker可能还没启动，或者网络波动。我一般设5秒重试一次，最多重试10次。
  • client.loop() 必须放在主循环里。它负责处理收发包和心跳。忘了它，连接会断。
避坑指南： 我曾经在loop()里写了个delay(10000)，结果client.loop()十秒才执行一次，Broker以为我掉线了，直接踢掉。记住：loop()里别用长延时，用millis()做非阻塞定时。

4.3 订阅主题：学会“听”消息

连上Broker之后，ESP32得知道自己要听哪些“频道”。这就是订阅主题。看代码：

void reconnect() {
  while (!client.connected()) {
    if (client.connect("ESP32_Client_001")) {
      // 连接成功后订阅主题
      client.subscribe("sensor/temperature");
      client.subscribe("actuator/led");
    }
  }
}

// 回调函数：收到消息时自动触发
void callback(char* topic, byte* payload, unsigned int length) {
  Serial.print("收到主题：");
  Serial.println(topic);
  
  // 把payload转成字符串
  String message = "";
  for (int i = 0; i < length; i++) {
    message += (char)payload[i];
  }
  Serial.print("消息内容：");
  Serial.println(message);
  
  // 根据主题做不同处理
  if (String(topic) == "actuator/led") {
    if (message == "ON") {
      digitalWrite(LED_BUILTIN, HIGH);
    } else if (message == "OFF") {
      digitalWrite(LED_BUILTIN, LOW);
    }
  }
}

这里有个关键点：回调函数。你想想看，ESP32不可能一直轮询“有没有新消息”，那样太浪费CPU了。PubSubClient的做法是：收到消息后，自动调用你注册的callback函数。你只需要在里面写处理逻辑就行。

我在项目里遇到过一个问题：回调函数里做了太多事情，比如写SD卡、发HTTP请求，导致处理时间过长，影响了后续消息的接收。后来我把耗时操作放到任务队列里异步处理，才解决。

4.4 发布消息：学会“说”话

订阅是“听”，发布就是“说”。ESP32采集到传感器数据后，要发出去。代码很简单：

void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();
  
  // 每5秒发布一次温度数据
  static unsigned long lastMsg = 0;
  if (millis() - lastMsg > 5000) {
    lastMsg = millis();
    
    float temperature = 25.6; // 假设从传感器读到的值
    char msg[50];
    snprintf(msg, 50, "{\"temp\":%.1f}", temperature);
    
    // 发布到主题
    client.publish("sensor/temperature", msg);
    Serial.print("发布消息：");
    Serial.println(msg);
  }
}

发布时，我建议用JSON格式。为什么？因为可扩展性好。今天你只发温度，明天想加湿度，直接加个字段就行，不用改主题结构。我习惯用 ArduinoJson 库来构建JSON，比手动拼字符串安全多了。

重要提醒： publish() 的第三个参数是 retained（是否保留消息）。如果设为true，新订阅者会立即收到最后一条保留消息。这个功能很有用，比如设备重启后能立刻知道当前状态。但别滥用，否则Broker内存会爆。

4.5 心跳保活机制：别让Broker把你忘了

MQTT协议里有个“心跳”机制，学名叫Keep Alive。说白了，就是ESP32每隔一段时间跟Broker说一声“我还活着”。如果Broker在指定时间内没收到心跳，就认为客户端掉线了，会清理资源。

PubSubClient默认的心跳间隔是15秒。你可以通过 setKeepAlive() 修改：

client.setKeepAlive(30); // 设为30秒

那心跳是怎么实现的呢？其实不用你手动发。PubSubClient的 client.loop() 会自动处理。它会在空闲时发送 PINGREQ 包，Broker回复 PINGRESP。整个过程对开发者透明。

但有个坑：如果网络不稳定，PINGRESP可能收不到。PubSubClient默认重试3次，每次间隔1秒。如果3次都失败，就认为连接断了，会触发重连。我曾经在信号差的厂房里调试，发现设备频繁重连，后来把重试次数改成5次、间隔改成2秒，问题就解决了。

调优建议： 心跳间隔别设太长，也别太短。太长的话，Broker发现掉线有延迟；太短的话，网络开销大。我一般设30-60秒，具体看应用场景。如果设备经常休眠，可以设长一点，比如300秒。

4.6 完整示例：一个能跑起来的MQTT客户端

最后，我把上面的代码整合成一个完整的示例。你直接复制到Arduino IDE里，改一下WiFi和Broker信息就能用：

#include <WiFi.h>
#include <PubSubClient.h>

const char* ssid = "你的WiFi";
const char* password = "你的密码";
const char* mqtt_server = "192.168.1.100";
const int mqtt_port = 1883;
const char* clientID = "ESP32_" + String(ESP.getEfuseMac(), HEX);

WiFiClient espClient;
PubSubClient client(espClient);

void setup() {
  Serial.begin(115200);
  pinMode(LED_BUILTIN, OUTPUT);
  setup_wifi();
  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(callback);
  client.setKeepAlive(30);
}

void setup_wifi() {
  delay(10);
  WiFi.begin(ssid, password);
  while (WiFi.status() != WL_CONNECTED) {
    delay(500);
    Serial.print(".");
  }
  Serial.println(" WiFi已连接");
}

void callback(char* topic, byte* payload, unsigned int length) {
  String message = "";
  for (int i = 0; i < length; i++) message += (char)payload[i];
  Serial.printf("主题: %s, 消息: %s\n", topic, message.c_str());
  
  if (String(topic) == "actuator/led") {
    digitalWrite(LED_BUILTIN, message == "ON" ? HIGH : LOW);
  }
}

void reconnect() {
  while (!client.connected()) {
    Serial.print("连接MQTT...");
    if (client.connect(clientID)) {
      Serial.println(" 成功");
      client.subscribe("actuator/#");
    } else {
      Serial.printf(" 失败, 错误码: %d\n", client.state());
      delay(5000);
    }
  }
}

void loop() {
  if (!client.connected()) reconnect();
  client.loop();
  
  static unsigned long lastMsg = 0;
  if (millis() - lastMsg > 5000) {
    lastMsg = millis();
    float temp = 25.0 + random(-10, 10) / 10.0;
    char msg[50];
    snprintf(msg, 50, "{\"temp\":%.1f}", temp);
    client.publish("sensor/temperature", msg);
    Serial.printf("发布: %s\n", msg);
  }
}

这个示例里，我用了 actuator/# 通配符订阅，这样所有以 actuator/ 开头的主题都能收到。客户端ID用了MAC地址，保证唯一性。心跳设了30秒，发布周期5秒。

嗯，到这里，MQTT客户端的基本功能就全了。下一节我们会讲怎么用这些基础功能，搭一个完整的远程监控系统。到时候你会看到，这些代码组合起来，能玩出很多花样。

← 上一章 📖 返回目录 下一章 →