ESP32 FreeRTOS MQTT: Sensor- und Publish-Tasks

Ein ESP32-FreeRTOS-MQTT-Design legt Sensorabfrage und Netzwerkveröffentlichung in getrennte Tasks, sodass ein langsamer WLAN-Reconnect nie eine Messung verzögert. In diesem Tutorial bauen Sie einen Sketch mit einem Sensor-Task, einem MQTT-Task und einer Queue dazwischen, testen ihn mit dem kostenlosen öffentlichen MQTT-Broker broker.connectmq.com und stellen ihn anschließend mit eigenen Zugangsdaten auf TLS um.

Warum FreeRTOS-Tasks für MQTT auf dem ESP32

Die meisten Einsteiger-Sketches erledigen alles in loop(): Sensor lesen, WLAN prüfen, client.loop() aufrufen, veröffentlichen, dann delay(). Das funktioniert, bis das Netzwerk Probleme macht. Ein connect()-Aufruf zu einem nicht erreichbaren Broker kann mehrere Sekunden blockieren, und in dieser Zeit wird nichts gemessen. Wenn Sie Impulse zählen, ein Analogsignal mitteln oder einen Schwellwert überwachen, sind diese Lücken relevant – etwa beim Erfassen von Stückzahlen oder Maschinenzuständen für OEE-Kennzahlen, wo jede verpasste Messung die Auswertung verfälscht.

Der ESP32-Arduino-Core läuft bereits auf FreeRTOS. loop() ist selbst nur ein auf Core 1 gepinnter Task. Nichts hindert Sie daran, weitere Tasks anzulegen – und sobald Sie das tun, hängt das Timing der einen Aufgabe nicht mehr von der anderen ab:

Der Sensor-Task wacht mit vTaskDelayUntil() in einem festen Takt auf, liest die Hardware und legt eine kleine Struct in eine Queue. Das Netzwerk berührt er nie.
Der MQTT-Task besitzt WLAN und MQTT-Client. Er leert die Queue, veröffentlicht, bedient Keep-Alives und verbindet sich bei Bedarf neu.
Die Queue ist das einzige gemeinsam genutzte Objekt. FreeRTOS-Queues kopieren Daten und sind zwischen Tasks sicher nutzbar, Sie brauchen für die Messwerte also keinen Mutex.

Eine Regel macht dieses Design robust: Nur ein einziger Task ruft jemals den MQTT-Client auf. PubSubClient ist nicht threadsicher, und gleichzeitige publish()-Aufrufe aus zwei Tasks beschädigen seinen internen Puffer. Leiten Sie stattdessen alles über die Queue.

Was Sie brauchen

Ein beliebiges ESP32-Board mit zwei Kernen (ESP32 DevKit, ESP32-S3). Single-Core-Chips wie der ESP32-C3 oder ESP32-C6 funktionieren ebenfalls, wenn Sie beide Tasks auf Core 0 pinnen.
Arduino IDE 2.x oder PlatformIO mit dem esp32-Core 3.x.
Die Bibliothek PubSubClient von Nick O'Leary.
Optional: einen beliebigen Sensor. Der Sketch simuliert Temperatur und Luftfeuchtigkeit, damit er auch auf einem nackten Board läuft.
mosquitto_sub auf Ihrem Rechner zum Testen.

Task-Aufteilung und Prioritäten

Task	Core	Priorität	Stack	Aufgabe
`sensorTask`	1	3	3072 Bytes	Jede Sekunde messen, in die Queue schreiben
`mqttTask`	0	2	6144 Bytes	WLAN, MQTT-Verbindung, Publish, Keep-Alive
Arduino `loop()`	1	1	Standard	Gelöscht, nicht benötigt

Der Sensor-Task erhält die höhere Priorität, weil er kurz und zeitkritisch ist: Er läuft, legt einen Eintrag in die Queue und schläft. Der MQTT-Task läuft auf Core 0 neben dem WLAN-Stack, der ebenfalls dort liegt, und erledigt die langsame Arbeit. Stackgrößen werden im ESP32-Arduino-Core in Bytes angegeben, nicht wie im Standard-FreeRTOS in Words.

Der vollständige ESP32-FreeRTOS-MQTT-Sketch

Ersetzen Sie die WLAN-Zugangsdaten und den zufälligen Teil des Topic-Präfixes und laden Sie den Sketch hoch.

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

// ---------- configuration ----------
const char* WIFI_SSID     = "YOUR_WIFI_SSID";
const char* WIFI_PASSWORD = "YOUR_WIFI_PASSWORD";
const char* MQTT_HOST     = "broker.connectmq.com";
const uint16_t MQTT_PORT  = 1883;
const char* TOPIC_PREFIX  = "connectmq/demo/k3v9q1";   // eindeutig machen

