Tutorial de Cliente MQTT em Go com Eclipse Paho autopaho e MQTT 5

Um cliente MQTT em Go se encaixa naturalmente em serviços de backend que ingerem dados de dispositivos, porque o Go lida com muitas conexões simultâneas e processos de longa duração com pouca sobrecarga. Neste tutorial você vai criar um serviço em Go com o pacote autopaho do Eclipse Paho que se conecta via TLS usando MQTT 5, publica telemetria, recebe comandos e sobrevive a quedas de rede com reconexão automática.

Escolhendo uma biblioteca MQTT para Go

Existem duas bibliotecas Eclipse Paho para Go, e escolher a certa poupa tempo mais adiante.

A biblioteca mais antiga, paho.mqtt.golang, é madura e amplamente usada, mas só fala MQTT 3.1.1. O módulo mais novo, paho.golang, implementa o MQTT 5, e seu pacote autopaho envolve o cliente de baixo nível com gerenciamento de conexão: ele reconecta com back-off, chama você de volta quando a conexão está ativa para que possa assinar novamente e permite aguardar a conexão antes de publicar. Para serviços novos, o autopaho é o melhor ponto de partida. O guia de bibliotecas de cliente MQTT compara opções em outras linguagens.

Configuração do projeto

Você precisa do Go 1.21 ou mais recente.

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

Na primeira execução, use o servidor MQTT público (gratuito) em broker.connectmq.com, que serve para testes e protótipos, não para dados de produção. Ele aceita conexões anônimas com qualquer client ID não vazio e único. Escolha um prefixo de topic aleatório como connectmq/demo/x9p4tz e lembre-se de que tudo o que for publicado ali pode ser lido por qualquer pessoa.

O cliente MQTT completo em Go

Salve isto como main.go. Ele lê a configuração de variáveis de ambiente, para que o mesmo binário rode tanto no servidor MQTT público quanto em um 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
		}
	}
}

Execute:

go run .

Em um segundo terminal, acompanhe o tráfego e envie um 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'

Como o código funciona

Gerenciamento de conexão

autopaho.NewConnection inicia uma goroutine em segundo plano que se conecta e continua reconectando com back-off sempre que a conexão cai. AwaitConnection bloqueia até que a primeira conexão tenha sucesso ou o contexto termine. Essa é toda a lógica de reconexão que você escreve; nenhum loop manual é necessário.

Assinando novamente em OnConnectionUp

O OnConnectionUp roda após cada conexão bem-sucedida, não apenas a primeira. Esse é o lugar certo para assinar. Com CleanStartOnInitialConnection: false e um SessionExpiryInterval de 600 segundos, o broker mantém a sessão (assinaturas e mensagens QoS 1 enfileiradas) por dez minutos após uma queda, então uma reconexão rápida retoma a sessão e a assinatura extra é inofensiva. Depois de uma interrupção mais longa, a assinatura a recria. No servidor MQTT público, a expiração de sessão é limitada a 30 minutos.

Recebendo mensagens

OnPublishReceived é uma slice de handlers. Cada um retorna se tratou a mensagem; isso permite encadear handlers, por exemplo um por família de topics. Mantenha os handlers rápidos e passe o trabalho pesado para um channel ou um pool de workers, porque eles rodam no caminho de recebimento do cliente.

Propriedades do MQTT 5

A publicação de telemetria define duas propriedades do MQTT 5. ContentType informa aos consumidores que o payload é JSON. MessageExpiry de 60 segundos diz ao broker para descartar a mensagem se ela não puder ser entregue em um minuto, que é exatamente o que você quer para uma temperatura que de qualquer forma será substituída. O guia de recursos do MQTT 5 explica as demais propriedades.

Last Will e status

A mensagem Will publica um offline retido no topic de status se o serviço sumir sem uma desconexão limpa. A cada conexão, o serviço publica um online retido. Qualquer dashboard pode assinar o topic de status e saber sempre se o serviço está ativo.

Migrando para um broker privado

Para produção, crie um broker no portal da ConnectMQ e use as credenciais geradas. Os client IDs são vinculados às credenciais emitidas, então passe exatamente o ID fornecido:

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 .

O certificado do broker é emitido pela Let's Encrypt, na qual o Go confia por meio do repositório raiz do sistema, então o TlsCfg só precisa de uma versão mínima. Em uma imagem de contêiner mínima como scratch ou distroless/static, garanta que os certificados de CA estejam presentes; caso contrário, o handshake falha com x509: certificate signed by unknown authority.

Checklist de produção para um cliente MQTT em Go

1. Um client ID por processo. Duas réplicas com o mesmo ID ficarão desconectando uma à outra. Dê a cada réplica credenciais próprias ou use shared subscriptions do MQTT 5 para consumidores com balanceamento de carga nos planos que oferecem suporte.
2. Use QoS 1 para dados importantes. O plano Free suporta QoS 0 e 1; o QoS 2 está disponível nos planos pagos. O guia de níveis de QoS explica os trade-offs.
3. Trate erros de publicação. Publish retorna um erro imediatamente quando a conexão está fora. Decida se vai descartar, armazenar em buffer ou tentar de novo. Versões recentes do autopaho também oferecem uma publicação baseada em fila para armazenar mensagens durante interrupções.
4. Encerre de forma limpa. Chamar Disconnect envia um DISCONNECT normal, então a Will não é disparada e os assinantes veem o seu status explícito.
5. Registre os reason codes. Brokers MQTT 5 explicam desconexões com reason codes; por exemplo, 0x97 significa que uma cota foi excedida. O OnServerDisconnect entrega esse código.
6. Mantenha segredos fora do código. Leia as credenciais do ambiente ou de um gerenciador de segredos, nunca da árvore de código-fonte.

Solução de problemas

Para a referência da biblioteca Paho, veja a documentação do paho.golang.

Outras linguagens de backend

Os mesmos padrões (client ID estável, nova assinatura ao conectar, Last Will e TLS) se aplicam em todo lugar. Compare com o tutorial de Rust com rumqttc e o tutorial de cliente MQTT em Java se a sua stack for mista.

Rode seus serviços Go na ConnectMQ

Um broker MQTT gerenciado permite que seus serviços Go se concentrem na lógica de negócio enquanto outra pessoa cuida do broker, dos certificados e das atualizações. A ConnectMQ suporta MQTT 5 com expiração de sessão, expiração de mensagem e reason codes, então tudo neste tutorial funciona sem alterações. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e aponte o MQTT_URL para ele.

Perguntas frequentes

Qual biblioteca MQTT para Go devo usar?

Para projetos novos, github.com/eclipse/paho.golang com o pacote autopaho, porque ele suporta MQTT 5 e gerencia as reconexões. Use paho.mqtt.golang se precisar de compatibilidade com MQTT 3.1.1 em código existente.

O autopaho assina novamente de forma automática após uma reconexão?

Ele chama OnConnectionUp após cada conexão. Faça a assinatura ali e combine isso com um intervalo de expiração de sessão para que o broker também possa manter a assinatura durante interrupções curtas.

Como conecto um cliente MQTT em Go via TLS?

Use uma URL de servidor tls:// com a porta 8883 e defina TlsCfg. Para brokers com certificados de confiança pública, o repositório raiz do sistema é suficiente.

Posso rodar várias instâncias do mesmo serviço Go?

Sim, mas cada instância precisa do próprio client ID. Para consumo com balanceamento de carga dos mesmos topics, use shared subscriptions do MQTT 5 em um plano que ofereça suporte.