AWS Lambda ve Serverless Fonksiyonlardan MQTT'ye Publish

AWS Lambda'dan (veya herhangi bir serverless fonksiyondan) sıradan bir MQTT istemci kütüphanesiyle bir broker'a publish edebilirsiniz; işin püf noktası yaşam döngüsüdür. Bir fonksiyon birkaç yüz milisaniye yaşar, çağrılar arasında dondurulur ve yük altında onlarca paralel kopya halinde çalışabilir. Bağlantıyı ihtiyaç anında kurun, istemciyi sıcak çağrılar arasında önbellekte tutun, dönmeden önce publish onayını bekleyin ve her çalışma ortamına kendi client ID'sini verin. Yalnızca publish ediyorsanız bir HTTP köprüsü daha da basit olabilir.

Bu eğitim production'a hazır bir Python handler'ı (paho-mqtt 2.x), aynı fikrin Node.js versiyonunu, HTTP alternatifini ve hepsinin nasıl test edileceğini gösteriyor.

Serverless yaşam döngüsü ve MQTT

MQTT, keep-alive ping'leri olan uzun ömürlü TCP bağlantılarını varsayar. Lambda ise kısa ve durumsuz çalışmaları varsayar. Uyumsuzluk dört yerde ortaya çıkar:

Dondurma en sinsi olanıdır. Bir ortam keep-alive aralığının yaklaşık 1,5 katından uzun süre boşta kalırsa broker bağlantıyı kapatır. Ortam çözüldüğünde istemci, yazmayı deneyene kadar hâlâ bağlı olduğunu sanır. Bu yüzden saf bir şekilde önbelleğe alınmış bir istemci, sakin bir dönemin ardından gelen ilk çağrıda hata verir veya takılır.

Strateji 1: her çağrıda bağlanmak

En basit doğru kalıp: bağlan, publish et, PUBACK bekle, bağlantıyı kes. Çağrılar arasında hiçbir şey kalmadığı için hiçbir şey bayatlayamaz.

• Artıları: bayat bağlantı hatası yok, rastgele ID kullanırsanız client ID çakışması yok, anlaşılması kolay.
• Eksileri: her çağrı bağlantı el sıkışmasının bedelini öder — TLS ile ve uzak bir bölgeden belirgin şekilde daha fazla — ve bir çağrı patlaması broker'da bir yeni bağlantı patlamasına dönüşür.

Düşük hacimli tetikleyiciler için iyi bir seçimdir: gece çalışan bir iş, bir S3 yükleme bildirimi, ara sıra gelen bir webhook.

Strateji 2: bağlantıyı sıcak çağrılar arasında yeniden kullanmak

Sık çalışan her şey için istemciyi global kapsamda oluşturun ki sıcak çağrılar el sıkışmayı atlasın; önbellekteki bağlantıya ise bir boşta kalma süresinden sonra şüpheyle yaklaşın.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "8883"))
KEEPALIVE = 60
MAX_IDLE = KEEPALIVE            # bu aradan sonra broker'ın bizi düşürdüğünü varsay

_client = None
_connected = threading.Event()
_last_used = 0.0

def _on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect refused: {reason_code}")
    else:
        _connected.set()

def _on_disconnect(client, userdata, flags, reason_code, properties):
    _connected.clear()

def _new_client():
    c = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"lambda-{uuid.uuid4().hex[:12]}",   # ortam başına benzersiz
        protocol=mqtt.MQTTv5,
    )
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()                                         # sistem CA paketi
    c.on_connect, c.on_disconnect = _on_connect, _on_disconnect
    c.connect_async(HOST, PORT, keepalive=KEEPALIVE)
    c.loop_start()
    return c

def _get_client(timeout=5.0):
    global _client
    if _client is not None and time.time() - _last_used > MAX_IDLE:
        _client.disconnect()                            # dondurmadan sonra bayat
        _client.loop_stop()
        _connected.clear()
        _client = None
    if _client is None:
        _client = _new_client()
    if not _connected.wait(timeout):
        raise TimeoutError("MQTT connect timed out")
    return _client

def handler(event, context):
    global _last_used
    client = _get_client()
    payload = json.dumps({"source": "lambda", "event": event, "ts": time.time()})
    info = client.publish("testmqtt/lambda/events", payload, qos=1)
    info.wait_for_publish(timeout=5)
    if not info.is_published():
        raise RuntimeError("PUBACK not received")       # Lambda yeniden denesin
    _last_used = time.time()
    return {"statusCode": 200, "body": json.dumps({"mid": info.mid})}

Her parça ne yapıyor:

• connect_async + loop_start — ağ döngüsü bir arka plan thread'inde çalışır ve fonksiyon çalışırken CONNACK, PUBACK ve yeniden bağlanmaları yönetir.
• MAX_IDLE — ortam keep-alive süresinden uzun boşta kaldıysa kod, ölü bir socket'e yazmak yerine istemciyi atıp yeniden bağlanır.
• wait_for_publish — QoS 1 onayı gelmeden asla dönmeyin. Handler döndükten sonra ortam, mesaj hâlâ istemcinin tamponundayken dondurulabilir.
• Hata durumunda exception fırlatmak — asenkron tetikleyicilerde Lambda başarısız çağrıları yeniden dener; kaybolan bir publish için tam da istediğiniz budur.
• Rastgele client ID — aynı ID'ye sahip iki eşzamanlı ortam birbirini sürekli koparır (çift client ID).

pip install paho-mqtt -t package/ ile paketleyip klasörü handler ile birlikte zip'leyin veya bir Lambda layer kullanın. Fonksiyon zaman aşımını bağlantı zaman aşımının rahatça üzerine ayarlayın (10–15 sn makul bir başlangıç noktasıdır) ve kimlik bilgilerini Secrets Manager veya Parameter Store destekli ortam değişkenlerinde tutun.