// ---------- shared objects ----------
struct Reading {
  uint32_t seq;
  uint32_t uptimeMs;
  float    temperature;
  float    humidity;
};

QueueHandle_t readingQueue;
WiFiClient    net;
PubSubClient  mqtt(net);
String        clientId;

// ---------- sensor task ----------
void sensorTask(void* param) {
  const TickType_t period = pdMS_TO_TICKS(1000);
  TickType_t lastWake = xTaskGetTickCount();
  uint32_t seq = 0;

  for (;;) {
    Reading r;
    r.seq         = seq++;
    r.uptimeMs    = millis();
    // Durch echtes Sensorauslesen ersetzen (DHT22, BME280, ADC...)
    r.temperature = 21.0f + (esp_random() % 400) / 100.0f;
    r.humidity    = 45.0f + (esp_random() % 1000) / 100.0f;

    // Nicht blockieren: Ist die Queue voll, ältesten Eintrag verwerfen und erneut versuchen
    if (xQueueSend(readingQueue, &r, 0) != pdTRUE) {
      Reading discarded;
      xQueueReceive(readingQueue, &discarded, 0);
      xQueueSend(readingQueue, &r, 0);
    }
    vTaskDelayUntil(&lastWake, period);
  }
}

// ---------- MQTT helpers (only called from mqttTask) ----------
bool ensureWifi() {
  if (WiFi.status() == WL_CONNECTED) return true;
  WiFi.mode(WIFI_STA);
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
  for (int i = 0; i < 40 && WiFi.status() != WL_CONNECTED; i++) {
    vTaskDelay(pdMS_TO_TICKS(250));
  }
  return WiFi.status() == WL_CONNECTED;
}

bool ensureMqtt() {
  if (mqtt.connected()) return true;
  String willTopic = String(TOPIC_PREFIX) + "/status";
  bool ok = mqtt.connect(clientId.c_str(), willTopic.c_str(), 1, true, "offline");
  if (ok) {
    mqtt.publish(willTopic.c_str(), "online", true);
  }
  return ok;
}

// ---------- MQTT task ----------
void mqttTask(void* param) {
  mqtt.setServer(MQTT_HOST, MQTT_PORT);
  mqtt.setBufferSize(512);
  String topic = String(TOPIC_PREFIX) + "/telemetry";
  uint32_t backoffMs = 1000;

  for (;;) {
    if (!ensureWifi() || !ensureMqtt()) {
      vTaskDelay(pdMS_TO_TICKS(backoffMs));
      backoffMs = min<uint32_t>(backoffMs * 2, 30000);
      continue;
    }
    backoffMs = 1000;
    mqtt.loop();

    Reading r;
    // Kurz auf Daten warten, damit der Task im Leerlauf die CPU abgibt
    if (xQueueReceive(readingQueue, &r, pdMS_TO_TICKS(50)) == pdTRUE) {
      char payload[160];
      snprintf(payload, sizeof(payload),
               "{\"seq\":%lu,\"uptimeMs\":%lu,\"temperature\":%.2f,\"humidity\":%.2f,\"queued\":%u}",
               (unsigned long)r.seq, (unsigned long)r.uptimeMs,
               r.temperature, r.humidity,
               (unsigned)uxQueueMessagesWaiting(readingQueue));
      if (!mqtt.publish(topic.c_str(), payload)) {
        // Wieder vorne einreihen und nach dem Reconnect erneut versuchen
        xQueueSendToFront(readingQueue, &r, 0);
      }
    }
  }
}

void setup() {
  Serial.begin(115200);
  clientId = "esp32-rtos-" + String((uint32_t)ESP.getEfuseMac(), HEX);

  readingQueue = xQueueCreate(60, sizeof(Reading));   // eine Minute Puffer
  if (readingQueue == nullptr) {
    Serial.println("Queue allocation failed");
    while (true) delay(1000);
  }

  xTaskCreatePinnedToCore(sensorTask, "sensor", 3072, nullptr, 3, nullptr, 1);
  xTaskCreatePinnedToCore(mqttTask,   "mqtt",   6144, nullptr, 2, nullptr, 0);
}

void loop() {
  vTaskDelete(nullptr);   // der Arduino-Loop-Task wird nicht benötigt
}

Verfolgen Sie die Ausgabe von Ihrem Rechner aus:

mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-k3v9q1 -t "connectmq/demo/k3v9q1/#" -v

Sie sollten eine status-Nachricht online und eine Telemetrie-Nachricht pro Sekunde sehen. Ziehen Sie nun für zwanzig Sekunden den Stecker Ihres Routers. Das Feld queued wächst, solange das Board offline ist, und wenn die Verbindung zurückkehrt, arbeitet der MQTT-Task den Rückstau der Reihe nach ab. Genau darum geht es bei diesem Design: Die Messungen erfolgten planmäßig, obwohl das Veröffentlichen stockte.

