Go-MQTT-Client-Tutorial mit Eclipse Paho autopaho und MQTT 5

Ein Go-MQTT-Client passt hervorragend zu Backend-Diensten, die Gerätedaten aufnehmen, denn Go bewältigt viele gleichzeitige Verbindungen und lang laufende Prozesse mit wenig Overhead. In diesem Tutorial bauen Sie mit dem Paket autopaho von Eclipse Paho einen Go-Dienst, der sich per TLS mit MQTT 5 verbindet, Telemetrie veröffentlicht, Befehle empfängt und Netzwerkabbrüche dank automatischer Reconnects übersteht.

Eine Go-MQTT-Bibliothek auswählen

Es gibt zwei Eclipse-Paho-Bibliotheken für Go, und die richtige Wahl spart später Zeit.

Die ältere Bibliothek paho.mqtt.golang ist ausgereift und weit verbreitet, spricht aber nur MQTT 3.1.1. Das neuere Modul paho.golang implementiert MQTT 5, und sein Paket autopaho umhüllt den Low-Level-Client mit einem Verbindungsmanagement: Es verbindet sich mit Back-off neu, ruft Sie zurück, sobald die Verbindung steht, damit Sie erneut abonnieren können, und lässt Sie vor dem Veröffentlichen auf die Verbindung warten. Für neue Dienste ist autopaho der bessere Ausgangspunkt. Der Leitfaden zu MQTT-Client-Bibliotheken vergleicht die Optionen in anderen Sprachen.

Projekt einrichten

Sie brauchen Go 1.21 oder neuer.

mkdir go-mqtt-demo && cd go-mqtt-demo
go mod init example.com/go-mqtt-demo
go get github.com/eclipse/paho.golang@latest

Für den ersten Lauf verwenden Sie den öffentlichen MQTT-Broker unter broker.connectmq.com. Er akzeptiert anonyme Verbindungen mit jeder nicht leeren, eindeutigen Client ID. Wählen Sie ein zufälliges Topic-Präfix wie connectmq/demo/x9p4tz und denken Sie daran, dass alles, was dort veröffentlicht wird, für jeden lesbar ist; er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten.

Der vollständige Go-MQTT-Client

Speichern Sie dies als main.go. Der Client liest seine Konfiguration aus Umgebungsvariablen, sodass dasselbe Binary gegen den öffentlichen und einen privaten Broker läuft.

package main

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"log"
	"math/rand"
	"net/url"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/eclipse/paho.golang/autopaho"
	"github.com/eclipse/paho.golang/paho"
)

type Reading struct {
	Device      string    `json:"device"`
	Temperature float64   `json:"temperature"`
	Timestamp   time.Time `json:"ts"`
}

func env(key, def string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return def
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	serverURL, err := url.Parse(env("MQTT_URL", "tls://broker.connectmq.com:8883"))
	if err != nil {
		log.Fatal(err)
	}
	clientID := env("MQTT_CLIENT_ID", "go-demo-x9p4tz")
	prefix := env("MQTT_PREFIX", "connectmq/demo/x9p4tz")
	statusTopic := prefix + "/status"
	commandTopic := prefix + "/cmd/#"
	telemetryTopic := prefix + "/telemetry"

	cfg := autopaho.ClientConfig{
		ServerUrls:                    []*url.URL{serverURL},
		TlsCfg:                        &tls.Config{MinVersion: tls.VersionTLS12},
		KeepAlive:                     30,
		CleanStartOnInitialConnection: false,
		SessionExpiryInterval:         600,
		WillMessage: &paho.WillMessage{
			Topic:   statusTopic,
			Payload: []byte("offline"),
			QoS:     1,
			Retain:  true,
		},
		OnConnectionUp: func(cm *autopaho.ConnectionManager, connAck *paho.Connack) {
			log.Println("connected to broker")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: commandTopic, QoS: 1}},
			}); err != nil {
				log.Printf("subscribe failed: %v", err)
			}
			if _, err := cm.Publish(context.Background(), &paho.Publish{
				Topic: statusTopic, Payload: []byte("online"), QoS: 1, Retain: true,
			}); err != nil {
				log.Printf("status publish failed: %v", err)
			}
		},
		OnConnectError: func(err error) {
			log.Printf("connection attempt failed: %v", err)
		},
		ClientConfig: paho.ClientConfig{
			ClientID: clientID,
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("command on %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) {
				log.Printf("client error: %v", err)
			},
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("server sent disconnect, reason code %d", d.ReasonCode)
			},
		},
	}

	if user := os.Getenv("MQTT_USERNAME"); user != "" {
		cfg.ConnectUsername = user
		cfg.ConnectPassword = []byte(os.Getenv("MQTT_PASSWORD"))
	}

	cm, err := autopaho.NewConnection(ctx, cfg)
	if err != nil {
		log.Fatal(err)
	}
	if err := cm.AwaitConnection(ctx); err != nil {
		log.Fatal(err)
	}

	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	expiry := uint32(60)

	for {
		select {
		case <-ticker.C:
			payload, _ := json.Marshal(Reading{
				Device:      clientID,
				Temperature: 20 + rand.Float64()*5,
				Timestamp:   time.Now().UTC(),
			})
			_, err := cm.Publish(ctx, &paho.Publish{
				Topic:   telemetryTopic,
				QoS:     1,
				Payload: payload,
				Properties: &paho.PublishProperties{
					ContentType:   "application/json",
					MessageExpiry: &expiry,
				},
			})
			if err != nil {
				log.Printf("publish failed: %v", err)
			}
		case <-ctx.Done():
			log.Println("shutting down")
			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(shutdownCtx)
			<-cm.Done()
			return
		}
	}
}

