ESP32 FreeRTOS MQTT: Sensör ve Publish Görevleri

Bir ESP32 FreeRTOS MQTT tasarımı, sensör örneklemeyi ve ağ üzerinden yayını ayrı görevlere (task) yerleştirir; böylece yavaş bir Wi-Fi yeniden bağlantısı hiçbir ölçümü geciktirmez. Bu eğitimde bir sensör görevi, bir MQTT görevi ve aralarında bir kuyruk bulunan bir sketch oluşturacak, bunu broker.connectmq.com'un ücretsiz herkese açık alanında test edecek ve ardından kendi kimlik bilgilerinizle TLS'e geçireceksiniz.

ESP32'de MQTT İçin Neden FreeRTOS Görevleri Kullanılmalı

Başlangıç seviyesindeki sketch'lerin çoğu her şeyi loop() içinde yapar: sensörü oku, Wi-Fi'ı kontrol et, client.loop() çağır, publish yap, sonra delay(). Bu, ağ sorun çıkarana kadar işe yarar. Erişilemeyen bir broker'a yapılan connect() çağrısı birkaç saniye bloklayabilir ve bu süre boyunca hiçbir örnek alınmaz. Darbe sayıyorsanız, analog bir sinyalin ortalamasını alıyorsanız veya bir eşiği izliyorsanız bu boşluklar önemlidir.

ESP32 Arduino core zaten FreeRTOS üzerinde çalışır. loop() da aslında core 1'e sabitlenmiş bir görevden ibarettir. Daha fazla görev oluşturmanızı engelleyen hiçbir şey yoktur ve bunu yaptığınızda bir işin zamanlaması artık diğerine bağlı olmaz:

Sensör görevi vTaskDelayUntil() ile sabit bir periyotta uyanır, donanımı okur ve küçük bir struct'ı kuyruğa bırakır. Ağa asla dokunmaz.
MQTT görevi Wi-Fi'ın ve MQTT istemcisinin sahibidir. Kuyruğu boşaltır, publish yapar, keep-alive'ları yönetir ve gerektiğinde yeniden bağlanır.
Kuyruk tek paylaşılan nesnedir. FreeRTOS kuyrukları veriyi kopyalar ve görevler arasında güvenle kullanılabilir; bu yüzden ölçümler için mutex'e ihtiyacınız yoktur.

Bu tasarımı sağlam kılan tek bir kural vardır: MQTT istemcisini yalnızca bir görev çağırır. PubSubClient thread-safe değildir ve iki görevden aynı anda publish() çağırmak dahili tamponunu bozar. Bunun yerine her şeyi kuyruk üzerinden yönlendirin.

Gerekenler

İki çekirdekli herhangi bir ESP32 kartı (ESP32 DevKit, ESP32-S3). ESP32-C3 veya ESP32-C6 gibi tek çekirdekli çipler de her iki görevi core 0'a sabitlerseniz çalışır.
Arduino IDE 2.x veya esp32 core 3.x ile PlatformIO.
Nick O'Leary'nin PubSubClient kütüphanesi.
İsteğe bağlı: herhangi bir sensör. Sketch, çıplak bir kartta da çalışması için sıcaklık ve nemi simüle eder.
Test için bilgisayarınızda mosquitto_sub.

Görev Düzeni ve Öncelikler

Görev	Core	Öncelik	Stack	İş
`sensorTask`	1	3	3072 bayt	Her 1 sn'de örnekle, kuyruğa ekle
`mqttTask`	0	2	6144 bayt	Wi-Fi, MQTT bağlantısı, publish, keep-alive
Arduino `loop()`	1	1	varsayılan	Silinir, gerekmez

Sensör görevi kısa ve zaman açısından kritik olduğu için daha yüksek önceliği alır; çalışır, kuyruğa bir öğe ekler ve uyur. MQTT görevi, yine orada bulunan Wi-Fi yığınıyla birlikte core 0'da çalışır ve yavaş işleri yapar. ESP32 Arduino core'da stack boyutları, standart FreeRTOS'taki gibi word cinsinden değil bayt cinsinden verilir.

Eksiksiz ESP32 FreeRTOS MQTT Sketch'i

Wi-Fi kimlik bilgilerini ve topic önekinin rastgele kısmını değiştirin, ardından yükleyin.

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

// ---------- yapılandırma ----------
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";   // bunu benzersiz yapın

// ---------- paylaşılan nesneler ----------
struct Reading {
  uint32_t seq;
  uint32_t uptimeMs;
  float    temperature;
  float    humidity;
};

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

// ---------- sensör görevi ----------
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();
    // Gerçek bir sensör okumasıyla değiştirin (DHT22, BME280, ADC...)
    r.temperature = 21.0f + (esp_random() % 400) / 100.0f;
    r.humidity    = 45.0f + (esp_random() % 1000) / 100.0f;

    // Bloklama: kuyruk doluysa en eski öğeyi at ve tekrar dene
    if (xQueueSend(readingQueue, &r, 0) != pdTRUE) {
      Reading discarded;
      xQueueReceive(readingQueue, &discarded, 0);
      xQueueSend(readingQueue, &r, 0);
    }
    vTaskDelayUntil(&lastWake, period);
  }
}

