Tutorial de cliente MQTT en Go con Eclipse Paho autopaho y MQTT 5

Un cliente MQTT en Go encaja de forma natural en servicios backend que ingieren datos de dispositivos, porque Go gestiona muchas conexiones concurrentes y procesos de larga duración con poca sobrecarga. En este tutorial construirás un servicio en Go con el paquete autopaho de Eclipse Paho que se conecta por TLS usando MQTT 5, publica telemetría, recibe comandos y sobrevive a las caídas de red con reconexiones automáticas.

Cómo elegir una biblioteca MQTT para Go

Hay dos bibliotecas Eclipse Paho para Go, y elegir la adecuada ahorra tiempo más adelante.

La biblioteca más antigua, paho.mqtt.golang, es madura y muy utilizada, pero solo habla MQTT 3.1.1. El módulo más reciente, paho.golang, implementa MQTT 5, y su paquete autopaho envuelve el cliente de bajo nivel con gestión de la conexión: se reconecta con back-off, te avisa mediante un callback cuando la conexión está activa para que puedas volver a suscribirte y te permite esperar a la conexión antes de publicar. Para servicios nuevos, autopaho es el mejor punto de partida. La guía de bibliotecas cliente MQTT compara opciones en otros lenguajes.

Preparación del proyecto

Necesitas Go 1.21 o posterior.

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

Para la primera ejecución, usa el servidor MQTT público gratuito en broker.connectmq.com. Acepta conexiones anónimas con cualquier client ID no vacío y único. Elige un prefijo de topic aleatorio como connectmq/demo/x9p4tz y recuerda que cualquiera puede leer lo que se publica allí, así que úsalo solo para pruebas y prototipos, nunca con datos de producción.

El cliente MQTT en Go completo

Guárdalo como main.go. Lee su configuración de variables de entorno, de modo que el mismo binario funciona contra el servidor público y contra un broker privado.

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

Ejecútalo:

go run .

En un segundo terminal, observa el tráfico y envía un comando:

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'

Cómo funciona el código

Gestión de la conexión

autopaho.NewConnection inicia una goroutine en segundo plano que se conecta y sigue reconectándose con back-off cada vez que la conexión se cae. AwaitConnection se bloquea hasta que la primera conexión tiene éxito o el contexto termina. Esa es toda la lógica de reconexión que escribes; no hacen falta bucles manuales.

Volver a suscribirse en OnConnectionUp

OnConnectionUp se ejecuta tras cada conexión correcta, no solo la primera. Es el lugar adecuado para suscribirse. Con CleanStartOnInitialConnection: false y un SessionExpiryInterval de 600 segundos, el broker conserva la sesión (suscripciones y mensajes QoS 1 en cola) durante diez minutos tras una caída, de modo que una reconexión rápida reanuda la sesión y la suscripción adicional no causa ningún daño. Tras una interrupción más larga, la suscripción la vuelve a crear. En el servidor público, la expiración de la sesión está limitada a 30 minutos.

Recepción de mensajes

OnPublishReceived es un slice de handlers. Cada uno devuelve si ha gestionado el mensaje; esto te permite encadenar handlers, por ejemplo uno por familia de topics. Mantén los handlers rápidos y delega el trabajo pesado en un canal o un pool de workers, porque se ejecutan en la ruta de recepción del cliente.

Propiedades de MQTT 5

La publicación de telemetría establece dos propiedades de MQTT 5. ContentType indica a los consumidores que el payload es JSON. MessageExpiry de 60 segundos indica al broker que descarte el mensaje si no puede entregarse en un minuto, que es exactamente lo que quieres para una temperatura que de todos modos quedará obsoleta. La guía de funciones de MQTT 5 explica las demás propiedades.

Last Will y estado

El mensaje Will publica un offline retenido en el topic de estado si el servicio desaparece sin una desconexión limpia. En cada conexión, el servicio publica un online retenido. Cualquier panel puede suscribirse al topic de estado y saber siempre si el servicio está vivo.

Paso a un broker privado

Para producción, crea un broker en el portal de ConnectMQ y usa las credenciales generadas. Los client ID están vinculados a las credenciales emitidas, así que pasa exactamente el que corresponde:

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 .

El certificado del broker lo emite Let's Encrypt, en el que Go confía a través del almacén de raíces del sistema, así que TlsCfg solo necesita una versión mínima. En una imagen de contenedor mínima como scratch o distroless/static, asegúrate de que los certificados de CA están presentes; de lo contrario, el handshake falla con x509: certificate signed by unknown authority.

Lista de comprobación de producción para un cliente MQTT en Go

1. Un client ID por proceso. Dos réplicas con el mismo ID se desconectarán mutuamente sin parar. Da a cada réplica sus propias credenciales o usa suscripciones compartidas de MQTT 5 para consumidores con balanceo de carga en los planes que las admiten.
2. Usa QoS 1 para los datos que te importan. El plan Free admite QoS 0 y 1; QoS 2 está disponible en los planes de pago. La guía de niveles de QoS explica las ventajas e inconvenientes.
3. Gestiona los errores de publicación. Publish devuelve un error inmediatamente cuando la conexión está caída. Decide si descartar, almacenar en búfer o reintentar. Las versiones recientes de autopaho también ofrecen una publicación respaldada por cola para almacenar en búfer durante las interrupciones.
4. Apaga de forma limpia. Llamar a Disconnect envía un DISCONNECT normal, de modo que el Will no se activa y los suscriptores ven tu estado explícito.
5. Registra los reason codes. Los brokers MQTT 5 explican las desconexiones con reason codes; por ejemplo, 0x97 significa que se ha superado una cuota. OnServerDisconnect te da ese código.
6. Mantén los secretos fuera del código. Lee las credenciales del entorno o de un gestor de secretos, nunca del árbol de código fuente.

Solución de problemas

Para la referencia de la biblioteca Paho, consulta la documentación de paho.golang.

Otros lenguajes de backend

Los mismos patrones (un client ID estable, volver a suscribirse al conectar, Last Will y TLS) se aplican en todas partes. Compáralos con el tutorial de Rust con rumqttc y el tutorial de cliente MQTT en Java si tu stack es mixto.

Ejecuta tus servicios Go en ConnectMQ

Un broker MQTT gestionado permite que tus servicios Go se centren en la lógica de negocio mientras otro se ocupa del broker, los certificados y las actualizaciones. ConnectMQ admite MQTT 5 con expiración de sesión, expiración de mensajes y reason codes, así que todo lo de este tutorial funciona sin cambios. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y apunta a él MQTT_URL.

Preguntas frecuentes

¿Qué biblioteca MQTT para Go debo usar?

Para proyectos nuevos, github.com/eclipse/paho.golang con el paquete autopaho, porque admite MQTT 5 y gestiona las reconexiones. Usa paho.mqtt.golang si necesitas compatibilidad con MQTT 3.1.1 en código existente.

¿autopaho vuelve a suscribirse automáticamente tras una reconexión?

Llama a OnConnectionUp tras cada conexión. Suscríbete allí y combínalo con un intervalo de expiración de sesión para que el broker también pueda conservar la suscripción durante interrupciones breves.

¿Cómo conecto un cliente MQTT en Go por TLS?

Usa una URL de servidor tls:// con el puerto 8883 y establece TlsCfg. Con brokers que tienen certificados de confianza pública, basta con el almacén de raíces del sistema.

¿Puedo ejecutar varias instancias del mismo servicio Go?

Sí, pero cada instancia necesita su propio client ID. Para consumir los mismos topics con balanceo de carga, usa suscripciones compartidas de MQTT 5 en un plan que las admita.