Aus AWS Lambda und Serverless-Funktionen an MQTT publishen

Du kannst aus AWS Lambda (oder jeder anderen Serverless-Funktion) mit einer ganz normalen MQTT-Client-Bibliothek an einen Broker publishen – der Haken ist der Lebenszyklus. Eine Funktion lebt ein paar hundert Millisekunden, wird zwischen Aufrufen eingefroren und läuft unter Last in Dutzenden parallelen Kopien. Verbinde dich lazy, cache den Client über warme Aufrufe hinweg, warte vor dem Return auf die Publish-Bestätigung und gib jeder Ausführungsumgebung eine eigene Client-ID. Wenn du nur publishst, ist eine HTTP-Bridge eventuell noch einfacher.

Dieses Tutorial zeigt einen produktionsreifen Python-Handler (paho-mqtt 2.x), dieselbe Idee in Node.js, die HTTP-Alternative und wie du das Ganze testest.

Serverless-Lebenszyklus gegen MQTT

MQTT geht von langlebigen TCP-Verbindungen mit Keep-Alive-Pings aus. Lambda geht von kurzen, zustandslosen Ausführungen aus. Der Widerspruch zeigt sich an vier Stellen:

Der Freeze ist die tückische Stelle. Ist eine Umgebung länger als etwa das 1,5-Fache des Keep-Alive-Intervalls inaktiv, schließt der Broker die Verbindung. Taut die Umgebung wieder auf, glaubt der Client noch verbunden zu sein, bis er schreiben will. Ein naiv gecachter Client schlägt deshalb beim ersten Aufruf nach einer ruhigen Phase fehl oder hängt.

Strategie 1: Verbindung pro Aufruf

Das einfachste korrekte Muster: verbinden, publishen, auf PUBACK warten, trennen. Zwischen den Aufrufen überlebt nichts, also kann auch nichts veralten.

• Vorteile: keine Bugs durch veraltete Verbindungen, keine Client-ID-Kollisionen bei zufälliger ID, leicht nachvollziehbar.
• Nachteile: Jeder Aufruf zahlt den Verbindungs-Handshake – mit TLS und aus einer entfernten Region deutlich spürbar –, und eine Welle von Aufrufen wird zu einer Welle neuer Verbindungen am Broker.

Für Trigger mit geringem Volumen ist das eine gute Wahl: ein nächtlicher Job, eine S3-Upload-Benachrichtigung, ein gelegentlicher Webhook.

Strategie 2: Verbindung über warme Aufrufe wiederverwenden

Für alles, was häufig läuft, erzeugst du den Client im globalen Scope, damit warme Aufrufe den Handshake überspringen – und misstraust der gecachten Verbindung nach einer Pause.

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            # nach dieser Pause: annehmen, dass der Broker uns getrennt hat

_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]}",   # eindeutig pro Umgebung
        protocol=mqtt.MQTTv5,
    )
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()                                         # CA-Bundle des Systems
    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()                            # nach einem Freeze veraltet
        _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 erneut versuchen lassen
    _last_used = time.time()
    return {"statusCode": 200, "body": json.dumps({"mid": info.mid})}

Was die einzelnen Teile tun:

• connect_async + loop_start – die Netzwerkschleife läuft in einem Hintergrund-Thread und kümmert sich um CONNACK, PUBACK und Reconnects, solange die Funktion läuft.
• MAX_IDLE – war die Umgebung länger als das Keep-Alive inaktiv, verwirft der Code den Client und verbindet neu, statt in einen toten Socket zu schreiben.
• wait_for_publish – kehre nie zurück, bevor die QoS-1-Bestätigung da ist. Nach dem Return kann die Umgebung eingefroren werden, während die Nachricht noch im Puffer des Clients liegt.
• Exception bei Fehlern – bei asynchronen Triggern wiederholt Lambda fehlgeschlagene Aufrufe, genau das willst du bei einem verlorenen Publish.
• Zufällige Client-ID – zwei gleichzeitige Umgebungen mit derselben ID würden sich ständig gegenseitig trennen (doppelte Client-ID).

Pack das Ganze mit pip install paho-mqtt -t package/ und zippe den Ordner zusammen mit dem Handler, oder nutze einen Lambda Layer. Setze das Funktions-Timeout deutlich über das Connect-Timeout (10–15 s sind ein vernünftiger Startwert) und lege die Credentials in Umgebungsvariablen ab, die aus Secrets Manager oder Parameter Store kommen.

