MQTT in Dart (serverseitig) mit mqtt_client

Die meisten mqtt_client-Beispiele sind Flutter-Apps, doch das Paket ist reines Dart: MqttServerClient läuft in jedem Dart-VM-Programm. Damit eignet sich Dart gut für MQTT-Kommandozeilen-Tools, kleine Backend-Services und Geräte-Gateways, die zu einer einzigen nativen Binary kompiliert werden.

In diesem Tutorial baust du zwei Tools – einen dauerhaft laufenden Subscriber mit Auto-Reconnect und einen One-Shot-Publisher, der auf die Bestätigung des Brokers wartet – und ergänzt dann TLS, WebSockets und einen nativen Build. Alles zielt auf broker.connectmq.com, du kannst jeden Schritt also im Online-MQTT-Client gegenprüfen. Baust du stattdessen eine Mobile-App, lies MQTT in Flutter.

Projekt anlegen

dart create -t console mqtt_cli
cd mqtt_cli
dart pub add mqtt_client

environment:
  sdk: ^3.4.0

dependencies:
  mqtt_client: ^10.2.0

mqtt_client bringt zwei Client-Klassen mit: MqttServerClient (Sockets über dart:io, für die VM) und MqttBrowserClient (WebSockets, für Dart im Web). Auf dem Server willst du immer MqttServerClient. Das Paket spricht MQTT 3.1.1; für MQTT 5 hat das Schwesterpaket mqtt5_client eine fast identische API – in MQTT 5 vs. 3.1.1 findest du, ob du es brauchst.

Exportiere die Zugangsdaten deines Brokers, damit sie nie in der Versionsverwaltung landen:

export MQTT_USER="username"
export MQTT_PASS="password"

Ein dauerhaft laufender Subscriber

import 'dart:convert';
import 'dart:io';

import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

Future<void> main() async {
  final clientId = 'testmqtt-dart-sub-$pid';
  final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 1883)
    ..logging(on: false)
    ..setProtocolV311()
    ..keepAlivePeriod = 30
    ..autoReconnect = true
    ..resubscribeOnAutoReconnect = true
    ..onConnected = (() => print('connected'))
    ..onDisconnected = (() => print('disconnected'))
    ..onAutoReconnect = (() => print('connection lost, reconnecting...'))
    ..onAutoReconnected = (() => print('reconnected'))
    ..onSubscribed = ((topic) => print('subscribed to $topic'));

  client.connectionMessage = MqttConnectMessage()
      .withClientIdentifier(clientId)
      .startClean()
      .authenticateAs(Platform.environment['MQTT_USER'], Platform.environment['MQTT_PASS']);

  try {
    await client.connect();
  } on NoConnectionException catch (e) {
    stderr.writeln('broker rejected or ignored CONNECT: $e');
    exit(1);
  } on SocketException catch (e) {
    stderr.writeln('network error: $e');
    exit(1);
  }

  if (client.connectionStatus!.state != MqttConnectionState.connected) {
    stderr.writeln('connect failed: ${client.connectionStatus!.returnCode}');
    exit(1);
  }

  client.subscribe('testmqtt/dart/#', MqttQos.atLeastOnce);

  client.updates!.listen((messages) {
    for (final m in messages) {
      final publish = m.payload as MqttPublishMessage;
      final text = utf8.decode(publish.payload.message, allowMalformed: true);
      print('${m.topic}  $text');
    }
  });

  ProcessSignal.sigint.watch().listen((_) {
    client.disconnect();
    exit(0);
  });
}

Ein paar Details, die in Produktion zählen:

