Tutoriel client MQTT en Go avec Eclipse Paho autopaho et MQTT 5

Un client MQTT en Go convient naturellement aux services backend qui ingèrent des données d'appareils, car Go gère de nombreuses connexions concurrentes et des processus de longue durée avec peu de surcoût. Dans ce tutoriel, vous allez construire un service Go avec le package autopaho d'Eclipse Paho qui se connecte en TLS avec MQTT 5, publie de la télémétrie, reçoit des commandes et survit aux coupures réseau grâce aux reconnexions automatiques.

Choisir une bibliothèque MQTT pour Go

Il existe deux bibliothèques Eclipse Paho pour Go, et choisir la bonne fait gagner du temps par la suite.

L'ancienne bibliothèque paho.mqtt.golang est mature et largement utilisée, mais elle ne parle que MQTT 3.1.1. Le module plus récent paho.golang implémente MQTT 5, et son package autopaho enveloppe le client bas niveau avec une gestion de connexion : il se reconnecte avec un délai croissant, vous rappelle quand la connexion est établie pour que vous puissiez vous réabonner, et vous permet d'attendre la connexion avant de publier. Pour les nouveaux services, autopaho est le meilleur point de départ. Le guide des bibliothèques clientes MQTT compare les options dans d'autres langages.

Préparer le projet

Il vous faut Go 1.21 ou plus récent.

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

Pour le premier essai, utilisez le serveur MQTT public gratuit broker.connectmq.com. Il accepte les connexions anonymes avec n'importe quel client ID unique et non vide. Choisissez un préfixe de topic aléatoire comme connectmq/demo/x9p4tz, et rappelez-vous que tout ce qui y est publié est lisible par tous.

Le client MQTT Go complet

Enregistrez ceci sous main.go. Il lit sa configuration dans des variables d'environnement, si bien que le même binaire fonctionne avec le serveur MQTT public gratuit et avec un broker privé.

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

Lancez-le :

go run .

Dans un second terminal, observez le trafic et envoyez une commande :

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'

Comment fonctionne le code

Gestion de la connexion

autopaho.NewConnection lance une goroutine en arrière-plan qui se connecte, puis se reconnecte avec un délai croissant à chaque coupure. AwaitConnection bloque jusqu'à ce que la première connexion réussisse ou que le contexte se termine. C'est toute la logique de reconnexion que vous avez à écrire ; aucune boucle manuelle n'est nécessaire.

Se réabonner dans OnConnectionUp

OnConnectionUp s'exécute après chaque connexion réussie, pas seulement la première. C'est le bon endroit pour s'abonner. Avec CleanStartOnInitialConnection: false et un SessionExpiryInterval de 600 secondes, le broker conserve la session (abonnements et messages QoS 1 en file d'attente) pendant dix minutes après une coupure ; une reconnexion rapide reprend donc la session et l'abonnement supplémentaire est sans conséquence. Après une coupure plus longue, l'abonnement la recrée. Sur le serveur MQTT public gratuit, l'expiration de session est plafonnée à 30 minutes.

Recevoir des messages

OnPublishReceived est une liste de handlers. Chacun indique s'il a traité le message, ce qui permet de les chaîner, par exemple un par famille de topics. Gardez les handlers rapides et confiez les traitements lourds à un channel ou à un pool de workers, car ils s'exécutent sur le chemin de réception du client.

Propriétés MQTT 5

La publication de télémétrie définit deux propriétés MQTT 5. ContentType indique aux consommateurs que le payload est du JSON. MessageExpiry de 60 secondes indique au broker de supprimer le message s'il ne peut pas être livré dans la minute, ce qui est exactement ce que vous voulez pour une température qui sera de toute façon remplacée. Le guide des fonctionnalités de MQTT 5 explique les autres propriétés.

Last Will et statut

Le message Will publie un offline retenu sur le topic de statut si le service disparaît sans déconnexion propre. À chaque connexion, le service publie un online retenu. N'importe quel tableau de bord peut s'abonner au topic de statut et savoir à tout moment si le service est actif.

