Eclipse Paho autopaho ve MQTT 5 ile Go MQTT Client Eğitimi

Go, çok sayıda eşzamanlı bağlantıyı ve uzun süre çalışan süreçleri düşük ek yükle yönettiği için, cihaz verisi toplayan backend servisleri için bir Go MQTT client doğal bir seçimdir. Bu eğitimde, Eclipse Paho'nun autopaho paketiyle MQTT 5 kullanarak TLS üzerinden bağlanan, telemetri yayınlayan, komut alan ve otomatik yeniden bağlanma sayesinde ağ kesintilerini atlatan bir Go servisi oluşturacaksınız.

Go MQTT Kütüphanesi Seçimi

Go için iki Eclipse Paho kütüphanesi vardır ve doğru olanı seçmek ileride zaman kazandırır.

Eski paho.mqtt.golang kütüphanesi olgun ve yaygın olarak kullanılır, ancak yalnızca MQTT 3.1.1 konuşur. Daha yeni paho.golang modülü MQTT 5'i uygular; autopaho paketi de alt seviye client'ı bağlantı yönetimiyle sarmalar: Geri çekilme (back-off) ile yeniden bağlanır, yeniden abone olabilmeniz için bağlantı kurulduğunda sizi geri çağırır ve yayın yapmadan önce bağlantıyı beklemenize olanak tanır. Yeni servisler için autopaho daha iyi bir başlangıç noktasıdır. MQTT client kütüphaneleri rehberi diğer dillerdeki seçenekleri karşılaştırır.

Proje Kurulumu

Go 1.21 veya daha yeni bir sürüm gerekir.

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

İlk çalıştırma için broker.connectmq.com adresindeki herkese açık MQTT test broker'ını kullanın. Boş olmayan, benzersiz herhangi bir client ID ile anonim bağlantıları kabul eder. connectmq/demo/x9p4tz gibi rastgele bir topic prefix'i seçin ve oraya yayınlanan her şeyin herkes tarafından okunabildiğini unutmayın.

Eksiksiz Go MQTT Client

Bunu main.go olarak kaydedin. Yapılandırmasını ortam değişkenlerinden okur; böylece aynı binary hem herkese açık broker'a hem de özel bir broker'a karşı çalışır.

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

Çalıştırın:

go run .

İkinci bir terminalde trafiği izleyin ve bir komut gönderin:

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'

Kod Nasıl Çalışır?

Bağlantı yönetimi

autopaho.NewConnection, bağlanan ve bağlantı her koptuğunda geri çekilme ile yeniden bağlanmayı sürdüren bir arka plan goroutine'i başlatır. AwaitConnection, ilk bağlantı başarılı olana veya context sona erene kadar bekler. Yazmanız gereken yeniden bağlanma mantığı bundan ibarettir; manuel döngülere gerek yoktur.

OnConnectionUp içinde yeniden abone olmak

OnConnectionUp yalnızca ilk bağlantıdan sonra değil, her başarılı bağlantıdan sonra çalışır. Abone olmak için doğru yer burasıdır. CleanStartOnInitialConnection: false ve 600 saniyelik SessionExpiryInterval ile broker, bir kopuştan sonra oturumu (abonelikleri ve kuyruktaki QoS 1 mesajlarını) on dakika boyunca saklar; böylece hızlı bir yeniden bağlanma oturumu kaldığı yerden sürdürür ve ek abonelik zararsızdır. Daha uzun bir kesintiden sonra ise abonelik oturumu yeniden oluşturur. Herkese açık broker'da oturum süresi 30 dakika ile sınırlıdır.

Mesaj almak

OnPublishReceived bir handler dilimidir (slice). Her biri mesajı işleyip işlemediğini döndürür; bu da handler'ları zincirlemenize, örneğin her topic ailesi için bir tane kullanmanıza olanak tanır. Handler'lar client'ın alma yolunda çalıştığından onları hızlı tutun ve ağır işleri bir channel'a veya worker havuzuna devredin.

MQTT 5 özellikleri

Telemetri yayını iki MQTT 5 özelliği ayarlar. ContentType, tüketicilere payload'ın JSON olduğunu bildirir. 60 saniyelik MessageExpiry ise broker'a mesaj bir dakika içinde teslim edilemezse onu atmasını söyler; zaten yerini yenisine bırakacak bir sıcaklık değeri için tam olarak istediğiniz budur. MQTT 5 özellikleri rehberi diğer özellikleri açıklar.

Last Will ve durum

