Async MQTT Client ESP32: Nicht blockierendes MQTT

Mit einem Async MQTT Client auf dem ESP32 kann Ihre Firmware weiter Sensoren auslesen, Displays ansteuern und Webseiten ausliefern, während der MQTT-Verkehr im Hintergrund abgewickelt wird. In dieser Anleitung bauen Sie einen vollständigen AsyncMqttClient-Sketch mit Event-Callbacks, automatischen Reconnects und QoS-1-Publishing gegen einen echten Broker. Außerdem lernen Sie die Fallstricke der Bibliothek kennen, damit Sie entscheiden können, wann sie die richtige Wahl ist.

Warum ein Async MQTT Client

Blockierende Bibliotheken wie PubSubClient funktionieren gut, setzen aber voraus, dass Sie loop() häufig aufrufen, und connect() kann Ihren Sketch bei schlechtem Netzwerk bis zum Socket-Timeout einfrieren. Auf einem Gerät, das zusätzlich einen Motor antreibt, einen LED-Streifen aktualisiert oder eine Konfigurationsseite bereitstellt, sind diese Pausen sichtbar.

AsyncMqttClient von Marvin Roger verfolgt einen anderen Ansatz. Die Bibliothek setzt auf AsyncTCP auf, das die Netzwerkkommunikation in einem eigenen FreeRTOS-Task ausführt. Ihr Code registriert Callbacks für Ereignisse (verbunden, getrennt, Nachricht empfangen, Publish bestätigt), und die Bibliothek ruft sie auf, sobald etwas passiert. Es gibt kein loop(), das Sie aufrufen müssten.

Aspekt	PubSubClient	AsyncMqttClient
Programmiermodell	Polling, `loop()` aufrufen	Event-Callbacks
Verhalten von `connect()`	Blockiert bis CONNACK oder Timeout	Kehrt sofort zurück
Publish-QoS	0	0, 1 und 2
Subscribe-QoS	0 und 1	0, 1 und 2
Große eingehende Payloads	Müssen in den Puffer passen	Werden in Teilstücken geliefert
TLS auf dem ESP32	Ja, über `WiFiClientSecure`	Mit Standard-AsyncTCP nicht praktikabel
Abhängigkeiten	Keine	AsyncTCP

Die letzte Zeile ist entscheidend. Die TLS-Unterstützung von AsyncMqttClient hängt von einem SSL-fähigen Build von AsyncTCP ab, der in der Standard-AsyncTCP-Bibliothek für den ESP32 nicht verfügbar ist. In der Praxis eignet sich diese Bibliothek daher am besten für reines MQTT auf Port 1883 innerhalb eines vertrauenswürdigen Netzwerks oder für Prototypen. Wenn Sie TLS mit einer ereignisgesteuerten API brauchen, bietet die aktiv gepflegte Bibliothek espMqttClient einen sehr ähnlichen Callback-Stil mit TLS-Unterstützung auf dem ESP32. Die Abwägungen behandeln wir am Ende.

Die Bibliotheken installieren

Installieren Sie in der Arduino IDE das Board-Paket "esp32 by Espressif Systems".
Installieren Sie AsyncTCP für den ESP32. Im Bibliotheksverwalter gibt es mehrere gepflegte Forks; wählen Sie einen, der zu Ihrer ESP32-Core-Version passt.
Installieren Sie AsyncMqttClient. Erscheint die Bibliothek nicht im Bibliotheksverwalter, laden Sie das Release-ZIP aus dem Projekt-Repository herunter und nutzen Sie Sketch → Bibliothek einbinden → .ZIP-Bibliothek hinzufügen.
Mit PlatformIO fügen Sie beide in platformio.ini unter lib_deps hinzu.

[env:esp32dev]
platform = espressif32
board = esp32dev
framework = arduino
monitor_speed = 115200
lib_deps =
    marvinroger/AsyncMqttClient
    esp32async/AsyncTCP

Der vollständige Async-MQTT-Client-Sketch für den ESP32

Dieses Beispiel verbindet sich auf Port 1883 mit broker.connectmq.com, dem öffentlichen MQTT-Broker von ConnectMQ, der für Tests und Prototypen gedacht ist. Es stellt WLAN- und MQTT-Verbindung mit FreeRTOS-Timern wieder her, veröffentlicht alle zehn Sekunden Telemetrie mit QoS 1 und schaltet die Onboard-LED, wenn ein Befehl eintrifft.

#include <WiFi.h>
#include <AsyncMqttClient.h>

extern "C" {
  #include "freertos/FreeRTOS.h"
  #include "freertos/timers.h"
}

