Go ve autopaho ile MQTT 5: otomatik yeniden bağlanan istemci rehberi

autopaho, Go için Eclipse Paho MQTT 5 istemcisi olan github.com/eclipse/paho.golang içindeki bağlantı yöneticisidir. Ona bir broker URL listesi ve birkaç callback verirsiniz; o da bağlanır, ağ koptuğunda backoff ile yeniden bağlanır ve her bağlantı (yeniden) kurulduğunda sizi haberdar eder. Bu rehberde autopaho ile eksiksiz bir subscriber ve publisher yazıyor, kimlik bilgileri ile TLS ekliyor ve yeniden bağlanmanın nasıl test edileceğini gösteriyoruz.

Yalnızca MQTT 3.1.1 gerekiyorsa eski paho.mqtt.golang istemcisi hâlâ iş görür. Otomatik yeniden bağlanmanın yanında MQTT 5 özelliklerini — reason code'lar, session expiry, user property'ler, message expiry — istiyorsanız autopaho'yu seçin.

paho ve autopaho

Modülde iki katman vardır ve hangisini çağırdığınızı bilmek işinizi kolaylaştırır:

Pratikte, client ID ve mesaj callback'leri için bir paho.ClientConfig barındıran autopaho.ClientConfig'i yapılandırırsınız; istek paketlerini ise paho paketinden oluşturursunuz.

Kurulum

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

İki paket de aynı modülden geldiği için tek bir go get yeterlidir. Güncel bir Go sürümü kullanın; kütüphane güncel Go sürümlerini takip eder.

Eksiksiz bir istemci

Bu program bir topic'e abone olur, her saniye bir mesaj yayınlar ve Ctrl+C ile düzgünce kapanır. Herkese açık test uç noktası broker.connectmq.com üzerinde çalışır; hazır olduğunuzda host'u kendi broker'ınızla değiştirin.

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("bağlandı, abone olunuyor")
			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
				Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}},
			}); err != nil {
				log.Printf("subscribe başarısız: %v", err)
			}
		},
		OnConnectError: func(err error) { log.Printf("bağlantı hatası: %v", err) },
		ClientConfig: paho.ClientConfig{
			ClientID: "testmqtt-autopaho-demo",
			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
				func(pr paho.PublishReceived) (bool, error) {
					log.Printf("alındı %s: %s", pr.Packet.Topic, pr.Packet.Payload)
					return true, nil
				},
			},
			OnClientError: func(err error) { log.Printf("istemci hatası: %v", err) },
			OnServerDisconnect: func(d *paho.Disconnect) {
				log.Printf("sunucu bağlantıyı kesti, 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 başarısız: %v", err)
			}
		case <-ctx.Done():
			dctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			_ = cm.Disconnect(dctx)
			<-cm.Done()
			return
		}
	}
}

go run . ile çalıştırın; yayınlanan her mesaj için bir "alındı" satırı görmelisiniz. Aynı trafiği dışarıdan izlemek için online MQTT istemcisini açın, broker.connectmq.com'a bağlanın ve testmqtt/go-autopaho/# topic'ine abone olun.

Bazı ayrıntılar önemlidir:

• NewConnection hemen döner. Bağlantıyı arka planda başlatır; AwaitConnection ilk bağlantı başarılı olana ya da context iptal edilene kadar bekler.
• Subscribe işlemini OnConnectionUp içinde yapın. Bu callback ilk bağlantıda ve her yeniden bağlanmada çalışır; böylece ağ koptuktan sonra abonelikleriniz kendiliğinden geri gelir.
• OnPublishReceived bir slice'tır. Her handler (handled bool, err error) döndürür; birden fazla handler aynı mesajı inceleyebilir, örneğin her topic öneki için bir tane.
• QoS 1 ile Publish, PUBACK'i bekler ve broker'ın reason code'unu içeren bir *paho.PublishResponse döndürür; MQTT 5, tek bir publish'teki "not authorized" gibi sorunları bu şekilde bildirir.

Kimlik bilgileri ve TLS

