MQTT 5 in Go mit autopaho: Tutorial für einen Client mit Auto-Reconnect

autopaho ist der Verbindungsmanager in github.com/eclipse/paho.golang, dem MQTT-5-Client von Eclipse Paho für Go. Du übergibst eine Liste von Broker-URLs und ein paar Callbacks; autopaho verbindet sich, verbindet sich mit Backoff neu, wenn das Netz wegbricht, und ruft dich bei jeder (Wieder-)Herstellung der Verbindung zurück. Dieses Tutorial baut damit einen vollständigen Subscriber und Publisher, ergänzt Zugangsdaten und TLS und zeigt, wie du das Reconnect testest.

Wenn du nur MQTT 3.1.1 brauchst, ist der ältere paho.mqtt.golang-Client weiterhin in Ordnung. Nimm autopaho, wenn du MQTT-5-Features — Reason Codes, Session Expiry, User Properties, Message Expiry — zusammen mit automatischem Reconnect willst.

paho vs. autopaho

Das Modul hat zwei Schichten, und es hilft zu wissen, welche du gerade aufrufst:

In der Praxis konfigurierst du autopaho.ClientConfig, das eine paho.ClientConfig für Client-ID und Nachrichten-Callbacks einbettet, und baust die Request-Pakete aus dem Paket paho.

Installation

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

Beide Pakete stammen aus demselben Modul, ein go get reicht also. Nutze eine aktuelle Go-Version; die Bibliothek folgt den aktuellen Releases.

Ein vollständiger Client

Dieses Programm abonniert ein Topic, veröffentlicht jede Sekunde eine Nachricht und beendet sich mit Strg+C sauber. Es läuft gegen den öffentlichen Test-Endpoint broker.connectmq.com; ersetze den Host durch deinen eigenen Broker, sobald du so weit bist.

package main

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"os"
	"os/signal"
	"syscall"
	"time"

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