Will mesajı, servis düzgün bir bağlantı kesme olmadan ortadan kaybolursa durum topic'ine retained offline yayınlar. Servis her bağlantıda retained online yayınlar. Herhangi bir dashboard durum topic'ine abone olarak servisin çalışıp çalışmadığını her zaman bilebilir.

Özel Bir Broker'a Geçiş

Üretim için ConnectMQ portalında bir broker oluşturun ve üretilen kimlik bilgilerini kullanın. Client ID'ler verilen kimlik bilgilerine bağlıdır, bu yüzden tam ID'yi aktarın:

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 .

Broker'ın sertifikası, Go'nun sistem kök sertifika deposu aracılığıyla güvendiği Let's Encrypt tarafından verilir; bu yüzden TlsCfg için yalnızca minimum sürüm yeterlidir. scratch veya distroless/static gibi minimal bir container imajında CA sertifikalarının bulunduğundan emin olun; aksi halde el sıkışma x509: certificate signed by unknown authority hatasıyla başarısız olur.

Go MQTT Client için Üretim Kontrol Listesi

1. Her süreç için bir client ID. Aynı ID'ye sahip iki replika birbirinin bağlantısını sürekli koparır. Her replikaya kendi kimlik bilgilerini verin veya destekleyen planlarda yük dengeli tüketiciler için MQTT 5 shared subscription kullanın.
2. Önemsediğiniz veriler için QoS 1 kullanın. Free plan QoS 0 ve 1'i destekler; QoS 2 ücretli planlarda mevcuttur. QoS seviyeleri rehberi ödünleşimleri açıklar.
3. Yayın hatalarını ele alın. Bağlantı kopukken Publish hemen bir hata döndürür. Atmaya, tamponlamaya veya yeniden denemeye karar verin. Güncel autopaho sürümleri kesintiler sırasında tamponlama için kuyruk destekli bir publish da sunar.
4. Düzgün kapatın. Disconnect çağrısı normal bir DISCONNECT gönderir; böylece Will tetiklenmez ve aboneler bunun yerine sizin açık durum mesajınızı görür.
5. Reason code'ları loglayın. MQTT 5 broker'ları bağlantı kesmelerini reason code'larla açıklar; örneğin 0x97 bir kotanın aşıldığı anlamına gelir. OnServerDisconnect size bu kodu verir.
6. Gizli bilgileri kodun dışında tutun. Kimlik bilgilerini kaynak ağacından değil, ortam değişkenlerinden veya bir secret manager'dan okuyun.

Sorun Giderme

Paho kütüphane referansı için paho.golang dokümantasyonuna bakın.

Diğer Backend Dilleri

Aynı desenler, yani kararlı bir client ID, bağlantıda yeniden abonelik, Last Will ve TLS, her yerde geçerlidir. Yığınınız karmaşıksa Rust rumqttc eğitimi ve Java MQTT client eğitimi ile karşılaştırın.

Go Servislerinizi ConnectMQ Üzerinde Çalıştırın

Bir yönetilen MQTT broker, broker'ı, sertifikaları ve güncellemeleri başkası yönetirken Go servislerinizin iş mantığına odaklanmasını sağlar. ConnectMQ oturum süresi, mesaj süresi ve reason code'larıyla MQTT 5'i destekler; bu yüzden bu eğitimdeki her şey değiştirmeden çalışır. app.connectmq.com adresinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun ve MQTT_URL değerini ona yönlendirin.

Sık Sorulan Sorular

Hangi Go MQTT kütüphanesini kullanmalıyım?

Yeni projeler için, MQTT 5'i desteklediği ve yeniden bağlanmaları yönettiği için autopaho paketiyle birlikte github.com/eclipse/paho.golang. Mevcut kodla MQTT 3.1.1 uyumluluğuna ihtiyacınız varsa paho.mqtt.golang kullanın.

autopaho yeniden bağlandıktan sonra otomatik olarak yeniden abone olur mu?

Her bağlantıdan sonra OnConnectionUp çağrılır. Orada abone olun ve bunu, broker'ın kısa kesintiler boyunca aboneliği de saklayabilmesi için bir oturum süresi (session expiry) ile birleştirin.

Bir Go MQTT client'ı TLS üzerinden nasıl bağlarım?

8883 portuyla bir tls:// sunucu URL'si kullanın ve TlsCfg ayarlayın. Herkesçe güvenilen sertifikalara sahip broker'lar için sistem kök deposu yeterlidir.

Aynı Go servisinin birden fazla örneğini çalıştırabilir miyim?

Evet, ancak her örneğin kendi client ID'si olmalıdır. Aynı topic'lerin yük dengeli tüketimi için, destekleyen bir planda MQTT 5 shared subscription kullanın.