Dasselbe Muster in Node.js

Mit MQTT.js 5 ist der gecachte Client eine Variable auf Modulebene. Die Bibliothek verbindet sich selbst neu, nach einem Freeze solltest du die Verbindung vor dem Publish aber trotzdem prüfen.

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 };
};

publishAsync mit QoS 1 resolved erst nach dem PUBACK, der Return nach dem await ist also sicher.

Strategie 3: stattdessen über HTTP publishen

Wenn deine Funktionen nur publishen und clientseitig keine MQTT-Semantik brauchen, passt ein HTTP-Request perfekt zu Serverless: kein Verbindungszustand, keine Client-IDs, keine Freeze-Probleme.

• HTTP-APIs des Brokers. Manche Broker bieten einen REST-Endpoint, der eine Nachricht für dich publisht – etwa die HTTP-API von EMQX oder der HTTPS-Data-Plane-Endpoint von AWS IoT Core. Authentifizierung und Payload-Format stehen in der Doku deines Brokers.
• Eigene Bridge. Ein kleiner, ständig laufender Service, der eine MQTT-Verbindung hält und authentifizierte POST /publish-Requests annimmt. Er bündelt die Credentials und macht aus tausend kurzlebigen Lambda-Verbindungen eine einzige dauerhafte.

Der Preis: Du verlierst die Ende-zu-Ende-QoS – das HTTP 200 sagt dir, dass die Bridge die Nachricht angenommen hat, nicht dass ein Subscriber sie erhalten hat –, und du hast eine Komponente mehr zu betreiben. Einen breiteren Vergleich beider Protokolle findest du unter MQTT vs. HTTP.

Und das Abonnieren?

Eine Lambda-Funktion kann kein langlebiger MQTT-Subscriber sein: Sie läuft nur während eines Aufrufs. Dreh die Richtung um – lass etwas Dauerhaftes abonnieren und die Funktion aufrufen:

• Eine Broker-Regel oder -Integration, die passende Nachrichten an einen HTTP-Endpoint weiterleitet (API Gateway oder eine Lambda Function URL). Das allgemeine Muster beschreibt MQTT zu Webhook.
• Ein kleiner Subscriber-Service, der Nachrichten in SQS schiebt, das wiederum Lambda in Batches auslöst.

Häufige Fehler

• Task timed out after 3.00 seconds – das Standard-Timeout von Lambda ist kürzer als ein kalter TLS-Connect aus einer anderen Region. Erhöhe es und prüfe, ob der Broker-Port erreichbar ist.
• Connect hängt in einer VPC – Funktionen in einer VPC brauchen ein NAT Gateway für Internet-Egress. Siehe MQTT-Verbindungs-Timeout.
• Not authorized / Reason Code 0x87 – falsche Credentials oder eine ACL, die das Topic nicht erlaubt; die Reason-Code-Referenz nennt die genaue Bedeutung.
• Nachrichten fehlen, nachdem die Funktion zurückgekehrt ist – du hast vor dem PUBACK zurückgegeben. Warte immer, bis der Publish abgeschlossen ist.

Testen

1. Öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und abonniere testmqtt/lambda/#.
2. Rufe die Funktion in der Konsole auf oder mit aws lambda invoke --function-name mqtt-publisher --payload '{"hello":"world"}' --cli-binary-format raw-in-base64-out out.json.
3. Die Nachricht sollte innerhalb einer Sekunde im Browser erscheinen. Ruf sofort noch einmal auf: Der warme Aufruf sollte in der Duration-Metrik der Funktion spürbar schneller sein.
4. Warte ein paar Minuten, ruf erneut auf und prüfe, dass der Pfad für veraltete Verbindungen sauber neu verbindet, statt in einen Timeout zu laufen.

Sieh die Nachrichten deiner Lambda ankommen

Der geteilte Test-Endpoint ist ratenbegrenzt und speichert keine Sessions – für diesen Check reicht das. Für Lasttests mit vielen parallelen Aufrufen richte die Funktion auf deine eigene private Broker-Instanz.

Serverless und MQTT passen gut zusammen, wenn du den Lebenszyklus respektierst: lazy verbinden, Verbindungen nach einem Freeze misstrauen, auf die Bestätigung warten und jeder Umgebung eine eigene Identität geben. Brauchst du nur Fire-and-Forget-Publishing in großem Maßstab, nimmt dir eine HTTP-Bridge das Verbindungsmanagement komplett ab.

Häufig gestellte Fragen