C ve Eclipse Paho C ile MQTT: senkron ve asenkron örnekler

Eclipse Paho C, Linux, macOS ve Windows'ta C için başvuru niteliğindeki MQTT istemci kütüphanesidir. İki API sunar: her çağrının bitene kadar beklediği senkron MQTTClient ve yerleşik otomatik yeniden bağlanmaya sahip, callback tabanlı, bloklamayan MQTTAsync. İkisi de MQTT 3.1.1 ve MQTT 5, OpenSSL ile TLS ve WebSocket destekler. Bu rehberde kütüphaneyi kuruyor, senkron bir publisher ile asenkron bir subscriber yazıyor, ardından TLS ve MQTT 5 ekliyoruz.

İşletim sistemi olmayan mikrodenetleyiciler için Paho C genellikle fazla ağırdır; bunun yerine FreeRTOS üzerinde coreMQTT yazısına bakın.

Kütüphaneyi kurun ya da derleyin

En hızlı yol paketlerdir:

Paketlerden kurulum

# Debian / Ubuntu
sudo apt install libpaho-mqtt-dev

# macOS (Homebrew)
brew install libpaho-mqtt

En son sürüm için ya da derleme seçeneklerini kontrol etmek için CMake ile derleyin:

TLS ile kaynak koddan derleme

git clone https://github.com/eclipse/paho.mqtt.c.git
cd paho.mqtt.c
cmake -Bbuild -H. -DPAHO_WITH_SSL=ON -DPAHO_ENABLE_TESTING=OFF
sudo cmake --build build/ --target install
sudo ldconfig   # yalnızca Linux

PAHO_WITH_SSL=ON için bir C derleyicisi, CMake ve OpenSSL geliştirme paketi (libssl-dev) gerekir. Derleme dört kütüphane üretir; programınıza uyanı seçin:

Kütüphane	API	TLS
`paho-mqtt3c`	`MQTTClient` (senkron)	Hayır
`paho-mqtt3cs`	`MQTTClient` (senkron)	Evet
`paho-mqtt3a`	`MQTTAsync`	Hayır
`paho-mqtt3as`	`MQTTAsync`	Evet

Senkron bir publisher

Senkron API yukarıdan aşağıya okunur: oluştur, bağlan, yayınla, teslimi bekle, bağlantıyı kes.

pub.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"

#define ADDRESS  "tcp://broker.connectmq.com:1883"
#define CLIENTID "testmqtt-c-sync-pub"
#define TOPIC    "testmqtt/c-paho/hello"
#define TIMEOUT  10000L

int main(void)
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message msg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    rc = MQTTClient_create(&client, ADDRESS, CLIENTID,
                           MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "create başarısız: %d\n", rc);
        return EXIT_FAILURE;
    }

    conn_opts.keepAliveInterval = 30;
    conn_opts.cleansession = 1;
    conn_opts.username = getenv("MQTT_USER");   /* açık broker'lar için NULL sorun değil */
    conn_opts.password = getenv("MQTT_PASS");

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "bağlantı başarısız: %d (%s)\n", rc, MQTTClient_strerror(rc));
        MQTTClient_destroy(&client);
        return EXIT_FAILURE;
    }

    const char *payload = "hello from Paho C";
    msg.payload = (void *)payload;
    msg.payloadlen = (int)strlen(payload);
    msg.qos = 1;
    msg.retained = 0;

    MQTTClient_publishMessage(client, TOPIC, &msg, &token);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("teslim %s (token %d)\n", rc == MQTTCLIENT_SUCCESS ? "onaylandı" : "başarısız", token);

    MQTTClient_disconnect(client, 1000);
    MQTTClient_destroy(&client);
    return rc == MQTTCLIENT_SUCCESS ? EXIT_SUCCESS : EXIT_FAILURE;
}

Derleyin ve çalıştırın

gcc pub.c -o pub -lpaho-mqtt3c
./pub