// ---------- MQTT yardımcıları (yalnızca mqttTask'tan çağrılır) ----------
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 görevi ----------
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;
    // Görev boştayken CPU'yu bırakması için veriyi kısa süre bekle
    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)) {
        // Kuyruğun başına geri koy ve yeniden bağlandıktan sonra tekrar dene
        xQueueSendToFront(readingQueue, &r, 0);
      }
    }
  }
}

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

  readingQueue = xQueueCreate(60, sizeof(Reading));   // bir dakikalık tampon
  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);   // Arduino loop görevine gerek yok
}

Çıktıyı bilgisayarınızdan izleyin:

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

online değerinde bir status mesajı ve saniyede bir telemetri mesajı görmelisiniz. Şimdi modeminizin fişini yirmi saniyeliğine çekin. Kart çevrimdışıyken queued alanı büyür ve bağlantı geri geldiğinde MQTT görevi birikmiş mesajları sırayla boşaltır. Tasarımın tüm amacı budur: yayın durmuş olsa bile ölçümler zamanında alındı.

Broker'ın herkese açık alanı herkes tarafından okunabilir; bu yüzden demo verilerini genel tutun. Bu alan ayrıca her istemciyi saniyede 10 mesajla sınırlar ve birikmiş mesajların boşaltılması bu sınıra takılabilir. Çok sayıda ölçümü tamponluyorsanız, uxQueueMessagesWaiting() yüksekken her publish'ten sonra kısa bir vTaskDelay(pdMS_TO_TICKS(120)) ekleyin. Ücretsiz herkese açık MQTT broker rehberi diğer sınırları listeler.

Parçalar Nasıl Çalışır

Tampon olarak kuyruk

xQueueCreate(60, sizeof(Reading)) heap üzerinde sabit boyutlu 60 yuva ayırır. Her xQueueSend 16 baytlık struct'ı kopyalar; böylece sensör görevi yerel değişkenini hemen yeniden kullanabilir. Kuyruğu boyutlandırmak, RAM ile atlatmak istediğiniz kesintinin süresi arasında bir dengedir. 16 baytlık altmış yuva 1 KB'ın altındadır; birkaç bin yuva bile klasik bir ESP32'ye kolayca sığar, ancak verinin yeniden başlatmada kaybolduğunu unutmayın.

Kuyruk dolduğunda sketch en yeni ölçümü değil en eskisini atar. Telemetri için genellikle en son değer en yararlı olanıdır. Her örnek önemliyse bunun yerine birkaç tick bloklayın veya kuyruğa eklemeden önce ölçümleri toplulaştırın (minimum, maksimum, ortalama).

vTaskDelayUntil ile delay karşılaştırması

vTaskDelayUntil() görevi bir önceki uyanma zamanından ölçülen sabit bir hızda uyandırır; böylece periyot, sensör okumasının ne kadar sürdüğüne bağlı olarak kaymaz. 30 ms'lik bir okumadan sonra düz bir delay(1000) size yavaşça kayan 1,03 sn'lik bir periyot verir. millis() ile yapılan meşgul bir döngünün aksine her iki fonksiyon da CPU'yu bırakır.

Backoff ile yeniden bağlanma

MQTT görevi, 30 saniyeyle sınırlanmış üstel backoff ile yeniden dener. Broker'ı anlık yeniden denemelerle bombardımana tutmak enerji israf eder ve yönetilen bir serviste flood korumasını veya tekrarlanan hatalar nedeniyle IP engellemeyi tetikleyebilir. ESP32 MQTT Last Will rehberi, retained offline will mesajının neden bağlantı anında kaydedildiğini açıklar; böylece dashboard'lar ölü bir cihazı fark eder.

Tuzaklar ve Sorun Giderme

Guru Meditation "Stack canary watchpoint triggered". Görev stack'i çok küçüktür. Float'larla snprintf ve TLS'in ikisi de pay gerektirir. Geliştirme sırasında her görevin içinde uxTaskGetStackHighWaterMark(nullptr) yazdırın; bu, görülen minimum boş bayt miktarını döndürür. En az 512 bayt boş bırakın.
Task watchdog sıfırlamaları. Hiç bloklamadan dönen bir görev, kendi core'undaki idle görevini aç bırakır. Her döngü yinelemesi CPU'yu bırakan bir şey çağırmalıdır: vTaskDelay, zaman aşımlı xQueueReceive veya bloklayan bir soket çağrısı.
Karışık seri çıktı. Aynı anda yazdıran iki görev karakterleri iç içe geçirir. Ya yalnızca tek bir görevden yazdırın ya da Serial'ı xSemaphoreCreateMutex() ile oluşturulmuş bir mutex ile koruyun.
Sensör görevinden publish yapmak. Çalışıyor gibi görünür, sonra yük altında rastgele başarısız olur. MQTT istemcisini tek bir görevde tutun.
Callback'ler MQTT görevinde çalışır. Komutlara subscribe olursanız PubSubClient callback'i mqtt.loop() içinde çalışır. Orada yavaş işler yapmayın; komutu başka bir görevin işlemesi için ikinci bir kuyruğa itin.
Büyük JSON'un kesilmesi. PubSubClient'ın varsayılan tamponu topic dahil 256 bayttır. Sketch bunu setBufferSize(512) ile artırır.