const topic = "testmqtt/go-autopaho/demo"

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

	u, err := url.Parse("mqtt://broker.connectmq.com:1883")
	if err != nil {
		log.Fatal(err)
	}

	cfg := autopaho.ClientConfig{
		ServerUrls:                    []*url.URL{u},
		KeepAlive:                     30,
		CleanStartOnInitialConnection: true,
		SessionExpiryInterval:         0,
		OnConnectionUp: func(cm *autopaho.ConnectionManager, ack *paho.Connack) {
			log.Println("verbunden, abonniere")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("Subscribe fehlgeschlagen: %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("Verbindungsfehler: %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("empfangen %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("Client-Fehler: %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("Server-Disconnect, Reason Code %d", d.ReasonCode)
			},
		},
	}

	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(time.Second)
	defer ticker.Stop()
	for i := 0; ; i++ {
		select {
		case <-ticker.C:
			_, err := cm.Publish(ctx, &paho.Publish{
				Topic:   topic,
				QoS:     1,
				Payload: []byte(fmt.Sprintf("message %d", i)),
			})
			if err != nil {
				log.Printf("Publish fehlgeschlagen: %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

Starte es mit go run . — für jede veröffentlichte Nachricht solltest du eine Empfangszeile sehen. Um denselben Traffic von außen zu beobachten, öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und abonniere testmqtt/go-autopaho/#.

Ein paar Details sind wichtig:

• NewConnection kehrt sofort zurück. Der Verbindungsaufbau läuft im Hintergrund; AwaitConnection blockiert, bis die erste Verbindung steht oder der Context abgebrochen wird.
• Abonniere in OnConnectionUp. Dieser Callback feuert bei der ersten Verbindung und bei jedem Reconnect, deine Subscriptions kommen nach einem Netzabbruch also automatisch zurück.
• OnPublishReceived ist ein Slice. Jeder Handler gibt (handled bool, err error) zurück; mehrere Handler können dieselbe Nachricht prüfen, etwa einer pro Topic-Präfix.
• Publish mit QoS 1 wartet auf das PUBACK und liefert eine *paho.PublishResponse mit dem Reason Code des Brokers — so meldet MQTT 5 Probleme wie "not authorized" für ein einzelnes Publish.

Zugangsdaten und TLS

Aktuelle autopaho-Versionen nehmen die Zugangsdaten direkt in der Konfiguration entgegen. Für TLS stellst du das URL-Schema auf mqtts:// und den Port auf 8883 um; die Go-Standardbibliothek prüft das Zertifikat gegen die System-Roots.

u, _ := url.Parse("mqtts://broker.connectmq.com:8883")

cfg := autopaho.ClientConfig{
	ServerUrls:      []*url.URL{u},
	KeepAlive:       30,
	ConnectUsername: os.Getenv("MQTT_USER"),
	ConnectPassword: []byte(os.Getenv("MQTT_PASS")),
	// Nur für eine private CA oder Client-Zertifikate nötig:
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls ist eine Liste: autopaho probiert die URLs der Reihe nach, was einen einfachen Failover zwischen zwei Endpoints ergibt. ws://- und wss://-URLs (zum Beispiel wss://broker.connectmq.com:8084/mqtt) werden ebenfalls akzeptiert, wenn TCP-Ports gesperrt sind. Schlägt der Handshake fehl, zeigt der Guide zum TLS-Verbindungstest, wie du die Zertifikatskette mit openssl s_client prüfst.

Persistente Sessions statt Neu-Abonnieren

In OnConnectionUp zu abonnieren ist einfach und robust. Die Alternative: Der Broker merkt sich deinen Client.

cfg.CleanStartOnInitialConnection = false
cfg.SessionExpiryInterval = 3600 // Sekunden

Der geteilte Test-Endpoint speichert keine Offline-Sessions; teste diesen Teil daher gegen deine eigene private Broker-Instanz.

Mit fester Client-ID behält der Broker Subscriptions und gepufferte QoS-1/2-Nachrichten bis zu eine Stunde nach einem Disconnect. Das Feld ack.SessionPresent in OnConnectionUp verrät, ob die Session überlebt hat — dann kannst du das Subscribe überspringen. Session Expiry ist ein MQTT-5-Feature; wie Broker damit umgehen, steht unter persistente Sessions.

Für ausgehende Nachrichten, die einen Disconnect überleben müssen, bietet autopaho außerdem PublishViaQueue: Nachrichten landen in einer Queue (standardmäßig im Speicher oder mit dem dateibasierten Queue-Paket auf der Platte) und werden gesendet, sobald die Verbindung wieder steht.

Reconnect-Verhalten

autopaho wiederholt fehlgeschlagene Verbindungen mit einer Pause zwischen den Versuchen. Neuere Releases bieten dafür ReconnectBackoff in der Konfiguration (ältere nutzten ein festes ConnectRetryDelay); der Standard passt für die meisten Clients. Bei einer großen Flotte solltest du verhindern, dass alle Clients nach einem Broker-Neustart im selben Moment reconnecten — füge Jitter oder wachsenden Backoff hinzu, wie in Reconnect Best Practices beschrieben.

Fehlgeschlagene Versuche landen in OnConnectError. Typische Meldungen und ihre Bedeutung:

Jeden numerischen Code schlägst du in der MQTT-Reason-Code-Referenz nach.

Reconnect testen

Warte nicht auf die Produktion, um herauszufinden, ob das Neu-Abonnieren funktioniert. Ein schneller manueller Check:

1. Programm starten und prüfen, dass Nachrichten fließen.
2. Netz länger als das Keep-Alive kappen (WLAN aus, oder sudo ifconfig en0 down / sudo ip link set eth0 down).
3. Zusehen, wie OnConnectError die fehlgeschlagenen Versuche protokolliert.
4. Netz wiederherstellen. OnConnectionUp sollte "verbunden, abonniere" loggen und die Empfangszeilen sollten weiterlaufen.

Für automatisierte Tests packst du den Publish/Receive-Roundtrip in go test mit Timeout — MQTT-Clients in Go testen enthält einen fertigen autopaho-Helper.

Deinen autopaho-Client im Browser beobachten

autopaho gibt Go-Programmen mit sehr wenig Code einen produktionstauglichen MQTT-5-Client: URLs und Callbacks konfigurieren, in OnConnectionUp abonnieren und die zurückgegebenen Reason Codes prüfen. Teste einmal einen Netzabbruch, halte die Client-IDs eindeutig, und du hast einen Client, der sich selbst erholt.

Häufig gestellte Fragen