Çalıştırmadan önce online MQTT istemcisini açın ve testmqtt/c-paho/# topic'ine abone olun; pub "teslim onaylandı" yazar yazmaz mesaj orada görünür. QoS 1 ile MQTTClient_waitForCompletion ancak broker'ın PUBACK'inden sonra döner; yani başarı, mesajın gerçekten broker'a ulaştığı anlamına gelir.

Otomatik yeniden bağlanan asenkron bir subscriber

Uzun süre çalışan bir süreç için MQTTAsync daha uygundur. Hiçbir şey bloklamaz; callback'leri kaydedersiniz, kütüphanenin arka plan thread'i onları çağırır. Önemli olan connected callback'idir: otomatik yeniden bağlanmalar dahil her başarılı bağlantıdan sonra çalışır; dolayısıyla abone olmak için doğru yer burasıdır.

sub_async.c

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "MQTTAsync.h"

#define ADDRESS  "tcp://broker.connectmq.com:1883"
#define CLIENTID "testmqtt-c-async-sub"
#define TOPIC    "testmqtt/c-paho/#"

static void on_subscribe(void *ctx, MQTTAsync_successData *resp)
{
    printf("%s topic'ine abone olundu\n", TOPIC);
}

static void on_subscribe_failure(void *ctx, MQTTAsync_failureData *resp)
{
    printf("subscribe başarısız: %d\n", resp ? resp->code : 0);
}

/* İlk bağlantıdan ve her otomatik yeniden bağlanmadan sonra çalışır. */
static void on_connected(void *ctx, char *cause)
{
    MQTTAsync client = (MQTTAsync)ctx;
    MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
    opts.onSuccess = on_subscribe;
    opts.onFailure = on_subscribe_failure;
    opts.context = client;
    printf("bağlandı (%s)\n", cause ? cause : "");
    MQTTAsync_subscribe(client, TOPIC, 1, &opts);
}

static void on_connect_failure(void *ctx, MQTTAsync_failureData *resp)
{
    printf("bağlantı başarısız: %d %s\n", resp ? resp->code : 0,
           resp && resp->message ? resp->message : "");
}

static void on_connection_lost(void *ctx, char *cause)
{
    printf("bağlantı koptu: %s\n", cause ? cause : "bilinmiyor");
}

static int on_message(void *ctx, char *topic, int topic_len, MQTTAsync_message *m)
{
    printf("alındı %s: %.*s\n", topic, m->payloadlen, (char *)m->payload);
    MQTTAsync_freeMessage(&m);
    MQTTAsync_free(topic);
    return 1;   /* 1 = mesaj işlendi */
}