Der öffentliche MQTT-Broker ist für jeden lesbar, halten Sie Demodaten also allgemein. Außerdem begrenzt er jeden Client auf 10 Nachrichten pro Sekunde, was beim Abarbeiten eines Rückstaus erreicht werden kann. Wenn Sie viele Messwerte puffern, fügen Sie nach jedem Publish ein kurzes vTaskDelay(pdMS_TO_TICKS(120)) ein, solange uxQueueMessagesWaiting() hoch ist. Die Anleitung zum kostenlosen öffentlichen MQTT-Broker listet die weiteren Grenzen auf.

So funktionieren die einzelnen Teile

Die Queue als Puffer

xQueueCreate(60, sizeof(Reading)) reserviert 60 Plätze fester Größe auf dem Heap. Jedes xQueueSend kopiert die 16-Byte-Struct, sodass der Sensor-Task seine lokale Variable sofort wiederverwenden kann. Die Größe der Queue ist ein Kompromiss zwischen RAM und der Dauer eines Ausfalls, den Sie überbrücken möchten. Sechzig Plätze zu 16 Bytes sind weniger als 1 KB; selbst einige Tausend Plätze passen problemlos auf einen klassischen ESP32, doch denken Sie daran, dass die Daten bei einem Neustart verloren gehen.

Ist die Queue voll, verwirft der Sketch den ältesten statt des neuesten Messwerts. Bei Telemetrie ist der aktuellste Wert in der Regel der nützlichste. Wenn jede Messung zählt, blockieren Sie stattdessen für einige Ticks oder aggregieren Sie die Messwerte (Minimum, Maximum, Mittelwert), bevor Sie sie einreihen.

vTaskDelayUntil statt delay

vTaskDelayUntil() weckt den Task in einem festen Takt, gemessen ab dem vorherigen Aufwachzeitpunkt, sodass die Periode nicht um die Dauer des Sensorauslesens driftet. Ein einfaches delay(1000) nach einem 30-ms-Lesevorgang ergibt eine Periode von 1,03 s, die langsam wegläuft. Beide Funktionen geben die CPU ab – anders als eine Busy-Loop mit millis().

Reconnect mit Backoff

Der MQTT-Task versucht es mit exponentiellem Backoff erneut, begrenzt auf 30 Sekunden. Den Broker mit sofortigen Wiederholungen zu bombardieren, verschwendet Energie und kann bei einem verwalteten MQTT-Broker den Flood-Schutz oder eine IP-Sperre wegen wiederholter Fehlversuche auslösen. Die Anleitung zu ESP32 MQTT Last Will erklärt, warum die Retained-Will-Nachricht offline beim Verbinden registriert wird, damit Dashboards ein ausgefallenes Gerät bemerken.

Stolperfallen und Fehlerbehebung

Guru Meditation „Stack canary watchpoint triggered“. Der Task-Stack ist zu klein. snprintf mit Floats und TLS brauchen beide Reserven. Geben Sie während der Entwicklung in jedem Task uxTaskGetStackHighWaterMark(nullptr) aus; die Funktion liefert die minimale jemals beobachtete Anzahl freier Bytes. Lassen Sie mindestens 512 Bytes Reserve.
Resets durch den Task Watchdog. Ein Task, der in einer Schleife läuft, ohne jemals zu blockieren, lässt den Idle-Task auf seinem Core verhungern. Jeder Schleifendurchlauf muss etwas aufrufen, das die CPU abgibt: vTaskDelay, xQueueReceive mit Timeout oder einen blockierenden Socket-Aufruf.
Verstümmelte serielle Ausgabe. Zwei Tasks, die gleichzeitig ausgeben, vermischen ihre Zeichen. Geben Sie entweder nur aus einem Task aus oder schützen Sie Serial mit einem Mutex, den Sie mit xSemaphoreCreateMutex() erzeugen.
Veröffentlichen aus dem Sensor-Task. Es scheint zu funktionieren und scheitert dann unter Last zufällig. Halten Sie den MQTT-Client in einem einzigen Task.
Callbacks laufen im MQTT-Task. Wenn Sie Befehle abonnieren, läuft der PubSubClient-Callback innerhalb von mqtt.loop(). Erledigen Sie dort keine langsame Arbeit, sondern reichen Sie den Befehl über eine zweite Queue an einen anderen Task weiter.
Abgeschnittenes großes JSON. Der Standardpuffer von PubSubClient umfasst 256 Bytes einschließlich Topic. Der Sketch erhöht ihn mit setBufferSize(512).