#define WIFI_SSID     "YOUR_WIFI"
#define WIFI_PASSWORD "YOUR_WIFI_PASSWORD"

#define MQTT_HOST "broker.connectmq.com"
#define MQTT_PORT 1883

#define LED_PIN 2

static const char* DEVICE_ID = "async-4b81d2";   // eindeutig machen
String topicTelemetry = String("connectmq/demo/") + DEVICE_ID + "/telemetry";
String topicCommand   = String("connectmq/demo/") + DEVICE_ID + "/led/set";
String topicStatus    = String("connectmq/demo/") + DEVICE_ID + "/status";

AsyncMqttClient mqttClient;
TimerHandle_t mqttReconnectTimer;
TimerHandle_t wifiReconnectTimer;

unsigned long lastPublish = 0;

void connectToWifi() {
  Serial.println("Connecting to Wi-Fi...");
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
}

void connectToMqtt() {
  Serial.println("Connecting to MQTT...");
  mqttClient.connect();
}

void onWifiEvent(WiFiEvent_t event) {
  switch (event) {
    case ARDUINO_EVENT_WIFI_STA_GOT_IP:
      Serial.print("Wi-Fi connected, IP ");
      Serial.println(WiFi.localIP());
      connectToMqtt();
      break;
    case ARDUINO_EVENT_WIFI_STA_DISCONNECTED:
      Serial.println("Wi-Fi lost");
      xTimerStop(mqttReconnectTimer, 0);   // kein MQTT, solange das WLAN weg ist
      xTimerStart(wifiReconnectTimer, 0);
      break;
    default:
      break;
  }
}

void onMqttConnect(bool sessionPresent) {
  Serial.printf("MQTT connected, session present: %d\n", sessionPresent);
  mqttClient.subscribe(topicCommand.c_str(), 1);
  mqttClient.publish(topicStatus.c_str(), 1, true, "online");
}

void onMqttDisconnect(AsyncMqttClientDisconnectReason reason) {
  Serial.printf("MQTT disconnected, reason %d\n", static_cast<int>(reason));
  if (WiFi.isConnected()) {
    xTimerStart(mqttReconnectTimer, 0);
  }
}

void onMqttSubscribe(uint16_t packetId, uint8_t qos) {
  Serial.printf("Subscribe ack, packetId %u, granted QoS %u\n", packetId, qos);
}

void onMqttPublish(uint16_t packetId) {
  Serial.printf("Publish ack (PUBACK), packetId %u\n", packetId);
}

void onMqttMessage(char* topic, char* payload,
                   AsyncMqttClientMessageProperties properties,
                   size_t len, size_t index, size_t total) {
  // payload ist nicht nullterminiert und kann ein Teilstück einer größeren Nachricht sein
  if (index != 0 || len != total) {
    Serial.println("Fragmented message ignored in this demo");
    return;
  }
  String msg;
  msg.reserve(len);
  for (size_t i = 0; i < len; i++) msg += payload[i];

  Serial.printf("Message on %s (QoS %u): %s\n", topic, properties.qos, msg.c_str());

  if (msg == "ON")  digitalWrite(LED_PIN, HIGH);
  if (msg == "OFF") digitalWrite(LED_PIN, LOW);
}

void setup() {
  Serial.begin(115200);
  pinMode(LED_PIN, OUTPUT);

  mqttReconnectTimer = xTimerCreate("mqttTimer", pdMS_TO_TICKS(2000), pdFALSE,
                                    nullptr, reinterpret_cast<TimerCallbackFunction_t>(connectToMqtt));
  wifiReconnectTimer = xTimerCreate("wifiTimer", pdMS_TO_TICKS(2000), pdFALSE,
                                    nullptr, reinterpret_cast<TimerCallbackFunction_t>(connectToWifi));

  WiFi.onEvent(onWifiEvent);

  mqttClient.onConnect(onMqttConnect);
  mqttClient.onDisconnect(onMqttDisconnect);
  mqttClient.onSubscribe(onMqttSubscribe);
  mqttClient.onPublish(onMqttPublish);
  mqttClient.onMessage(onMqttMessage);

  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
  mqttClient.setClientId(DEVICE_ID);              // Zeiger muss gültig bleiben
  mqttClient.setKeepAlive(30);
  mqttClient.setWill(topicStatus.c_str(), 1, true, "offline");

  connectToWifi();
}

