关于如何在Windows下设置mqtt服务器请移步上一篇博文：python MQTT服务通信

环境准备：

1. mosqutto服务端程序，需要进行一些配置，重启服务，默认服务端口为1883
2. mqttx客户端程序，方便订阅和发布信息：https://github.com/emqx/MQTTX/releases
3. Arduino通过包管理器安装PubSubClient
4. esp32连接的网络和运行mosqutto服务程序的电脑处在同一个网段

arduino代码

/*********
  @author: Wenqing Zhou (zhou.wenqing@qq.com)
  @github: https://github.com/ouening
  
  功能
  ESP32搭建一个MQTT客户端，订阅主题"esp32/output"，以此控制led灯；发布主题
  "esp32/dht11/temperature"和"esp32/dht11/humidity"将DHT11获取的温湿度数据推给
  MQTT服务器，如果其他客户端在相同的MQTT服务器下订阅了该主题，便可获取对应的温度或者湿度
  数据。
  
  参考链接：Complete project details at https://randomnerdtutorials.com  
*********/
#include <WiFi.h>
#include <PubSubClient.h>
#include <Wire.h>
#include <WiFiClient.h>
#include <Adafruit_Sensor.h>
#include <DHT.h>
#include <DHT_U.h>

// Replace the next variables with your SSID/Password combination
const char* ssid = "wifi名称";
const char* password = "wifi密码";

// Add your MQTT Broker IP address, example:
const char* mqtt_server = "192.168.28.87"; //先测试本机mqtt服务，该地址在windows下通过ipconfig查看，要和esp32连接在同一个网络
//const char* mqtt_server = "YOUR_MQTT_BROKER_IP_ADDRESS";
const char *id = "ESP32";
const char *user = "kindy";
const char *pass = "kindy";

WiFiClient espClient;
PubSubClient client(espClient); // MQTT服务设置了非账号密码不能使用，所有在connect的时候要设置账号密码
long lastMsg = 0;
char msg[50];
int value = 0;

/* 设置DHT11 */
#define DHTPIN 4     // Digital pin connected to the DHT sensor 
#define DHTTYPE    DHT11     // DHT 11
DHT_Unified dht(DHTPIN, DHTTYPE);

float temperature = 0;
float humidity = 0;
// LED Pin
const int ledPin = 2;

void setup() {
  Serial.begin(115200);
  pinMode(ledPin, OUTPUT);
  // default settings
  setup_wifi();
  client.setServer(mqtt_server, 1883);
  client.setCallback(callback); // 绑定回调函数
}

void setup_wifi() {
  delay(10);
  // We start by connecting to a WiFi network
  Serial.println();
  Serial.print("Connecting to ");
  Serial.println(ssid);
  WiFi.mode(WIFI_STA); // station mode
  WiFi.begin(ssid, password);

  while (WiFi.status() != WL_CONNECTED) {
    delay(1000);
    Serial.print(".");
  }

  Serial.println("");
  Serial.println("WiFi connected");
  Serial.println("IP address: ");
  Serial.println(WiFi.localIP());

    /* ====== 初始化DHT11 ========*/
  dht.begin();
  Serial.println(F("DHTxx Unified Sensor Example"));
  // Print temperature sensor details.
  sensor_t sensor;
  dht.temperature().getSensor(&sensor);
  Serial.println(F("------------------------------------"));
  Serial.println(F("Temperature Sensor"));
  Serial.print  (F("Sensor Type: ")); Serial.println(sensor.name);
  Serial.print  (F("Driver Ver:  ")); Serial.println(sensor.version);
  Serial.print  (F("Unique ID:   ")); Serial.println(sensor.sensor_id);
  Serial.print  (F("Max Value:   ")); Serial.print(sensor.max_value); Serial.println(F("°C"));
  Serial.print  (F("Min Value:   ")); Serial.print(sensor.min_value); Serial.println(F("°C"));
  Serial.print  (F("Resolution:  ")); Serial.print(sensor.resolution); Serial.println(F("°C"));
  Serial.println(F("------------------------------------"));
  // Print humidity sensor details.
  dht.humidity().getSensor(&sensor);
  Serial.println(F("Humidity Sensor"));
  Serial.print  (F("Sensor Type: ")); Serial.println(sensor.name);
  Serial.print  (F("Driver Ver:  ")); Serial.println(sensor.version);
  Serial.print  (F("Unique ID:   ")); Serial.println(sensor.sensor_id);
  Serial.print  (F("Max Value:   ")); Serial.print(sensor.max_value); Serial.println(F("%"));
  Serial.print  (F("Min Value:   ")); Serial.print(sensor.min_value); Serial.println(F("%"));
  Serial.print  (F("Resolution:  ")); Serial.print(sensor.resolution); Serial.println(F("%"));
  Serial.println(F("------------------------------------"));
}

void callback(char* topic, byte* message, unsigned int length) {
  Serial.print("Message arrived on topic: ");
  Serial.print(topic);
  Serial.print(". Message: ");
  String messageTemp;
  
  for (int i = 0; i < length; i++) {
    Serial.print((char)message[i]);
    messageTemp += (char)message[i];
  }
  Serial.println();

  // Feel free to add more if statements to control more GPIOs with MQTT

  // If a message is received on the topic esp32/output, you check if the message is either "on" or "off". 
  // Changes the output state according to the message
  if (String(topic) == "esp32/output") {
    Serial.print("Changing output to ");
    if(messageTemp == "on"){
      Serial.println("on");
      digitalWrite(ledPin, HIGH);
    }
    else if(messageTemp == "off"){
      Serial.println("off");
      digitalWrite(ledPin, LOW);
    }
  }
}

/* 重连mqtt服务器 */
void reconnect() {
  // Loop until we're reconnected
  while (!client.connected()) {
    Serial.print("Attempting MQTT connection...");
    //MQTT服务设置了非账号密码不能使用，所有在connect的时候要设置账号密码
    if (client.connect(id, user, pass)) { 
      Serial.println("connected");
      // 订阅mqtt主题
      client.subscribe("esp32/output");
    } else {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
      // Wait 5 seconds before retrying
      delay(5000);
    }
  }
}
void loop() {
  if (!client.connected()) {
    reconnect();
  }
  client.loop();

  long now = millis();
  if (now - lastMsg > 5000) {
    lastMsg = now;
    char tempString[8];
    // Convert the value to a char array
    char humString[8];

    // 获取温度数据
    sensors_event_t event;
    dht.temperature().getEvent(&event);
    if (isnan(event.temperature)) {
      Serial.println(F("Error reading temperature!"));
    }
    else {
      dtostrf((float)event.temperature,2,2,tempString); // convert to String
      Serial.print("Temperature: ");
      Serial.println(tempString);
      client.publish("esp32/dht11/temperature", tempString); // 发布信息，第一个参数是主题
    }
    // Get humidity event and print its value.
    dht.humidity().getEvent(&event);
    if (isnan(event.relative_humidity)) {
      Serial.println(F("Error reading humidity!"));
    }
    else {
      dtostrf((float)event.relative_humidity,2,2,humString);
      Serial.print("Humidity: ");
      Serial.println(humString);
      client.publish("esp32/dht11/humidity", humString);
    }
  }
}

MQTTX订阅主题：

MQTTX发布信息：