Umstieg auf TLS und einen eigenen Broker

Der öffentliche MQTT-Broker ist ideal zum Testen, doch Produktionsgeräte sollten sich authentifizieren und verschlüsseln. Erstellen Sie ein kostenloses Konto unter app.connectmq.com, legen Sie einen Broker an und kopieren Sie den generierten Benutzernamen, das Passwort und die Client ID. Ändern Sie dann nur den Netzwerkteil des MQTT-Tasks:

#include <WiFiClientSecure.h>

// ISRG Root X1, die Root von Let's Encrypt. Das von Let's Encrypt veröffentlichte PEM einfügen.
static const char ROOT_CA[] = R"PEM(
-----BEGIN CERTIFICATE-----
...paste ISRG Root X1 here...
-----END CERTIFICATE-----
)PEM";

WiFiClientSecure net;
PubSubClient mqtt(net);

// in mqttTask, vor der Schleife:
net.setCACert(ROOT_CA);
mqtt.setServer("broker.connectmq.com", 8883);

// in ensureMqtt():
mqtt.connect("YOUR_CLIENT_ID", "YOUR_USERNAME", "YOUR_PASSWORD",
             willTopic.c_str(), 1, true, "offline");

Erhöhen Sie den Stack des MQTT-Tasks auf etwa 8192 Bytes, denn der TLS-Handshake benötigt spürbar mehr Stack und Heap. Bei ConnectMQ ist die Client ID an die ausgegebenen Zugangsdaten gebunden, verwenden Sie also genau die, die Ihnen das Portal gibt. Sobald Daten fließen, öffnen Sie den Live-Topic-Explorer im Portal, um zu bestätigen, dass Nachrichten ankommen, und legen Sie eine Alarmregel mit Schwellwert für das Temperaturfeld an.

FreeRTOS auf dem ESP32 ist in der ESP-IDF-FreeRTOS-Referenz ausführlich dokumentiert, einschließlich der SMP-Unterschiede zum Upstream-FreeRTOS. Einen breiteren Überblick über Geräteprojekte bietet das ESP32-IoT-Tutorial, das den Rest dieser Reihe bündelt, und NTP-Zeitstempel in die Struct Reading aufzunehmen, ist ein naheliegender nächster Schritt, damit gepufferte Messwerte ihren tatsächlichen Erfassungszeitpunkt tragen.

Starten Sie mit ConnectMQ

Ein Task-basierter Sketch ist nur die halbe Miete für ein zuverlässiges System; die andere Hälfte ist ein Broker, der verfügbar bleibt, während sich Ihre Geräte neu verbinden. ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS auf Port 8883 und einem Portal, das Ihre Geräte und Topics live anzeigt. Starten Sie ohne Kreditkarte mit dem kostenlosen Serverless-Plan und wechseln Sie zu einem kostenpflichtigen Serverless- oder Dedicated-Plan Ihres verwalteten MQTT-Brokers, wenn Ihre Flotte über 25 Verbindungen hinauswächst. Sobald die Firmware stabil ist, funktioniert dasselbe Muster mit echten Sensoren wie im ESP32-DHT22-Projekt.

Häufig gestellte Fragen

Kann ich PubSubClient aus mehr als einem FreeRTOS-Task aufrufen?

Nicht sicher. PubSubClient hat einen einzigen gemeinsamen Puffer und keine Sperren, sodass gleichzeitige Aufrufe Pakete beschädigen. Halten Sie den Client in einem Task und senden Sie Daten über eine Queue an diesen Task.

Auf welchem Core sollte der MQTT-Task laufen?

Core 0 ist eine gute Standardwahl, weil die WLAN- und lwIP-Tasks bereits dort laufen. Auf Single-Core-Chips wie dem ESP32-C3 oder ESP32-C6 pinnen Sie jeden Task auf Core 0 oder verwenden xTaskCreate ohne Pinning.

Wie groß sollte die Queue für Messwerte sein?

Dimensionieren Sie sie für den längsten Ausfall, den Sie überbrücken möchten. Multiplizieren Sie die Abtastrate mit der Ausfalldauer und der Struct-Größe und halten Sie die Summe deutlich unter dem freien Heap, den ESP.getFreeHeap() meldet.

Funktioniert das auch mit AsyncMqttClient?

Ja, doch AsyncMqttClient betreibt bereits einen eigenen Netzwerk-Task und eigene Callbacks, sodass Sie nur den Sensor-Task und eine Queue benötigen. Veröffentlichen Sie aus dem Task, der die Queue leert, nicht aus den Callbacks der Bibliothek.

ESP32 FreeRTOS MQTT: getrennte Sensor- und Publish-Tasks mit Queues