void loop() {
  // Ihr Anwendungscode läuft hier ungehindert; MQTT wird im Hintergrund abgewickelt.
  if (mqttClient.connected() && millis() - lastPublish > 10000) {
    lastPublish = millis();
    char payload[96];
    snprintf(payload, sizeof(payload), "{\"uptime\":%lu,\"rssi\":%d,\"heap\":%u}",
             millis() / 1000, WiFi.RSSI(), (unsigned)ESP.getFreeHeap());
    uint16_t id = mqttClient.publish(topicTelemetry.c_str(), 1, false, payload);
    Serial.printf("Queued publish, packetId %u\n", id);
  }
}

Beobachten Sie den Verkehr und senden Sie Befehle von Ihrem Computer aus:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-4b81d2 -t "connectmq/demo/async-4b81d2/#" -v -q 1
mosquitto_pub -h broker.connectmq.com -p 1883 -i cmd-4b81d2 -t "connectmq/demo/async-4b81d2/led/set" -m ON -q 1

So funktioniert der Ereignisablauf

Wer die Reihenfolge der Ereignisse versteht, vermeidet die meisten Bugs:

connectToWifi() startet die WLAN-Verbindung und kehrt sofort zurück.
Sobald die Station eine IP-Adresse erhält, ruft onWifiEvent die Funktion connectToMqtt() auf.
mqttClient.connect() kehrt sofort zurück. Später empfängt der AsyncTCP-Task das CONNACK, und die Bibliothek ruft onMqttConnect auf.
In onMqttConnect abonnieren Sie Topics und veröffentlichen Ihren Status "online". Das ist der richtige Ort, denn Subscriptions einer Clean Session müssen nach jedem Reconnect erneuert werden.
Trennt Sie der Broker oder das Netzwerk, plant onMqttDisconnect einen einmaligen Timer, der zwei Sekunden später connectToMqtt() aufruft.

publish() liefert bei QoS-1- und QoS-2-Nachrichten eine Paket-ID zurück, 1 bei einer eingereihten QoS-0-Nachricht und 0, wenn die Nachricht nicht gesendet werden konnte. Das passende onMqttPublish(packetId) wird ausgelöst, sobald der Broker die Zustellung bestätigt. So können Sie nachverfolgen, ob wichtige Nachrichten tatsächlich angekommen sind.

Fallstricke im Produktivbetrieb

Callbacks laufen in einem anderen Task

Ihre Callbacks werden im AsyncTCP-Task ausgeführt, nicht in loop(). Rufen Sie dort weder delay() auf noch führen Sie langsame I/O-Operationen aus; das blockiert den gesamten Netzwerkverkehr und kann den Task-Watchdog auslösen. Kopieren Sie die Daten in eine Variable oder Queue und verarbeiten Sie sie in loop(). Teilen sich loop() und ein Callback eine Variable, schützen Sie sie, etwa mit einer FreeRTOS-Queue oder in einfachen Fällen mit einem volatile-Flag.

Lebensdauer von Strings

setClientId(), setCredentials(), setServer() mit Hostnamen und setWill() speichern Zeiger, keine Kopien. Wer c_str() eines temporären String übergibt, sendet beim Reconnect Datenmüll. Verwenden Sie globale Konstanten oder globale String-Objekte, wie im Beispiel.

Fragmentierte Payloads

Große eingehende Nachrichten kommen in mehreren onMessage-Aufrufen mit steigendem index und gleichem total an. Erwarten Sie Payloads von mehr als ein paar hundert Bytes, sammeln Sie die Teilstücke in einem Puffer der Größe total und verarbeiten Sie sie, wenn index + len == total gilt.

Kein eingebauter Reconnect

Die Bibliothek verbindet sich nicht von selbst neu. Das obige Timer-Muster ist der Standardansatz. Stoppen Sie den MQTT-Timer immer, wenn das WLAN wegfällt, sonst überfluten Sie den Broker mit Verbindungsversuchen, die nicht gelingen können.

Persistente Sessions für verpasste Befehle

Standardmäßig verbindet sich der Client mit einer Clean Session, sodass Befehle, die gesendet wurden, während das Gerät offline war, verloren gehen. Rufen Sie vor dem Verbinden mqttClient.setCleanSession(false) auf und abonnieren Sie mit QoS 1: Der Broker behält dann die Session und reiht QoS-1-Nachrichten für das Gerät ein, bis es zurückkehrt. Die Session lebt nicht ewig; Broker lassen sie nach einer Weile ablaufen (auf dem öffentlichen MQTT-Broker nach höchstens 30 Minuten). Prüfen Sie sessionPresent in onMqttConnect, um zu sehen, ob sich der Broker an Sie erinnert hat, und abonnieren Sie erneut, wenn nicht.

Broker-Limits gelten trotzdem