Passer à un broker privé

Pour la production, créez un broker dans le portail ConnectMQ et utilisez les identifiants générés. Les client IDs sont liés aux identifiants émis ; transmettez donc exactement celui fourni :

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 .

Le certificat du broker est émis par Let's Encrypt, auquel Go fait confiance via le magasin de racines du système ; TlsCfg n'a donc besoin que d'une version minimale. Dans une image de conteneur minimale comme scratch ou distroless/static, assurez-vous que les certificats d'autorités de certification sont présents, sinon la négociation échoue avec x509: certificate signed by unknown authority.

Liste de contrôle production pour un client MQTT Go

1. Un client ID par processus. Deux réplicas avec le même identifiant se déconnecteront mutuellement en boucle. Donnez à chaque réplica ses propres identifiants, ou utilisez les abonnements partagés MQTT 5 pour des consommateurs en répartition de charge sur les plans qui les prennent en charge.
2. Utilisez la QoS 1 pour les données importantes. Le plan Free prend en charge les QoS 0 et 1 ; la QoS 2 est disponible sur les plans payants. Le guide des niveaux de QoS explique les compromis.
3. Gérez les erreurs de publication. Publish renvoie immédiatement une erreur lorsque la connexion est coupée. Décidez s'il faut abandonner, mettre en tampon ou réessayer. Les versions récentes d'autopaho proposent aussi une publication adossée à une file d'attente pour mettre en tampon pendant les coupures.
4. Arrêtez proprement. Appeler Disconnect envoie un DISCONNECT normal ; le Will n'est donc pas déclenché et les abonnés voient votre statut explicite.
5. Journalisez les reason codes. Les brokers MQTT 5 expliquent les déconnexions par des reason codes ; par exemple, 0x97 signifie qu'un quota a été dépassé. OnServerDisconnect vous fournit ce code.
6. Gardez les secrets hors du code. Lisez les identifiants depuis l'environnement ou un gestionnaire de secrets, jamais depuis l'arborescence des sources.

Dépannage

Pour la référence de la bibliothèque Paho, consultez la documentation de paho.golang.

Autres langages backend

Les mêmes principes, un client ID stable, le réabonnement à la connexion, le Last Will et TLS, s'appliquent partout. Comparez avec le tutoriel Rust rumqttc et le tutoriel client MQTT en Java si votre stack est mixte.

Faites tourner vos services Go sur ConnectMQ

Un broker MQTT managé permet à vos services Go de se concentrer sur la logique métier pendant que quelqu'un d'autre gère le broker, les certificats et les mises à jour. ConnectMQ prend en charge MQTT 5 avec l'expiration de session, l'expiration des messages et les reason codes ; tout ce tutoriel fonctionne donc sans modification. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et faites pointer MQTT_URL vers lui.

Questions fréquentes

Quelle bibliothèque MQTT utiliser en Go ?

Pour les nouveaux projets, github.com/eclipse/paho.golang avec le package autopaho, car il prend en charge MQTT 5 et gère les reconnexions. Utilisez paho.mqtt.golang si vous avez besoin de la compatibilité MQTT 3.1.1 avec du code existant.

autopaho se réabonne-t-il automatiquement après une reconnexion ?

Il appelle OnConnectionUp après chaque connexion. Abonnez-vous à cet endroit, et combinez cela avec un intervalle d'expiration de session pour que le broker puisse aussi conserver l'abonnement pendant les coupures courtes.

Comment connecter un client MQTT Go en TLS ?

Utilisez une URL de serveur tls:// avec le port 8883 et définissez TlsCfg. Pour les brokers dotés de certificats publiquement reconnus, le magasin de racines du système suffit.

Puis-je faire tourner plusieurs instances du même service Go ?

Oui, mais chaque instance a besoin de son propre client ID. Pour consommer les mêmes topics en répartition de charge, utilisez les abonnements partagés MQTT 5 sur un plan qui les prend en charge.