TLS'e ve Kendi Broker'ınıza Geçiş

Herkese açık alan test için idealdir, ancak production cihazlarının kimlik doğrulaması yapması ve şifreleme kullanması gerekir. app.connectmq.com adresinde ücretsiz bir hesap açın, bir broker oluşturun ve otomatik oluşturulan kullanıcı adını, parolayı ve client ID'yi kopyalayın. Ardından MQTT görevinin yalnızca ağ kısmını değiştirin:

#include <WiFiClientSecure.h>

// ISRG Root X1, Let's Encrypt kök sertifikası. Let's Encrypt'in yayımladığı PEM'i yapıştırın.
static const char ROOT_CA[] = R"PEM(
-----BEGIN CERTIFICATE-----
...paste ISRG Root X1 here...
-----END CERTIFICATE-----
)PEM";

WiFiClientSecure net;
PubSubClient mqtt(net);

// mqttTask içinde, döngüden önce:
net.setCACert(ROOT_CA);
mqtt.setServer("broker.connectmq.com", 8883);

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

TLS el sıkışması belirgin biçimde daha fazla stack ve heap kullandığı için MQTT görevinin stack'ini yaklaşık 8192 bayta çıkarın. ConnectMQ'da client ID verilen kimlik bilgilerine bağlıdır, bu yüzden portalın size verdiğini birebir kullanın. Veri akmaya başladığında, mesajların ulaştığını doğrulamak için portaldaki canlı topic gezginini açın ve sıcaklık alanı için bir eşik uyarı kuralı tanımlayın.

ESP32'deki FreeRTOS, upstream FreeRTOS'tan SMP farklılıkları da dahil olmak üzere ESP-IDF FreeRTOS referansında ayrıntılı olarak belgelenmiştir. Cihaz projelerine daha geniş bir bakış için ESP32 IoT eğitimi bu serinin geri kalanını bir araya getirir; Reading struct'ına NTP zaman damgaları eklemek ise tamponlanan örneklerin gerçek alınma zamanlarını taşıması için doğal bir sonraki adımdır.

ConnectMQ ile Başlayın

Görev tabanlı bir sketch güvenilir bir sistemin yalnızca yarısıdır; diğer yarısı, cihazlarınız yeniden bağlanırken ayakta kalan bir broker'dır. ConnectMQ, MQTT 3.1.1 ve 5.0, 8883 portunda TLS ve cihazlarınızı ve topic'lerinizi canlı gösteren bir portal sunan tam yönetilen bir MQTT servisidir. Kredi kartı gerektirmeyen ücretsiz Serverless planıyla başlayın, filonuz 25 bağlantıyı aştığında ücretli veya dedicated bir yönetilen MQTT broker planına geçin. Firmware sağlamlaştığında aynı desen, ESP32 DHT22 projesi gibi gerçek sensörlerle de çalışır.

Sık Sorulan Sorular

PubSubClient'ı birden fazla FreeRTOS görevinden çağırabilir miyim?

Güvenli bir şekilde değil. PubSubClient tek bir paylaşılan tampona sahiptir ve kilitleme yapmaz; bu yüzden eşzamanlı çağrılar paketleri bozar. Onu tek bir görevde tutun ve veriyi o göreve bir kuyruk üzerinden gönderin.

MQTT görevi hangi core üzerinde çalışmalı?

Wi-Fi ve lwIP görevleri zaten orada çalıştığı için core 0 iyi bir varsayılandır. ESP32-C3 veya ESP32-C6 gibi tek çekirdekli çiplerde her görevi core 0'a sabitleyin ya da sabitleme olmadan xTaskCreate kullanın.

Ölçüm kuyruğu ne kadar büyük olmalı?

Atlatmak istediğiniz en uzun kesintiye göre boyutlandırın. Örnekleme hızını kesinti süresi ve struct boyutuyla çarpın ve toplamı ESP.getFreeHeap() tarafından bildirilen boş heap'in oldukça altında tutun.

Bu yöntem AsyncMqttClient ile çalışır mı?

Evet, ancak AsyncMqttClient zaten kendi ağ görevini ve callback'lerini çalıştırır; bu yüzden yalnızca sensör görevine ve bir kuyruğa ihtiyacınız olur. Publish işlemini kütüphane callback'lerinin içinden değil, kuyruğu boşaltan görevden yapın.

ESP32 FreeRTOS MQTT: Kuyruklarla Ayrı Sensör ve Yayın Görevleri