Asynchron heißt nicht unbegrenzt. Der öffentliche MQTT-Broker erlaubt 10 Nachrichten pro Sekunde und Client, 10 Subscriptions pro Client und 64 KB pro Nachricht. Eine enge Schleife, die publish() aufruft, stößt schnell an diese Grenzen.

Async oder blockierender Client: die Wahl

Diese Checkliste hilft bei der Entscheidung:

TLS auf Port 8883 nötig? Wählen Sie PubSubClient mit WiFiClientSecure oder den Secure Client von espMqttClient. Das Standard-AsyncMqttClient passt hier nicht gut.
Publishing mit QoS 1 oder 2 vom Gerät nötig? AsyncMqttClient oder espMqttClient.
Läuft auf demselben Gerät ein Webserver (ESPAsyncWebServer)? AsyncMqttClient teilt sich den AsyncTCP-Stack problemlos.
Soll auch der ESP8266 unterstützt werden? AsyncMqttClient funktioniert dort mit ESPAsyncTCP, allerdings ist der Speicher knapper. Siehe ESP8266-MQTT-Tutorial.
Möglichst einfacher Code für ein Einsteigerprojekt? Das PubSubClient-Tutorial ist der sanftere Einstieg.

Geht es Ihnen um Zustellgarantien, lesen Sie MQTT-QoS-Stufen erklärt, bevor Sie nur wegen QoS 2 die Bibliothek wechseln; viele Sensor-Workloads sind mit QoS 0 oder 1 am besten bedient. Das umfassendere ESP32-IoT-Tutorial zeigt, wie sich MQTT in ein komplettes Sensorprojekt einfügt.

Produktiv-Setup mit ConnectMQ

Für Einsätze außerhalb Ihres Labors sollte das Gerät statt mit dem öffentlichen MQTT-Broker mit einem eigenen Cloud-MQTT-Broker arbeiten. Bei ConnectMQ legen Sie im Portal einen Broker an und erhalten Benutzername, Passwort und Client ID. Mit AsyncMqttClient setzen Sie diese wie folgt; beachten Sie, dass Client IDs an die ausgestellten Zugangsdaten gebunden sind:

mqttClient.setServer("broker.connectmq.com", 1883);
mqttClient.setClientId("YOUR_CLIENT_ID");
mqttClient.setCredentials("YOUR_USERNAME", "YOUR_PASSWORD");

Da Zugangsdaten auf Port 1883 unverschlüsselt übertragen werden, sollten Sie für Geräte außerhalb eines vertrauenswürdigen Netzwerks eine Bibliothek mit TLS-Unterstützung bevorzugen. Sitzt der ESP32 als Sensorknoten in einer Fertigungshalle, ersetzen Sie zudem das Demo-Präfix durch die ISA-95-Hierarchie Ihres Unified Namespace (UNS), etwa acme/werk-kassel/montage/linie-2/zelle-1/telemetry.

Starten Sie mit ConnectMQ

ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS, WebSockets, einem Live-Topic-Explorer und Schwellenwert-Alarmen im Portal, sodass Sie Ihrem ESP32 beim Publizieren in Echtzeit zusehen können. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless Broker ohne Kreditkarte und verbinden Sie Ihr erstes asynchrones Gerät in wenigen Minuten.

Häufig gestellte Fragen

Funktioniert AsyncMqttClient auf dem ESP32 mit TLS?

Nicht mit der Standard-Bibliothek AsyncTCP, für die es keinen SSL-Build für den ESP32 gibt. Verwenden Sie PubSubClient mit WiFiClientSecure oder eine gepflegte asynchrone Bibliothek wie espMqttClient, wenn Sie Port 8883 brauchen.

Warum brauche ich FreeRTOS-Timer für den Reconnect?

AsyncMqttClient hat keinen automatischen Reconnect. Einmalige Timer erlauben einen erneuten Versuch nach kurzer Verzögerung, ohne loop() oder den Netzwerk-Task zu blockieren.

Darf ich delay() in onMqttMessage aufrufen?

Nein. Callbacks laufen im AsyncTCP-Task, und Blockieren dort legt den gesamten Netzwerkverkehr lahm. Speichern Sie die Nachricht und verarbeiten Sie sie in loop().

Wird AsyncMqttClient noch gepflegt?

Die ursprüngliche Bibliothek wird seit einiger Zeit nicht mehr aktiv weiterentwickelt, ist aber nach wie vor weit verbreitet. Prüfen Sie den Status des Repositorys, bevor Sie ein neues, langlebiges Produkt beginnen, und ziehen Sie espMqttClient als gepflegte Alternative in Betracht.

Async MQTT Client auf dem ESP32: Nicht blockierendes MQTT mit AsyncMqttClient