Aynı kalıp Node.js'te

MQTT.js 5 ile önbellekteki istemci modül seviyesinde bir değişkendir. Kütüphane kendi kendine yeniden bağlanır, ancak bir dondurmanın ardından publish'ten önce bağlantıyı yine de doğrulamalısınız.

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

let client;
let lastUsed = 0;

async function getClient() {
  if (client && Date.now() - lastUsed > 60_000) { await client.endAsync(true); client = undefined; }
  if (!client) {
    client = await mqtt.connectAsync('mqtts://broker.connectmq.com:8883', {
      clientId: `lambda-${randomUUID().slice(0, 12)}`,
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      keepalive: 60,
      connectTimeout: 5000,
    });
  }
  return client;
}

export const handler = async (event) => {
  const c = await getClient();
  await c.publishAsync('testmqtt/lambda/events', JSON.stringify({ source: 'lambda', event }), { qos: 1 });
  lastUsed = Date.now();
  return { statusCode: 200 };
};

QoS 1 ile publishAsync yalnızca PUBACK geldikten sonra resolve olur; dolayısıyla await sonrasında dönmek güvenlidir.

Strateji 3: bunun yerine HTTP ile publish

Fonksiyonlarınız yalnızca publish ediyor ve istemci tarafında MQTT semantiğine ihtiyaç duymuyorsa, bir HTTP isteği serverless'a doğal olarak uyar: bağlantı durumu yok, client ID yok, dondurma sorunu yok.

• Broker HTTP API'leri. Bazı broker'lar sizin adınıza mesaj publish eden bir REST uç noktası sunar — örneğin EMQX'in HTTP API'si ve AWS IoT Core'un HTTPS data plane uç noktası. Kimlik doğrulama ve payload formatı için broker'ınızın dokümantasyonuna bakın.
• Kendi köprünüz. Tek bir MQTT bağlantısı tutan ve kimliği doğrulanmış POST /publish isteklerini kabul eden, sürekli çalışan küçük bir servis. Kimlik bilgilerini merkezileştirir ve bin kısa ömürlü Lambda bağlantısını tek bir kalıcı bağlantıya dönüştürür.

Bedeli: uçtan uca QoS'u kaybedersiniz — HTTP 200, köprünün mesajı kabul ettiğini söyler, bir subscriber'ın onu aldığını değil — ve işletmeniz gereken bir bileşen eklersiniz. İki protokolün daha geniş bir karşılaştırması için MQTT vs HTTP yazısına bakın.

Peki ya abonelik?

Bir Lambda fonksiyonu uzun ömürlü bir MQTT subscriber'ı olamaz: yalnızca bir çağrıyı işlerken çalışır. Yönü tersine çevirin — sürekli çalışan bir şeyin abone olup fonksiyonu çağırmasını sağlayın:

• Eşleşen mesajları bir HTTP uç noktasına (API Gateway veya bir Lambda function URL) ileten bir broker kuralı ya da entegrasyonu. Genel kalıp için MQTT'den webhook'a yazısına bakın.
• Mesajları SQS'e iten ve SQS'in de Lambda'yı toplu halde tetiklediği küçük bir abone servis.

Sık karşılaşılan hatalar

• Task timed out after 3.00 seconds — Lambda'nın varsayılan zaman aşımı, başka bir bölgeden yapılan soğuk bir TLS bağlantısından kısadır. Artırın ve broker portunun erişilebilir olduğunu kontrol edin.
• VPC içinde bağlantı takılıyor — VPC'deki fonksiyonların internete çıkmak için bir NAT gateway'e ihtiyacı vardır. MQTT bağlantı zaman aşımı yazısına bakın.
• Not authorized / reason code 0x87 — yanlış kimlik bilgileri veya topic'e izin vermeyen bir ACL; reason code referansı tam anlamını listeler.
• Fonksiyon döndükten sonra mesajlar eksik — PUBACK gelmeden döndünüz. Publish'in tamamlanmasını her zaman bekleyin.

Test etmek

1. Online MQTT istemcisini açın, broker.connectmq.com adresine bağlanın ve testmqtt/lambda/# topic'ine abone olun.
2. Fonksiyonu konsoldan veya aws lambda invoke --function-name mqtt-publisher --payload '{"hello":"world"}' --cli-binary-format raw-in-base64-out out.json ile çağırın.
3. Mesaj bir saniye içinde tarayıcıda görünmelidir. Hemen tekrar çağırın: sıcak çağrı, fonksiyonun süre metriğinde belirgin şekilde daha hızlı olmalıdır.
4. Birkaç dakika bekleyin, yeniden çağırın ve bayat bağlantı yolunun zaman aşımına uğramak yerine temiz şekilde yeniden bağlandığını doğrulayın.

Lambda mesajlarınızın gelişini izleyin

Paylaşılan test uç noktası hız sınırlıdır ve oturum tutmaz; bu kontrol için bu yeterlidir. Çok sayıda paralel çağrıyla yük testi yapacaksanız fonksiyonu kendi özel broker instance'ınıza yönlendirin.

Serverless ve MQTT, yaşam döngüsüne saygı gösterdiğinizde birlikte iyi çalışır: ihtiyaç anında bağlanın, dondurmadan sonra bağlantılara güvenmeyin, onayı bekleyin ve her ortama kendi kimliğini verin. Ölçekli, yanıt beklemeyen publish'e ihtiyacınız varsa bir HTTP köprüsü bağlantı yönetimini tamamen ortadan kaldırır.

Sıkça sorulan sorular