• Client-ID. pid macht die ID pro Prozess auf einer Maschine eindeutig. Läuft dasselbe Tool auf mehreren Hosts, nimm auch den Hostnamen mit auf – zwei Clients mit derselben ID werfen sich gegenseitig in einer Schleife raus.
• updates liefert Listen. Jedes Event ist eine List<MqttReceivedMessage>, meist mit einem Eintrag; iteriere immer.
• Mit utf8 dekodieren. MqttPublishPayload.bytesToStringAsString wandelt Bytes einzeln in Zeichen um und zerstört damit Nicht-ASCII-Text wie °C. utf8.decode macht es richtig.
• Auto-Reconnect. Mit autoReconnect und resubscribeOnAutoReconnect wird eine abgebrochene Verbindung wiederhergestellt und deine Subscriptions werden erneuert. Reconnect-Strategien allgemein behandelt Reconnect Best Practices.

Starte ihn mit dart run bin/subscriber.dart und publiziere dann eine Nachricht auf testmqtt/dart/hello aus dem Online-Client. Sie sollte innerhalb von Sekundenbruchteilen erscheinen.

Aus dem Browser an den Dart-Subscriber publizieren

Ein One-Shot-Publisher, der auf PUBACK wartet

Kommandozeilen-Publisher haben einen klassischen Bug: Sie rufen publishMessage auf und beenden sich sofort, bevor die Bytes überhaupt auf der Leitung sind. Mit QoS 1 kannst du statt eines Sleeps auf die Bestätigung des Brokers im published-Stream warten:

import 'dart:io';

import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

Future<void> main(List<String> args) async {
  if (args.length != 2) {
    stderr.writeln('usage: publish <topic> <message>');
    exit(64);
  }

  final clientId = 'testmqtt-dart-pub-$pid';
  final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 1883)
    ..logging(on: false)
    ..setProtocolV311();
  client.connectionMessage = MqttConnectMessage()
      .withClientIdentifier(clientId)
      .startClean()
      .authenticateAs(Platform.environment['MQTT_USER'], Platform.environment['MQTT_PASS']);

  try {
    await client.connect();
  } on Exception catch (e) {
    stderr.writeln('connect failed: $e');
    exit(1);
  }

  final payload = (MqttClientPayloadBuilder()..addUTF8String(args[1])).payload!;
  final messageId = client.publishMessage(args[0], MqttQos.atLeastOnce, payload);

  try {
    await client.published!
        .firstWhere((m) => m.variableHeader?.messageIdentifier == messageId)
        .timeout(const Duration(seconds: 10));
    print('delivered to broker (id $messageId)');
  } on Exception {
    stderr.writeln('no PUBACK within 10 s');
    exitCode = 2;
  } finally {
    client.disconnect();
  }
  exit(exitCode);
}

dart run bin/publish.dart testmqtt/dart/hello "21.5 °C"
echo "exit code: $?"

Die Exit-Codes machen das Tool skriptfähig: 0 heißt, der Broker hat die Nachricht bestätigt, 1 ein Verbindungsproblem, 2 eine fehlende Bestätigung. Ein PUBACK bedeutet, dass der Broker die Verantwortung für die Nachricht übernommen hat – nicht, dass ein Subscriber sie verarbeitet hat. Was jedes QoS-Level wirklich garantiert, steht in MQTT-QoS-Level.

TLS auf Port 8883

Für eine verschlüsselte Verbindung brauchst du zwei Properties. SecurityContext.defaultContext nutzt die vertrauenswürdigen CAs des Betriebssystems – genau richtig für einen Broker mit öffentlichem Zertifikat:

final client = MqttServerClient.withPort('broker.connectmq.com', clientId, 8883)
  ..secure = true
  ..securityContext = SecurityContext.defaultContext
  ..logging(on: false)
  ..setProtocolV311();

Für eine private CA erzeugst du einen SecurityContext(withTrustedRoots: false) und rufst darauf setTrustedCertificates('ca.pem') auf; für Mutual TLS ergänzt du useCertificateChain und usePrivateKey. Widerstehe der Abkürzung onBadCertificate = (_) => true aus Forenantworten – sie schaltet die Prüfung komplett ab. Schlägt der Handshake fehl, folge stattdessen SSL certificate verify failed.

WebSockets, wenn nur 443-artige Ports offen sind