Starten Sie ihn:

go run .

Beobachten Sie in einem zweiten Terminal den Verkehr und senden Sie einen Befehl:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-x9p4tz -t 'connectmq/demo/x9p4tz/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-x9p4tz -t 'connectmq/demo/x9p4tz/cmd/led' -m 'on'

Wie der Code funktioniert

Verbindungsmanagement

autopaho.NewConnection startet eine Hintergrund-Goroutine, die sich verbindet und bei jedem Verbindungsabbruch mit Back-off neu verbindet. AwaitConnection blockiert, bis die erste Verbindung gelingt oder der Kontext endet. Mehr Reconnect-Logik schreiben Sie nicht; manuelle Schleifen sind unnötig.

Erneut abonnieren in OnConnectionUp

OnConnectionUp läuft nach jeder erfolgreichen Verbindung, nicht nur nach der ersten. Das ist der richtige Ort zum Abonnieren. Mit CleanStartOnInitialConnection: false und einem SessionExpiryInterval von 600 Sekunden hält der Broker die Session (Abonnements und gepufferte QoS-1-Nachrichten) nach einem Abbruch zehn Minuten lang vor; ein schneller Reconnect setzt die Session also fort, und das zusätzliche Subscribe schadet nicht. Nach einem längeren Ausfall stellt das Subscribe sie wieder her. Auf dem öffentlichen MQTT-Broker ist die Session-Expiry auf 30 Minuten begrenzt.

Nachrichten empfangen

OnPublishReceived ist ein Slice von Handlern. Jeder gibt zurück, ob er die Nachricht behandelt hat; so können Sie Handler verketten, zum Beispiel einen pro Topic-Familie. Halten Sie Handler schnell und übergeben Sie aufwendige Arbeit an einen Channel oder Worker-Pool, denn sie laufen im Empfangspfad des Clients.

MQTT-5-Properties

Der Telemetrie-Publish setzt zwei MQTT-5-Properties. ContentType teilt Abnehmern mit, dass der Payload JSON ist. MessageExpiry von 60 Sekunden weist den Broker an, die Nachricht zu verwerfen, wenn sie nicht innerhalb einer Minute zugestellt werden kann, genau das Richtige für eine Temperatur, die ohnehin bald überholt ist. Der Leitfaden zu den MQTT-5-Funktionen erklärt die übrigen Properties.

Last Will und Status

Die Will-Nachricht veröffentlicht ein retained offline auf dem Status-Topic, falls der Dienst ohne sauberes Trennen verschwindet. Bei jeder Verbindung veröffentlicht der Dienst ein retained online. Jedes Dashboard kann das Status-Topic abonnieren und weiß immer, ob der Dienst läuft.

Umstieg auf einen privaten Broker

Für den Produktivbetrieb legen Sie im ConnectMQ-Portal einen Broker an und verwenden die generierten Zugangsdaten. Client IDs sind an die ausgestellten Zugangsdaten gebunden; übergeben Sie also genau diese ID:

export MQTT_URL=tls://broker.connectmq.com:8883
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=acme/site1/gateway1
go run .

In der Industrie richten Sie das Präfix am besten an der ISA-95-Hierarchie Ihres Unified Namespace (UNS) aus, etwa acme/werk-stuttgart/montage/linie-1/gateway-1. Ein Go-Dienst, der Maschinendaten für MES oder Historian aufnimmt, abonniert dann mit einer Wildcard eine ganze Linie oder ein ganzes Werk.