autopaho'nun güncel sürümleri kimlik bilgilerini doğrudan yapılandırmada alır. TLS için URL şemasını mqtts://, portu da 8883 yapın; Go standart kütüphanesi sertifikayı sistemin kök sertifikalarıyla doğrular.

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")),
	// Yalnızca özel bir CA veya istemci sertifikaları için gerekir:
	// TlsCfg: &tls.Config{RootCAs: pool, Certificates: []tls.Certificate{cert}},
	ClientConfig: paho.ClientConfig{ClientID: "testmqtt-autopaho-tls"},
}

ServerUrls bir listedir: autopaho URL'leri sırayla dener; bu da iki uç nokta arasında basit bir failover sağlar. TCP portları engelliyse ws:// ve wss:// URL'leri de (örneğin wss://broker.connectmq.com:8084/mqtt) kabul edilir. Handshake başarısız olursa TLS bağlantı testi rehberi, sertifika zincirini openssl s_client ile nasıl kontrol edeceğinizi gösterir.

Yeniden abone olmak yerine kalıcı oturumlar

OnConnectionUp içinde abone olmak basit ve sağlamdır. Alternatifi, broker'ın sizi hatırlamasını sağlamaktır:

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

Paylaşılan test uç noktası çevrimdışı oturumları saklamaz; bu bölümü kendi özel broker örneğinizde deneyin.

Sabit bir client ID ile broker, bağlantı koptuktan sonra bir saate kadar abonelikleri ve kuyruktaki QoS 1/2 mesajlarını saklar. OnConnectionUp içindeki ack.SessionPresent alanı oturumun korunup korunmadığını söyler; korunduysa subscribe'ı atlayabilirsiniz. Session expiry bir MQTT 5 özelliğidir; broker'ların bunu nasıl ele aldığı için kalıcı oturumlar yazısına bakın.

Bağlantı kopmasından sağ çıkması gereken giden mesajlar için autopaho ayrıca PublishViaQueue sunar: mesajları bir kuyrukta (varsayılan olarak bellekte ya da dosya tabanlı kuyruk paketiyle diskte) saklar ve bağlantı geri geldiğinde gönderir.

Yeniden bağlanma davranışı

autopaho başarısız bağlantıları denemeler arasında bekleyerek yeniden dener. Güncel sürümler bunu yapılandırmada ReconnectBackoff olarak sunar (eski sürümler sabit bir ConnectRetryDelay kullanıyordu); varsayılan değer çoğu istemci için uygundur. Büyük bir cihaz filosu işletiyorsanız, broker yeniden başladıktan sonra tüm istemcilerin aynı anda bağlanmasını önleyin: yeniden bağlanma için en iyi uygulamalar yazısında anlatıldığı gibi jitter ya da artan bir backoff ekleyin.

Başarısız denemeler OnConnectError içinde görünür. Tipik mesajlar ve anlamları:

Herhangi bir sayısal kodu MQTT reason code başvuru aracında arayabilirsiniz.

Yeniden bağlanmayı test etmek

Yeniden aboneliğin çalışıp çalışmadığını üretimde öğrenmeyi beklemeyin. Hızlı bir elle kontrol:

1. Programı çalıştırın ve mesajların aktığını doğrulayın.
2. Ağı keep-alive süresinden daha uzun bir süre kesin (Wi-Fi'ı kapatın ya da sudo ifconfig en0 down / sudo ip link set eth0 down).
3. OnConnectError'ın başarısız denemeleri logladığını izleyin.
4. Ağı geri açın. OnConnectionUp "bağlandı, abone olunuyor" yazmalı ve "alındı" satırları yeniden gelmeli.

Otomatik testler için publish/receive döngüsünü timeout'lu bir go test içine alın; Go'da MQTT istemcilerini test etmek yazısında hazır bir autopaho helper'ı var.

autopaho istemcinizi tarayıcıda izleyin

autopaho, Go programlarına çok az kodla üretime hazır bir MQTT 5 istemcisi kazandırır: URL'leri ve callback'leri yapılandırın, OnConnectionUp içinde abone olun ve size döndürülen reason code'ları kontrol edin. Bir kez ağ kesintisini test edin, client ID'leri benzersiz tutun; kendi kendine toparlanan bir istemciniz olur.

Sıkça sorulan sorular