Manche Netze blockieren 1883 und 8883, lassen aber HTTPS-ähnlichen Traffic durch. MqttServerClient kann MQTT über WebSockets tunneln; Schema und Pfad gehören in den Server-String, der Port separat:

final client = MqttServerClient.withPort('wss://broker.connectmq.com/mqtt', clientId, 8084)
  ..useWebSocket = true
  ..logging(on: false)
  ..setProtocolV311();

Nutze ws:// mit Port 8083 für einen unverschlüsselten Test. Lehnt der Broker das WebSocket-Upgrade ab, probiere ..websocketProtocols = MqttClientConstants.protocolsSingleDefault, das nur das Subprotokoll mqtt statt einer Liste anbietet. Mehr Hintergrund in MQTT über WebSockets.

Eine native Binary bauen

dart compile exe erzeugt ein eigenständiges Executable, das sofort startet und kein Dart-SDK auf der Zielmaschine braucht:

dart compile exe bin/subscriber.dart -o mqtt-sub
dart compile exe bin/publish.dart -o mqtt-pub
./mqtt-pub testmqtt/dart/hello "from a native binary"

Die Binary wird für das Betriebssystem und die CPU gebaut, auf denen du kompilierst. Baue deshalb in einem Linux-Container (das offizielle dart-Image eignet sich gut), wenn du auf Linux-Server auslieferst. Ein Multi-Stage-Dockerfile, das die Binary plus /runtime/ aus der Build-Stage in ein scratch-Image kopiert, hält das finale Image sehr klein.

Als Service betreiben

Ein Subscriber, der wochenlang läuft, braucht etwas mehr Sorgfalt als eine Demo:

• SIGTERM zusätzlich zu SIGINT behandeln. Docker, Kubernetes und systemd stoppen Prozesse mit SIGTERM. Höre auch auf ProcessSignal.sigterm.watch() (unter Windows nicht verfügbar) und rufe client.disconnect() auf, damit der Broker ein sauberes DISCONNECT statt einer abgelaufenen Verbindung sieht.
• Handler nicht blockieren. Der updates-Listener läuft im Haupt-Isolate. Lange CPU-Arbeit dort verzögert die Keep-Alive-Pings; verlagere sie in Isolate.run oder eine Queue.
• Zustandswechsel loggen, nicht jede Nachricht. onDisconnected, onAutoReconnect und onAutoReconnected sind die Events, die dir um 3 Uhr nachts sagen, dass etwas nicht stimmt.
• Konfiguration aus der Umgebung lesen. Host, Port, Topic-Präfix und Zugangsdaten als Umgebungsvariablen erlauben es, dieselbe Binary in Test, Staging und Produktion zu betreiben.

So testest du deine Tools

Bevor du die Binaries in einen Cronjob oder Container steckst, mach drei schnelle Checks. Starte den Subscriber, publiziere aus dem Online-Client und prüfe, dass die Nachricht erscheint. Trenne dann für eine halbe Minute das Netzwerk (oder blockiere den Port per Firewall-Regel) und prüfe, dass reconnected erscheint und Nachrichten ohne Neustart wieder fließen. Zuletzt startest du den Publisher mit falschem Passwort und prüfst, dass er mit Code 1 endet – Skripte, die ihn aufrufen, verlassen sich darauf.

Fehlerbehebung

client.logging(on: true) gibt jedes Paket aus, das der Client sendet und empfängt – so siehst du schnell, ob CONNACK, SUBACK und PUBACK ankommen.

Serverseitiges Dart bringt dir einen typisierten, durchgehend asynchronen MQTT-Client und native Binaries ohne zu installierende Runtime. Halte die Verbindungslogik an einer Stelle, warte auf Bestätigungen statt zu schlafen, nutze TLS außerhalb deines Laptops und prüfe jedes Tool gegen einen echten Broker, bevor es in einen Cronjob oder Container wandert.

Häufig gestellte Fragen