Das Zertifikat des Brokers stammt von Let's Encrypt, dem Go über den Root-Store des Systems vertraut; TlsCfg braucht daher nur eine Mindestversion. Stellen Sie in einem minimalen Container-Image wie scratch oder distroless/static sicher, dass CA-Zertifikate vorhanden sind, sonst scheitert der Handshake mit x509: certificate signed by unknown authority.

Produktions-Checkliste für einen Go-MQTT-Client

1. Eine Client ID pro Prozess. Zwei Replikas mit derselben ID trennen sich gegenseitig ständig. Geben Sie jeder Replika eigene Zugangsdaten oder verwenden Sie für lastverteilte Abnehmer MQTT-5-Shared-Subscriptions auf Tarifen, die sie unterstützen.
2. QoS 1 für wichtige Daten verwenden. Der Free-Tarif unterstützt QoS 0 und 1; QoS 2 ist in kostenpflichtigen Tarifen verfügbar. Der Leitfaden zu den QoS-Stufen erklärt die Abwägungen.
3. Publish-Fehler behandeln. Publish gibt sofort einen Fehler zurück, wenn die Verbindung unterbrochen ist. Entscheiden Sie, ob Sie verwerfen, puffern oder erneut versuchen. Neuere autopaho-Versionen bieten außerdem ein Publish mit Warteschlange zum Puffern bei Ausfällen.
4. Sauber herunterfahren. Disconnect sendet ein normales DISCONNECT, sodass der Will nicht ausgelöst wird und Subscriber stattdessen Ihren expliziten Status sehen.
5. Reason Codes protokollieren. MQTT-5-Broker erklären Verbindungsabbrüche mit Reason Codes; 0x97 bedeutet zum Beispiel, dass ein Kontingent überschritten wurde. OnServerDisconnect liefert Ihnen diesen Code.
6. Geheimnisse aus dem Code heraushalten. Lesen Sie Zugangsdaten aus der Umgebung oder einem Secret Manager, niemals aus dem Quellcode.

Fehlersuche

Die Referenz der Paho-Bibliothek finden Sie in der paho.golang-Dokumentation.

Andere Backend-Sprachen

Dieselben Muster, also eine stabile Client ID, erneutes Abonnieren beim Verbinden, Last Will und TLS, gelten überall. Vergleichen Sie mit dem Rust-Tutorial zu rumqttc und dem Java-MQTT-Client-Tutorial, wenn Ihr Stack gemischt ist.

Betreiben Sie Ihre Go-Dienste mit ConnectMQ

Mit einem verwalteten MQTT-Broker können sich Ihre Go-Dienste auf die Geschäftslogik konzentrieren, während sich jemand anderes um Broker, Zertifikate und Upgrades kümmert. ConnectMQ unterstützt MQTT 5 mit Session-Expiry, Message-Expiry und Reason Codes, sodass alles in diesem Tutorial unverändert funktioniert. Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com, ganz ohne Kreditkarte, und richten Sie MQTT_URL darauf aus.

Häufig gestellte Fragen

Welche Go-MQTT-Bibliothek sollte ich verwenden?

Für neue Projekte github.com/eclipse/paho.golang mit dem Paket autopaho, weil es MQTT 5 unterstützt und Reconnects verwaltet. Verwenden Sie paho.mqtt.golang, wenn Sie MQTT-3.1.1-Kompatibilität mit bestehendem Code brauchen.

Abonniert autopaho nach einem Reconnect automatisch neu?

Es ruft OnConnectionUp nach jeder Verbindung auf. Abonnieren Sie dort und kombinieren Sie das mit einem Session-Expiry-Intervall, damit der Broker das Abonnement auch über kurze Ausfälle hinweg vorhalten kann.

Wie verbinde ich einen Go-MQTT-Client per TLS?

Verwenden Sie eine Server-URL mit tls:// und Port 8883 und setzen Sie TlsCfg. Bei Brokern mit öffentlich vertrauenswürdigen Zertifikaten reicht der Root-Store des Systems.

Kann ich mehrere Instanzen desselben Go-Dienstes betreiben?

Ja, aber jede Instanz braucht eine eigene Client ID. Für die lastverteilte Verarbeitung derselben Topics verwenden Sie MQTT-5-Shared-Subscriptions auf einem Tarif, der sie unterstützt.