Kotlin ile Android'de MQTT

Kotlin ile bir Android uygulamasında MQTT kullanmak için bakımı süren bir Java MQTT istemcisi ekleyin (bugün en uygunu HiveMQ MQTT Client), bağlantı durumunu ve mesajları Kotlin flow'ları olarak sunan bir repository içine sarın ve bağlantıyı sonsuza kadar canlı tutmaya çalışmak yerine uygulamanızın yaşam döngüsüne bağlayın. Bu eğitim tam olarak bunu kurar, Android'in arka planda uzun ömürlü soketlere ne yaptığını açıklar ve sonucu gerçek bir broker'a karşı nasıl test edeceğinizi gösterir.

İstemci kütüphanesi seçimi

Aşağıdaki örnekler HiveMQ istemcisini kullanır. Aynı mimari Paho Java ile de çalışır; yalnızca bağlanma ve subscribe çağrıları farklıdır. İstemciyi JVM'den tanıyorsanız HiveMQ MQTT Client ile Java'da MQTT yazısı API'yi ayrıntılı anlatır.

Proje kurulumu

android {
    defaultConfig {
        minSdk = 24
    }
}

dependencies {
    implementation("com.hivemq:hivemq-mqtt-client:1.3.3")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1")
    implementation("androidx.lifecycle:lifecycle-process:2.8.7")
}

<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />

İstemci, API 24'ten itibaren mevcut olan CompletableFuture gibi Java 8 API'lerine dayanır. Release build sırasında R8, istemcinin Netty bağımlılığından eksik sınıflar bildirirse istemcinin Android dokümantasyonundaki keep kurallarını ekleyin.

Coroutine ve flow'larla bir repository

Uygulama başına tam olarak bir istemci tutun ve onu uzun ömürlü bir nesneye sahiplendirin; her döndürmede yeniden oluşturulan bir Activity'ye değil. Aşağıdaki repository bağlantı durumunu StateFlow, gelen mesajları ise SharedFlow olarak sunar:

import com.hivemq.client.mqtt.MqttClient
import com.hivemq.client.mqtt.datatypes.MqttQos
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.future.await

data class MqttMessage(val topic: String, val payload: String)

sealed interface ConnState {
    data object Disconnected : ConnState
    data object Connected : ConnState
    data class Failed(val reason: String) : ConnState
}

class MqttRepository(clientId: String) {

    private val _state = MutableStateFlow<ConnState>(ConnState.Disconnected)
    val state: StateFlow<ConnState> = _state

    private val _messages = MutableSharedFlow<MqttMessage>(extraBufferCapacity = 64)
    val messages: SharedFlow<MqttMessage> = _messages

    private val client: Mqtt5AsyncClient = MqttClient.builder()
        .useMqttVersion5()
        .identifier(clientId)
        .serverHost("broker.connectmq.com")
        .serverPort(8883)
        .sslWithDefaultConfig()
        .automaticReconnectWithDefaultConfig()
        .addConnectedListener { _state.value = ConnState.Connected }
        .addDisconnectedListener { _state.value = ConnState.Failed(it.cause.message ?: "disconnected") }
        .buildAsync()

    suspend fun connect(user: String, pass: String) {
        client.connectWith()
            .simpleAuth().username(user).password(pass.toByteArray()).applySimpleAuth()
            .keepAlive(60)
            .send()
            .await()
    }

    suspend fun subscribe(filter: String) {
        client.subscribeWith()
            .topicFilter(filter)
            .qos(MqttQos.AT_LEAST_ONCE)
            .callback { p ->
                _messages.tryEmit(MqttMessage(p.topic.toString(), String(p.payloadAsBytes)))
            }
            .send()
            .await()
    }

    suspend fun publish(topic: String, payload: String) {
        client.publishWith()
            .topic(topic)
            .qos(MqttQos.AT_LEAST_ONCE)
            .payload(payload.toByteArray())
            .send()
            .await()
    }

    suspend fun disconnect() {
        client.disconnect().await()
        _state.value = ConnState.Disconnected
    }
}

await(), coroutine çekirdek kütüphanesinin parçası olan kotlinx.coroutines.future paketinden gelir ve her CompletableFuture'ı suspend eden bir çağrıya dönüştürür. Abonelik callback'i istemcinin kendi thread'inde çalışır; tryEmit mesajı o thread'i bloklamadan devreder. Bir ViewModel de viewModelScope içinde repository.messages ve repository.state'i collect eder.

Bağlantıyı uygulama yaşam döngüsüne bağlayın

Çoğu uygulama için en basit güvenilir desen: uygulama ön plana geldiğinde bağlanmak, arka plana geçtiğinde bağlantıyı kesmek. ProcessLifecycleOwner bunu Activity bazında değil tüm süreç için bildirir; bu yüzden ekran döndürme ve ekran geçişleri yeniden bağlanmaya yol açmaz:

import androidx.lifecycle.DefaultLifecycleObserver
import androidx.lifecycle.LifecycleOwner
import androidx.lifecycle.ProcessLifecycleOwner
import androidx.lifecycle.lifecycleScope
import kotlinx.coroutines.launch

class App : android.app.Application() {

    lateinit var mqtt: MqttRepository

    override fun onCreate() {
        super.onCreate()
        mqtt = MqttRepository(clientId = InstallId.get(this))

        ProcessLifecycleOwner.get().lifecycle.addObserver(object : DefaultLifecycleObserver {
            override fun onStart(owner: LifecycleOwner) {
                owner.lifecycleScope.launch {
                    runCatching {
                        mqtt.connect(BuildConfig.MQTT_USER, BuildConfig.MQTT_PASS)
                        mqtt.subscribe("testmqtt/android/demo/#")
                    }
                }
            }

            override fun onStop(owner: LifecycleOwner) {
                owner.lifecycleScope.launch { runCatching { mqtt.disconnect() } }
            }
        })
    }
}