int main(void)
{
    MQTTAsync client;
    MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
    MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
    int rc;

    if ((rc = MQTTAsync_create(&client, ADDRESS, CLIENTID,
                               MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS) {
        fprintf(stderr, "create başarısız: %d\n", rc);
        return EXIT_FAILURE;
    }
    MQTTAsync_setCallbacks(client, client, on_connection_lost, on_message, NULL);
    MQTTAsync_setConnected(client, client, on_connected);

    conn_opts.keepAliveInterval = 30;
    conn_opts.cleansession = 1;
    conn_opts.automaticReconnect = 1;
    conn_opts.minRetryInterval = 1;    /* saniye, en fazla maksimuma kadar ikiye katlanır */
    conn_opts.maxRetryInterval = 30;
    conn_opts.onFailure = on_connect_failure;
    conn_opts.context = client;
    conn_opts.username = getenv("MQTT_USER");
    conn_opts.password = getenv("MQTT_PASS");

    if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS) {
        fprintf(stderr, "bağlantı başlatılamadı: %d\n", rc);
        return EXIT_FAILURE;
    }

    printf("çıkmak için Enter'a basın\n");
    getchar();

    MQTTAsync_disconnect(client, &disc_opts);
    sleep(1);
    MQTTAsync_destroy(&client);
    return EXIT_SUCCESS;
}

Derleyin ve çalıştırın

gcc sub_async.c -o sub_async -lpaho-mqtt3a
./sub_async

İkinci bir terminalde ./pub çalıştırın ya da tarayıcıdaki istemciden yayın yapın; subscriber her mesajı yazdırır. Üç kural MQTTAsync kodunu sağlıklı tutar:

Kütüphanenin verdiğini serbest bırakın. Mesaj callback'i MQTTAsync_freeMessage ve MQTTAsync_free(topic) çağırmalıdır; aksi hâlde süreç her mesajda bellek sızdırır.
Callback'lerden hızlı dönün. Callback'ler kütüphanenin thread'inde çalışır; orada uzun süren işler keep-alive yönetimini geciktirir. Mesajları kendi kuyruğunuza ya da worker'ınıza aktarın.
Connected callback'inde abone olun. Clean session ile broker, bağlantı koptuğunda abonelikleri unutur; orada yeniden abone olmak, her otomatik yeniden bağlanmadan sonra abonelikleri geri getirir.

CMake ile derleme

Kurulan kütüphane bir CMake paketiyle gelir; büyük projelerde bağlayıcı bayraklarını elle yazmanız gerekmez:

CMakeLists.txt

cmake_minimum_required(VERSION 3.16)
project(mqtt_c_demo C)

find_package(eclipse-paho-mqtt-c REQUIRED)

add_executable(pub pub.c)
target_link_libraries(pub eclipse-paho-mqtt-c::paho-mqtt3c)

add_executable(sub_async sub_async.c)
target_link_libraries(sub_async eclipse-paho-mqtt-c::paho-mqtt3a)

TLS eklediğinizde paho-mqtt3cs / paho-mqtt3as hedeflerini kullanın.

TLS ve kimlik bilgileri

URI şemasını ssl://, portu 8883 yapın, SSL seçeneklerini doldurun ve kütüphanenin TLS sürümünü bağlayın:

TLS seçenekleri (senkron API)

#define ADDRESS "ssl://broker.connectmq.com:8883"

MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
ssl_opts.enableServerCertAuth = 1;
ssl_opts.verify = 1;                 /* sertifikadaki host adını kontrol et */
/* ssl_opts.trustStore = "ca.pem";      yalnızca özel CA için */
/* ssl_opts.keyStore = "client.pem";    istemci sertifikası (karşılıklı TLS) */
conn_opts.ssl = &ssl_opts;

TLS kütüphanesini bağlayın

gcc pub_tls.c -o pub_tls -lpaho-mqtt3cs

trustStore verilmezse OpenSSL'in varsayılan sertifika konumları kullanılır; bu, herkesçe güvenilen sertifikalara sahip broker'larda çalışır. Asenkron API'de aynı alanlar MQTTAsync_SSLOptions içinde bulunur. Handshake başarısız olursa önce MQTT TLS bağlantısını test etme adımlarıyla sertifika zincirini eleyin.

Paho C ile MQTT 5

MQTT 5 için bir oluşturma seçeneği, bağlantı seçeneklerinde v5 başlatıcısı ve reason code döndüren ...5 çağrı sürümleri gerekir:

MQTT 5 bağlantısı (senkron API)

MQTTClient_createOptions create_opts = MQTTClient_createOptions_initializer;
create_opts.MQTTVersion = MQTTVERSION_5;
MQTTClient_createWithOptions(&client, ADDRESS, CLIENTID,
                             MQTTCLIENT_PERSISTENCE_NONE, NULL, &create_opts);

MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer5;
conn_opts.cleanstart = 1;
MQTTProperties props = MQTTProperties_initializer;

MQTTResponse resp = MQTTClient_connect5(client, &conn_opts, &props, NULL);
if (resp.reasonCode != MQTTREASONCODE_SUCCESS)
    fprintf(stderr, "connect: %s\n", MQTTReasonCode_toString(resp.reasonCode));
MQTTResponse_free(resp);

Yayınlama ve abonelik de MQTTClient_publishMessage5 ve MQTTClient_subscribe5 ile aynı şekilde çalışır. Herhangi bir reason code'u MQTT reason code başvuru aracında arayabilirsiniz.

Göreceğiniz dönüş kodları

Kod	Anlamı	Tipik çözüm
`-1` (`MQTTCLIENT_FAILURE`)	Genel hata; çoğu zaman erişilemeyen host ya da başarısız TLS handshake'i	Host, port, firewall ve sertifikaları kontrol edin
`-10` (`MQTTCLIENT_SSL_NOT_SUPPORTED`)	TLS içermeyen kütüphaneyle `ssl://` URI'si	`paho-mqtt3cs` / `paho-mqtt3as` bağlayın
`-14` (`MQTTCLIENT_BAD_PROTOCOL`)	Bilinmeyen URI şeması	`tcp://`, `ssl://`, `ws://` veya `wss://` kullanın
`4`	CONNACK: hatalı kullanıcı adı veya parola	Kimlik bilgilerini kontrol edin
`5`	CONNACK: yetkisiz	Broker'ın bu istemci için ACL'lerini kontrol edin

1–5 arası pozitif kodlar broker'ın MQTT 3.1.1 CONNACK dönüş kodlarıdır ve CONNACK reason code'ları yazısında ayrıntılı olarak açıklanır. Logların okunabilir olması için bunları MQTTClient_strerror ile yazdırın.

İstemcinizi test etmek

Önce sub_async, sonra pub çalıştırın ve gidiş-dönüşü doğrulayın; ardından bir tarafta online istemciyle tekrarlayın.
Ağı keep-alive süresinden uzun kesip yeniden bağlayın: önce "bağlantı koptu", sonra yeniden "bağlandı" ve "abone olundu" satırlarını görmelisiniz; otomatik yeniden bağlanma iş başında. Tüm cihaz filoları için backoff konusunu yeniden bağlanma en iyi uygulamaları ele alıyor.
MQTT_PASS'a yanlış bir değer verin ve hata callback'inin bunu bildirdiğini doğrulayın.
Mesaj callback'inin her şeyi serbest bıraktığını kanıtlamak için birkaç bin mesaj boyunca Valgrind ya da AddressSanitizer altında çalıştırın.

Demo topic'ine tarayıcıdan abone olun

Paho C küçük, taşınabilir ve yıllardır üretimde kullanılıyor. Araçlar ve testler için MQTTClient ile başlayın, servislerde otomatik yeniden bağlanan MQTTAsync'e geçin, TLS eklediğinizde s kütüphanelerini bağlayın ve aldığınız her mesajı serbest bırakın. Üstünde C++ veya Rust gerekiyorsa, C++ ile MQTT yazısındaki Paho C++ ve Rust için paho-mqtt bu kütüphaneyi sarar.

Sıkça sorulan sorular

Paho C'de MQTTClient ile MQTTAsync arasındaki fark nedir?

MQTTClient senkron API'dir: MQTTClient_connect gibi çağrılar tamamlanana kadar bekler; bu da basit programları kısa tutar. MQTTAsync asla bloklamaz; her işlem başarı ve hata callback'leri alır ve otomatik yeniden bağlanma ile çevrimdışı tamponlama gibi özellikler ekler. Uzun süre çalışan servisler için MQTTAsync kullanın.

Hangi Paho C kütüphanesini bağlamalıyım?

TLS'siz senkron API için paho-mqtt3c, TLS'li senkron için paho-mqtt3cs, TLS'siz asenkron için paho-mqtt3a ve TLS'li asenkron için paho-mqtt3as. TLS içermeyen bir kütüphaneyi bağlayıp ssl:// URI'sine bağlanmaya çalışmak -10 dönüş koduyla (SSL desteklenmiyor) başarısız olur.

MQTTClient_connect'in döndürdüğü 5 kodu ne anlama gelir?

1 ile 5 arasındaki pozitif kodlar, broker'ın gönderdiği MQTT 3.1.1 CONNACK dönüş kodlarıdır. 5 yetkisiz, 4 ise hatalı kullanıcı adı veya parola demektir. Negatif kodlar kütüphanenin kendisinden gelir; örneğin -1, erişilemeyen bir host gibi genel bir hatayı gösterir.

C ve Eclipse Paho C ile MQTT: senkron ve asenkron istemci örnekleri