InstallId, kurulum başına saklanan UUID'yi döndüren kendi yardımcı sınıfınızı temsil eder. Production uygulamasında gerçek broker kimlik bilgilerini BuildConfig içinde dağıtmayın; her kullanıcıya veya cihaza backend'inizden kendi kimlik bilgilerini verin. Hızlı testler için placeholder'lar yeterlidir.

Android arka planda ne yapar

Arka plandaki bir uygulamanın TCP bağlantısı uzun süre yaşamaz ve bu tasarım gereğidir:

• Doze ve App Standby, boştaki cihazlarda ve az kullanılan uygulamalarda ağ erişimini askıya alır. Keep alive PINGREQ'leriniz durur ve broker, keep alive aralığının 1,5 katından sonra bağlantıyı kapatır.
• Önbellekteki süreçler, bellek açmak için her an dondurulabilir veya sonlandırılabilir; soket de onlarla birlikte ölür.
• Üreticilerin pil tasarrufu modları birçok cihazda standart Android'den daha katıdır ve arka plan işlerini doğrudan sonlandırabilir.
• Foreground service'ler süreci çalışır durumda tutar, ancak görünür bir bildirim ve Android 14+ üzerinde bildirilmiş bir servis türü gerektirir; Android 15 de dataSync foreground service'lerin ne kadar süre çalışabileceğini sınırlar.

Pratik sonuçlar: Uygulama kapalıyken kullanıcılara gerçek zamanlı MQTT teslimatı vaat etmeyin. Önemli bir şeyin kapalı bir uygulamaya ulaşması gerekiyorsa (bir alarm, bir sohbet mesajı) backend'inizden Firebase Cloud Messaging ile push bildirimi gönderin ve uygulamayı onun açmasına izin verin; uygulama sonra yeniden bağlanıp senkronize olur. Foreground service'i yalnızca kullanıcı süregelen bir şey yaparken kullanın; örneğin bir yolculuk sırasında canlı konum takibi. Android'de arka plan bağlantı kopmaları her durumun teşhisine daha ayrıntılı girer.

Kalıcı oturumlar: geri dönünce kaçırılanları almak

Uygulama ön plana döndüğünde, broker oturumunu sakladıysa kaçırdığı QoS 1 mesajlarını alabilir. connectWith() içinde cleanStart(false) ve bir sessionExpiryInterval ayarlayın ve client ID'yi kararlı tutun. Broker böylece uygulama yokken aboneliklere gelen mesajları süre dolana kadar kuyruğa alır. Kaçırılan verinin ne kadar süre yararlı kaldığına uygun bir aralık seçin; bir günlük sensör okumasının bir anda gelmesi nadiren arayüzün istediği şeydir. Paylaşımlı test endpoint'i çevrimdışı oturumları saklamaz; bunu kendi özel broker instance'ınızla deneyin. Kesin kurallar için MQTT kalıcı oturumlar yazısını okuyun.

Ağ değişimleri ve keep alive

Telefonlar sürekli Wi-Fi ile mobil veri arasında geçiş yapar ve her geçiş mevcut TCP bağlantılarını sessizce öldürür. İstemcinin otomatik yeniden bağlanması bunu halleder, ancak tespit ancak keep alive'ınız kadar hızlıdır: 60 saniyede ölü bir bağlantı 90 saniyeye kadar fark edilmeyebilir. Daha hızlı tepki vermek için bir ConnectivityManager.NetworkCallback kaydedin ve onAvailable içinden yeniden bağlanmayı tetikleyin. Mobilde aralığı 30 ile 120 saniye arasında tutun; daha kısası pil harcar, daha uzunu tespiti geciktirir. Zamanlamayı MQTT keep alive yazısı açıklar.

Bir cihazda test edin

1. Uygulamayı bir cihazda veya emülatörde çalıştırın ve state'in Connected olmasını izleyin.
2. Bilgisayarınızda online MQTT istemcisini açın, aynı broker'a bağlanın ve testmqtt/android/demo/hello topic'ine publish edin. Mesaj uygulamada görünmelidir.
3. Diğer yönü doğrulamak için uygulamadan publish edin ve tarayıcıda testmqtt/android/# topic'ine subscribe olun.
4. Uygulamayı arka plana alın, keep alive zaman aşımından daha uzun bekleyin ve geri getirin. Yeni bir Connected durumu ve kalıcı oturumla birlikte arada publish edilen mesajları görmelisiniz.
5. Uçak modunu birkaç saniye açın ve istemcinin kendi kendine yeniden bağlandığını doğrulayın.

Android uygulamanızın mesajlarını izleyin

Emülatörden kendi bilgisayarınızdaki bir broker'a mı bağlanıyorsunuz? localhost yerine 10.0.2.2 kullanın. Cihaz tarafında daha fazla kontrol için Android'den MQTT broker test etmek yazısına bakın.

Android'de MQTT, platformun kurallarını kabul ettiğinizde en iyi şekilde çalışır: uzun ömürlü bir repository'de tek istemci, uygulama yaşam döngüsünü izleyen bir bağlantı, kaçırılanları almak için kalıcı oturumlar, uygulama kapalıyken acil şeyler için push bildirimleri ve her yerde TLS. Bunu bir kez kurun; uygulamanızın her ekranı yalnızca bir flow'u collect etsin.

Sıkça sorulan sorular